diff --git a/api/entrypoints/worker_streams.py b/api/entrypoints/worker_streams.py index a776d12270..18ee38bc88 100644 --- a/api/entrypoints/worker_streams.py +++ b/api/entrypoints/worker_streams.py @@ -25,10 +25,12 @@ from oss.src.core.events.service import EventsService from oss.src.core.secrets.services import VaultService +from oss.src.core.sessions.interactions.service import SessionInteractionsService from oss.src.core.sessions.records.service import RecordsService from oss.src.core.tracing.service import TracingService from oss.src.dbs.postgres.events.dao import EventsDAO from oss.src.dbs.postgres.secrets.dao import SecretsDAO +from oss.src.dbs.postgres.sessions.interactions.dao import SessionInteractionsDAO from oss.src.dbs.postgres.sessions.records.dao import RecordsDAO from oss.src.dbs.postgres.tracing.dao import TracingDAO from oss.src.dbs.postgres.webhooks.dao import WebhooksDAO @@ -79,6 +81,7 @@ async def _build_spans_worker(redis_client: Redis) -> StreamConsumer: async def _build_records_worker(redis_client: Redis) -> StreamConsumer: + watch_publisher = SessionsWatchPublisher(redis_client=redis_client) return RecordsWorker( service=RecordsService(records_dao=RecordsDAO()), redis_client=redis_client, @@ -86,7 +89,14 @@ async def _build_records_worker(redis_client: Redis) -> StreamConsumer: consumer_group="worker-records", # M3 live relay: post-append change notifications on the durable plane, # reusing this process's durable connection. - watch_publisher=SessionsWatchPublisher(redis_client=redis_client), + watch_publisher=watch_publisher, + # The gate safety net: this loop sees every turn's terminal record, so it is where a + # pending gate that outlived its turn gets cancelled, scoped to that turn's own gates + # so a newer turn's live park is never in range. + interactions_service=SessionInteractionsService( + interactions_dao=SessionInteractionsDAO(), + watch_publisher=watch_publisher, + ), ) diff --git a/api/oss/src/apis/fastapi/sessions/router.py b/api/oss/src/apis/fastapi/sessions/router.py index e4cd7f9e70..23da3df44e 100644 --- a/api/oss/src/apis/fastapi/sessions/router.py +++ b/api/oss/src/apis/fastapi/sessions/router.py @@ -554,6 +554,11 @@ async def watch_session_stream( Auth is the standard middleware (cookie ``sAccessToken``, ApiKey, or Bearer) evaluated once at connect; scope is the credential's project. + Browsers authenticate by cookie — ``EventSource`` cannot set headers — + so a connect landing on an expired access token 401s like any other + request. There is no interceptor to refresh-and-retry a stream, so the + client must refresh the session itself and reopen (see the web hooks). + The stream has no replay/cursor semantics — ``EventSource`` reconnects and clients revalidate once on every ``open``, which covers any missed notifications. @@ -581,6 +586,7 @@ async def watch_session_stream( # teardown story; revisit with a shared listener if counts grow). pubsub_factory=lambda: get_streams_engine().get_redis().pubsub(), heartbeat_seconds=env.sessions.watch_heartbeat_seconds, + retry_milliseconds=env.sessions.watch_retry_milliseconds, ) return StreamingResponse( stream, diff --git a/api/oss/src/apis/fastapi/sessions/watch.py b/api/oss/src/apis/fastapi/sessions/watch.py index 0a9e056e63..071c71c5fa 100644 --- a/api/oss/src/apis/fastapi/sessions/watch.py +++ b/api/oss/src/apis/fastapi/sessions/watch.py @@ -13,6 +13,7 @@ from oss.src.dbs.redis.sessions.contract import ( WATCH_EVENT_INTERACTION, WATCH_EVENT_LIFECYCLE, + WATCH_EVENT_READY, WATCH_EVENT_RECORDS_CHANGED, ) from oss.src.utils.logging import get_module_logger @@ -21,6 +22,17 @@ HEARTBEAT_FRAME = ": heartbeat\n\n" + +def retry_frame(retry_milliseconds: int) -> str: + """SSE `retry:` field — sets the client's built-in auto-reconnect delay.""" + return f"retry: {retry_milliseconds}\n\n" + + +def ready_frame() -> str: + """Emitted once the Redis subscription is live: the client's cue to revalidate.""" + return "event: " + WATCH_EVENT_READY + "\ndata: {}\n\n" + + _KNOWN_EVENTS = { WATCH_EVENT_RECORDS_CHANGED, WATCH_EVENT_LIFECYCLE, @@ -50,9 +62,22 @@ async def watch_event_stream( channel: str, pubsub_factory: Callable[[], Any], heartbeat_seconds: float, + retry_milliseconds: int, ) -> AsyncIterator[str]: """Subscribe to the session's watch channel and yield SSE frames forever. + The first frame is a ``retry:`` preamble: it pins the client's built-in + auto-reconnect delay (implementation-defined otherwise) so a server-side + drop — an API restart, a deploy — cannot reconnect-storm us. + + The second is a ``ready`` event, and that is what a client revalidates on. + ``onopen`` fires as soon as the response headers arrive, and Starlette flushes + those BEFORE it starts iterating this generator — so a revalidation driven by + ``onopen`` can read the record log, and a change can land and publish, all before + the ``subscribe`` below completes. That change would reach neither the refetch nor + the stream. ``ready`` is emitted once the subscription is live, so a revalidation + keyed on it cannot straddle the gap. + The subscription is torn down in ``finally`` — a client disconnect cancels the generator (GeneratorExit/CancelledError), which is exactly the cleanup path, so no Redis subscription outlives its SSE connection. @@ -60,6 +85,8 @@ async def watch_event_stream( pubsub = pubsub_factory() try: await pubsub.subscribe(channel) + yield retry_frame(retry_milliseconds) + yield ready_frame() while True: message = await pubsub.get_message( ignore_subscribe_messages=True, diff --git a/api/oss/src/core/sessions/interactions/interfaces.py b/api/oss/src/core/sessions/interactions/interfaces.py index f66f6e9bb1..60336b5139 100644 --- a/api/oss/src/core/sessions/interactions/interfaces.py +++ b/api/oss/src/core/sessions/interactions/interfaces.py @@ -46,6 +46,7 @@ async def cancel_session_pending( session_id: str, except_turn_id: Optional[str] = None, except_tokens: Optional[List[str]] = None, + only_turn_id: Optional[str] = None, ) -> int: ... @abstractmethod diff --git a/api/oss/src/core/sessions/interactions/service.py b/api/oss/src/core/sessions/interactions/service.py index 6887b018d3..02d685404f 100644 --- a/api/oss/src/core/sessions/interactions/service.py +++ b/api/oss/src/core/sessions/interactions/service.py @@ -101,12 +101,14 @@ async def cancel_session_pending( session_id: str, except_turn_id: Optional[str] = None, except_tokens: Optional[List[str]] = None, + only_turn_id: Optional[str] = None, ) -> int: cancelled = await self.interactions_dao.cancel_session_pending( project_id=project_id, session_id=session_id, except_turn_id=except_turn_id, except_tokens=except_tokens, + only_turn_id=only_turn_id, ) if cancelled: await self._publish_interaction( diff --git a/api/oss/src/core/sessions/streams/service.py b/api/oss/src/core/sessions/streams/service.py index 41031b1d73..6f8b362711 100644 --- a/api/oss/src/core/sessions/streams/service.py +++ b/api/oss/src/core/sessions/streams/service.py @@ -13,7 +13,7 @@ """ import uuid_utils.compat as uuid -from typing import List, Optional +from typing import Iterable, List, Optional from uuid import UUID from oss.src.utils.logging import get_module_logger @@ -35,10 +35,14 @@ force_cancel_alive, force_clear_owner, get_alive_owner, + get_owner, get_running_owner, get_session_liveness, + is_turn_superseded, + mark_turn_superseded, refresh_alive, refresh_running, + release_alive, release_attached, steal_attached, ) @@ -86,6 +90,57 @@ def __init__( self._lock = lock_engine self._watch = watch_publisher + async def _supersede_turns( + self, + *, + project_id: UUID, + session_id: str, + turn_ids: Iterable[Optional[str]], + ) -> None: + """Tombstone every turn displaced by this edit. `displaced ⇒ dead` is the invariant + that makes the ambiguous "`alive` held by another turn + no `running`" state safe to + resolve as a handover: only a turn that has never been displaced can reach it.""" + for turn_id in {t for t in turn_ids if t}: + await mark_turn_superseded( + self._lock, + project_id=str(project_id), + session_id=session_id, + turn_id=turn_id, + ) + + async def _displace_turns(self, *, project_id: UUID, session_id: str) -> None: + """Tear alive+running off whichever turn holds them, tombstoning it first. + + The order is the point. Clearing first leaves a window in which the turn being + displaced heartbeats, finds `alive` free and nx-acquires it straight back - a + cancelled session then reads as alive for a whole ALIVE_TTL. Tombstoning first makes + that beat refuse itself. The keys are still re-read after the clear, so a turn that + took them inside the window is tombstoned too. + """ + await self._supersede_turns( + project_id=project_id, + session_id=session_id, + turn_ids=( + await get_alive_owner( + self._lock, project_id=str(project_id), session_id=session_id + ), + await get_running_owner( + self._lock, project_id=str(project_id), session_id=session_id + ), + ), + ) + displaced_alive = await force_cancel_alive( + self._lock, project_id=str(project_id), session_id=session_id + ) + displaced_running = await clear_running( + self._lock, project_id=str(project_id), session_id=session_id + ) + await self._supersede_turns( + project_id=project_id, + session_id=session_id, + turn_ids=(displaced_alive, displaced_running), + ) + async def _publish_lifecycle( self, *, project_id: UUID, session_id: str, state: str ) -> None: @@ -138,12 +193,7 @@ async def command( ) elif mode == CommandMode.steer: - await force_cancel_alive( - self._lock, project_id=str(project_id), session_id=session_id - ) - await clear_running( - self._lock, project_id=str(project_id), session_id=session_id - ) + await self._displace_turns(project_id=project_id, session_id=session_id) turn_id = await self._start_turn( project_id=project_id, user_id=user_id, @@ -157,12 +207,7 @@ async def command( ) elif mode == CommandMode.cancel: - await force_cancel_alive( - self._lock, project_id=str(project_id), session_id=session_id - ) - await clear_running( - self._lock, project_id=str(project_id), session_id=session_id - ) + await self._displace_turns(project_id=project_id, session_id=session_id) await self._mark_stream_ended( project_id=project_id, user_id=user_id, @@ -241,12 +286,7 @@ async def kill( whose runner replica is unreachable, is still a no-op success (best-effort teardown). """ _validate_session_id(session_id) - await force_cancel_alive( - self._lock, project_id=str(project_id), session_id=session_id - ) - await clear_running( - self._lock, project_id=str(project_id), session_id=session_id - ) + await self._displace_turns(project_id=project_id, session_id=session_id) # Drop affinity too: claim_owner never steals, so a surviving owner key would lock # the session out of every other replica for the rest of OWNER_TTL_SECONDS. await force_clear_owner( @@ -292,6 +332,39 @@ async def heartbeat( ) -> SessionHeartbeatResult: _validate_session_id(request.session_id) + # A turn that was already displaced (handover, cancel, steer, kill, sweep) is dead + # forever: refuse the beat before it touches ANY lock or the row. This is what keeps + # the ambiguous "`alive` held by another turn + no `running`" state safe to resolve as + # a handover below — the dangerous reading of that state was a zombie beat from an + # older turn taking the nest of a session parked awaiting approval, which then made + # the user's approval resume look superseded and abort. A zombie is by definition a + # turn that already lost the nest, so the tombstone written at the moment it lost it + # is the discriminator the locks alone cannot provide. Refusing early also stops the + # zombie's own turn-end beat from clearing the LIVE turn's `running`. + if request.turn_id and await is_turn_superseded( + self._lock, + project_id=str(project_id), + session_id=request.session_id, + turn_id=request.turn_id, + ): + stream = await self._dao.get_by_session_id( + project_id=project_id, + session_id=request.session_id, + ) + # Read affinity, never claim it: renewing OWNER_TTL on a dead turn's beat pins the + # session to this replica for another full TTL, which is exactly what has to expire + # before another replica can take the session over. + owner = await get_owner( + self._lock, + project_id=str(project_id), + session_id=request.session_id, + ) + return SessionHeartbeatResult( + stream=stream, + replica_id=owner or request.replica_id, + is_current_turn=False, + ) + # replica_id claims affinity without stealing from a live different owner; turn_id # separately refreshes the alive/running TTLs. `owner` is the actual winner (this # replica if it won or already held it, another replica otherwise). @@ -337,10 +410,13 @@ async def heartbeat( # Acquire-then-refresh: the first heartbeat must establish the nest locks # itself (acquire_* is nx=True — a no-op if _start_turn already holds them). # A failed nx acquire is NOT by itself a takeover: nx fails whenever ANY value - # holds the key, and `alive` outlives its turn (release_alive has no callers, and + # holds the key, and `alive` outlives its turn (nothing releases it at turn end, and # the turn-end beat clears only `running`), so every follow-up turn on a warm # session sees the previous turn's key. `running` is the discriminator — a real - # takeover (steer/_start_turn) holds it under the usurper's turn id. + # takeover (steer/_start_turn) holds it under the usurper's turn id — and the + # supersession tombstone checked above is what makes the remaining "no running" + # case safe: any turn that could reach here dishonestly has already been + # tombstoned by whatever displaced it. if not await refresh_alive( self._lock, project_id=str(project_id), @@ -371,18 +447,36 @@ async def heartbeat( pass # a live different turn holds the session: real takeover else: # Stale `alive` from this session's own previous (ended or parked) - # turn — legitimate handover, not an interruption. - await force_cancel_alive( - self._lock, - project_id=str(project_id), - session_id=request.session_id, - ) - acquired = await acquire_alive( + # turn — legitimate handover, not an interruption. The displaced turn + # is tombstoned so it can never beat its way back in: that is the only + # thing standing between this branch and a zombie stealing the nest of + # a parked session. + # + # Compare-and-delete against the owner read just above, never an + # unconditional delete: an API `_start_turn` can land in the gap + # between that read and this write, and clearing the key then would + # tombstone the live turn that had just taken the session. Losing that + # race leaves `acquired` False, which is the truth. + displaced = alive_owner + if displaced is None or await release_alive( self._lock, project_id=str(project_id), session_id=request.session_id, - turn_id=request.turn_id, - ) + turn_id=displaced, + ): + if displaced and displaced != request.turn_id: + await mark_turn_superseded( + self._lock, + project_id=str(project_id), + session_id=request.session_id, + turn_id=displaced, + ) + acquired = await acquire_alive( + self._lock, + project_id=str(project_id), + session_id=request.session_id, + turn_id=request.turn_id, + ) if not acquired or turn_was_established: is_current_turn = False if not await refresh_running( diff --git a/api/oss/src/dbs/postgres/sessions/interactions/dao.py b/api/oss/src/dbs/postgres/sessions/interactions/dao.py index 33f8b0a159..97043a77b4 100644 --- a/api/oss/src/dbs/postgres/sessions/interactions/dao.py +++ b/api/oss/src/dbs/postgres/sessions/interactions/dao.py @@ -141,12 +141,13 @@ async def cancel_session_pending( session_id: str, except_turn_id: Optional[str] = None, except_tokens: Optional[List[str]] = None, + only_turn_id: Optional[str] = None, ) -> int: """Cancel still-pending interactions for a session. With `except_turn_id`, spare the current turn's own gates (used at turn start to cancel prior turns' unanswered gates; without it, cancel all of them, e.g. on kill). `except_tokens` spares prior-turn gates - the current turn answers in-band, so the resume can resolve them instead. Returns the - count cancelled.""" + the current turn answers in-band, so the resume can resolve them instead. With + `only_turn_id`, touch nothing but that one turn's gates. Returns the count cancelled.""" async with self.engine.session() as session: stmt = ( sa_update(SessionInteractionDBE) @@ -160,6 +161,8 @@ async def cancel_session_pending( updated_at=datetime.now(timezone.utc), ) ) + if only_turn_id is not None: + stmt = stmt.where(SessionInteractionDBE.turn_id == only_turn_id) if except_turn_id is not None: stmt = stmt.where(SessionInteractionDBE.turn_id != except_turn_id) if except_tokens: diff --git a/api/oss/src/dbs/redis/sessions/contract.py b/api/oss/src/dbs/redis/sessions/contract.py index f736f423e3..2bee7ee0b9 100644 --- a/api/oss/src/dbs/redis/sessions/contract.py +++ b/api/oss/src/dbs/redis/sessions/contract.py @@ -11,6 +11,10 @@ owner::session: — which replica currently owns this session displaced::session: — pub/sub for attach-steal notifications watch::session: — pub/sub for the live relay (SSE watch) + superseded::session::turn: + — tombstone: this turn lost the nest and is + dead forever (API-side only; the runner + learns it through `is_current_turn`) `session_id` is caller-supplied and Postgres uniqueness is (project_id, session_id), so two projects may legitimately hold the same one. The `project_id` segment is the tenant boundary: @@ -37,6 +41,10 @@ HEARTBEAT_INTERVAL_SECONDS: int = env.sessions.heartbeat_interval_seconds HEARTBEAT_WRITE_THRESHOLD_SECONDS: int = env.sessions.heartbeat_write_threshold_seconds +# API-side only — the runner never reads the tombstone key, so this constant is +# deliberately absent from the shared golden fixture (like `watch_heartbeat_seconds`). +SUPERSEDED_TTL_SECONDS: int = env.sessions.superseded_ttl_seconds + # --------------------------------------------------------------------------- # Key builders # --------------------------------------------------------------------------- @@ -58,6 +66,10 @@ def owner_key(project_id: str, session_id: str) -> str: return f"owner:{project_id}:session:{session_id}" +def superseded_key(project_id: str, session_id: str, turn_id: str) -> str: + return f"superseded:{project_id}:session:{session_id}:turn:{turn_id}" + + def displaced_channel(project_id: str, session_id: str) -> str: return f"displaced:{project_id}:session:{session_id}" @@ -88,6 +100,10 @@ def make_displacement_payload(*, by: str) -> dict: WATCH_EVENT_RECORDS_CHANGED = "records-changed" WATCH_EVENT_LIFECYCLE = "lifecycle" WATCH_EVENT_INTERACTION = "interaction" +# Emitted by the SSE endpoint itself, never published: it marks the point where the Redis +# subscription is live, so a client can revalidate without racing the events it is about to +# start receiving. +WATCH_EVENT_READY = "ready" WATCH_LIFECYCLE_RUNNING = "running" WATCH_LIFECYCLE_ENDED = "ended" diff --git a/api/oss/src/dbs/redis/sessions/locks.py b/api/oss/src/dbs/redis/sessions/locks.py index a979669ef6..8da9dcf991 100644 --- a/api/oss/src/dbs/redis/sessions/locks.py +++ b/api/oss/src/dbs/redis/sessions/locks.py @@ -16,12 +16,14 @@ OWNER_TTL_SECONDS, RELEASE_IF_OWNER_LUA, RUNNING_TTL_SECONDS, + SUPERSEDED_TTL_SECONDS, alive_key, attached_key, displaced_channel, make_displacement_payload, owner_key, running_key, + superseded_key, validate_session_id, # noqa: F401 — re-exported for callers that import from locks ) @@ -111,6 +113,47 @@ async def get_alive_owner( return current.decode() if current else None +# --------------------------------------------------------------------------- +# Turn supersession tombstones — "this turn lost the nest; it is dead forever" +# +# `alive` outlives its turn and a parked turn holds no `running`, so the state +# "`alive` held by another turn + no `running`" cannot, from the locks alone, tell a +# lapsed previous turn (a legitimate handover) from a live-but-parked one. Rather than +# guess, we record the one thing that IS knowable at the moment it happens: a turn that +# was displaced. A displaced turn's later beats are refused outright, so a zombie beat +# can never re-take a nest it already lost — which is what made the ambiguity reachable. +# --------------------------------------------------------------------------- + + +async def mark_turn_superseded( + engine: LockEngine, + *, + project_id: str, + session_id: str, + turn_id: str, +) -> None: + """Tombstone turn_id: it was displaced (handover, cancel, steer, kill, sweep).""" + key = superseded_key(project_id, session_id, turn_id) + await engine.set(key, b"1", ex=SUPERSEDED_TTL_SECONDS) + + +async def is_turn_superseded( + engine: LockEngine, + *, + project_id: str, + session_id: str, + turn_id: str, +) -> bool: + """True if turn_id was displaced. Refreshes the TTL on every hit so a long-lived + zombie that keeps beating stays dead instead of outliving its own tombstone.""" + key = superseded_key(project_id, session_id, turn_id) + current = await engine.get(key) + if current is None: + return False + await engine.expire(key, SUPERSEDED_TTL_SECONDS) + return True + + # --------------------------------------------------------------------------- # Running lock — "a turn is actively executing right now" # Nested under alive: a session can be alive-but-idle (running absent) between turns. @@ -175,7 +218,7 @@ async def clear_running( project_id: str, session_id: str, ) -> Optional[str]: - """Unconditionally clear the running lock (turn ended/cancelled). Returns prior turn.""" + """Unconditionally clear the running lock (displacement/sweep). Returns prior turn.""" key = running_key(project_id, session_id) current = await engine.get(key) await engine.delete(key) diff --git a/api/oss/src/tasks/asyncio/sessions/orphan_sweep.py b/api/oss/src/tasks/asyncio/sessions/orphan_sweep.py index 4acebe7045..40f9773055 100644 --- a/api/oss/src/tasks/asyncio/sessions/orphan_sweep.py +++ b/api/oss/src/tasks/asyncio/sessions/orphan_sweep.py @@ -22,6 +22,7 @@ force_cancel_alive, clear_running, force_clear_owner, + mark_turn_superseded, ) from sqlalchemy import and_, func, not_, or_, select @@ -90,12 +91,21 @@ async def run_orphan_sweep(engine: TransactionsEngine, lock_engine: LockEngine) # Bring the Redis locks the SEND gate reads in sync with the rows just written. for row in orphans: project_id = str(row.project_id) - await force_cancel_alive( + displaced_alive = await force_cancel_alive( lock_engine, project_id=project_id, session_id=row.session_id ) - await clear_running( + displaced_running = await clear_running( lock_engine, project_id=project_id, session_id=row.session_id ) + # A swept turn is declared dead; tombstone it so a late beat from it cannot + # re-nest the session it was just evicted from. + for turn_id in {t for t in (displaced_alive, displaced_running) if t}: + await mark_turn_superseded( + lock_engine, + project_id=project_id, + session_id=row.session_id, + turn_id=turn_id, + ) # A swept session is dead; free its affinity like kill does. await force_clear_owner( lock_engine, project_id=project_id, session_id=row.session_id diff --git a/api/oss/src/tasks/asyncio/sessions/records_worker.py b/api/oss/src/tasks/asyncio/sessions/records_worker.py index 6e9b185d0a..b107935a44 100644 --- a/api/oss/src/tasks/asyncio/sessions/records_worker.py +++ b/api/oss/src/tasks/asyncio/sessions/records_worker.py @@ -3,6 +3,7 @@ from redis.asyncio import Redis +from oss.src.core.sessions.interactions.service import SessionInteractionsService from oss.src.core.sessions.records.service import RecordsService from oss.src.core.sessions.records.streaming import deserialize_record from oss.src.core.sessions.watch.interfaces import SessionsWatchPublisherInterface @@ -17,6 +18,32 @@ from ee.src.core.access.entitlements.types import Counter +# The runner's terminal per-turn record, and the marker it stamps on that record when the turn +# stopped to wait for a human instead of finishing (services/runner/src/tracing/otel.ts: the +# field is written ONLY for a pause and omitted on every other stop reason). +TERMINAL_RECORD_TYPE = "done" +PAUSED_STOP_REASON = "paused" + + +def finished_turns_in_batch(events: List[Any]) -> Dict[str, str]: + """`session_id -> turn_id` for every turn in this batch that FINISHED without pausing. + + A gate row exists only because a turn paused for a human, so a turn whose terminal record + carries no pause marker is holding no gate — and neither is its session, whose live process + is gone. That is the reconciliation trigger. The last finished turn per session wins, and the + caller cancels only that turn's own gates. + """ + finished: Dict[str, str] = {} + for msg in events: + record = msg.record_event + if record.record_type != TERMINAL_RECORD_TYPE or not record.turn_id: + continue + if (record.attributes or {}).get("stopReason") == PAUSED_STOP_REASON: + continue + finished[record.session_id] = record.turn_id + return finished + + class RecordsWorker(StreamConsumer): """ Worker for record ingestion via dedicated Redis stream. @@ -30,7 +57,8 @@ class RecordsWorker(StreamConsumer): 3. Group by project_id 4. EE: L2 quota check per org (Counter.RECORDS_INGESTED) 5. Append record events to DB - 6. ACK + DEL messages — StreamConsumer + 6. Reconcile HITL gates orphaned by a finished turn + 7. ACK + DEL messages — StreamConsumer """ log_prefix = "[RECORDS]" @@ -47,6 +75,7 @@ def __init__( max_delay_ms: int = 250, max_batch_mb: int = 50, watch_publisher: Optional[SessionsWatchPublisherInterface] = None, + interactions_service: Optional[SessionInteractionsService] = None, ): super().__init__( redis_client=redis_client, @@ -60,6 +89,63 @@ def __init__( ) self.service = service self.watch_publisher = watch_publisher + # Absent disables gate reconciliation (minimal test compositions), which only loses the + # safety net — never the append. + self.interactions_service = interactions_service + + async def reconcile_orphaned_gates( + self, + *, + project_id: UUID, + events: List[Any], + ) -> None: + """Safety net: no HITL gate may outlive its turn. + + A `session_interactions` row is created only when a turn pauses for a human. Once a turn + reaches its terminal record WITHOUT pausing, no process is holding a gate for that + session, so any row still `pending` can never be answered — yet both inboxes keep + offering it forever. That is the orphaned gate. Cancel those rows here, at the one place + that sees every turn's terminal record. + + A legitimately parked gate is protected twice: + + * a terminal record carrying the pause marker is skipped outright — that turn IS the + live park, and its gate is exactly what the human is being asked to answer; + * the cancel is scoped to the finished turn's OWN gates. A newer turn carries a + different `turn_id`, so no interleaving can put its live park in range — this worker + may lag arbitrarily far behind the stream and still never cancel underneath a turn + that is parked right now. Prior turns' leftovers are not this sweep's job: the runner + clears them at turn start through `/sessions/interactions/cancel-stale`. + + The cancel fans out on the watch plane (the service publishes on a non-zero count), so a + client sitting on a stuck approval drops it without a reload. Best effort throughout — a + failure here must never re-drive the record append. + """ + if self.interactions_service is None: + return + + for session_id, turn_id in finished_turns_in_batch(events).items(): + try: + cancelled = await self.interactions_service.cancel_session_pending( + project_id=project_id, + session_id=session_id, + only_turn_id=str(turn_id), + ) + if cancelled: + log.info( + "[RECORDS] Cancelled gates orphaned by a finished turn", + project_id=str(project_id), + session_id=session_id, + turn_id=str(turn_id), + cancelled=cancelled, + ) + except Exception: + log.warning( + "[RECORDS] Gate reconciliation failed", + project_id=str(project_id), + session_id=session_id, + exc_info=True, + ) async def process_batch( self, @@ -161,6 +247,13 @@ async def process_batch( ) continue + # Strictly post-append, and BEFORE the relay tee: a client woken by the records + # notification below must already see the cancelled gate, not re-render it. + await self.reconcile_orphaned_gates( + project_id=project_batch["project_id"], + events=project_batch["events"], + ) + # Relay tee (M3): strictly post-append so a notified client that # revalidates always sees the new rows. One publish per distinct # session in the project batch; failures never re-drive the append. diff --git a/api/oss/src/utils/env.py b/api/oss/src/utils/env.py index b9d788f638..7c25d03837 100644 --- a/api/oss/src/utils/env.py +++ b/api/oss/src/utils/env.py @@ -1393,6 +1393,21 @@ class SessionsRedisConfig(BaseModel): _parse_optional_positive_int_env("AGENTA_SESSIONS_WATCH_HEARTBEAT_SECONDS") or 15 ) + # SSE `retry:` preamble — the browser's OWN auto-reconnect delay after a + # server-side drop (restart/deploy). Without it the interval is + # implementation-defined, and a restart reconnect-storms the API. + watch_retry_milliseconds: int = ( + _parse_optional_positive_int_env("AGENTA_SESSIONS_WATCH_RETRY_MILLISECONDS") + or 5000 + ) + # API-side only (turn-supersession tombstones) — NOT part of the runner golden + # fixture; the runner never reads this key, it learns supersession from + # `is_current_turn`. Defaults to the alive TTL so a tombstone always outlives the + # lock whose displacement created it. + superseded_ttl_seconds: int = ( + _parse_optional_positive_int_env("AGENTA_SESSIONS_REDIS_SUPERSEDED_TTL_SECONDS") + or 3600 + ) model_config = ConfigDict(extra="ignore") diff --git a/api/oss/tests/pytest/unit/sessions/test_heartbeat_lock_races.py b/api/oss/tests/pytest/unit/sessions/test_heartbeat_lock_races.py new file mode 100644 index 0000000000..e41784244d --- /dev/null +++ b/api/oss/tests/pytest/unit/sessions/test_heartbeat_lock_races.py @@ -0,0 +1,191 @@ +"""Interleavings between a heartbeat and the edits that displace it. + +`test_heartbeat_parked_zombie.py` pins the case where the tombstone has already been +written. These are the narrower windows around it: a beat that is not current but is not +(yet) tombstoned, and the read-then-write gaps inside the displacement paths themselves. +Each one is a state the locks alone cannot distinguish, so each is pinned by a test rather +than by a comment. +""" + +from typing import Optional +from unittest.mock import AsyncMock, patch +from uuid import uuid4 + +import pytest +import pytest_asyncio + +from oss.src.core.sessions.streams.dtos import ( + SessionHeartbeatRequest, + SessionStreamCommandRequest, +) +from oss.src.core.sessions.streams.service import SessionStreamsService +from oss.src.dbs.redis.sessions.locks import ( + clear_running, + force_clear_owner, + get_alive_owner, + get_owner, + get_running_owner, + is_turn_superseded, +) + +from unit.sessions.test_heartbeat_parked_zombie import _FakeStreamsDAO +from unit.sessions.test_project_scoped_locks import _FakeRedis + + +_PROJECT = uuid4() +_USER = uuid4() +_SESSION = "session_lock_races" + + +@pytest_asyncio.fixture +async def lock_engine(): + from oss.src.dbs.redis.shared.engine import LockEngine + + eng = LockEngine() + with patch.object(eng, "_client", return_value=_FakeRedis()): + yield eng + + +def _service(lock_engine, dao=None): + return SessionStreamsService( + streams_dao=dao or _FakeStreamsDAO(), lock_engine=lock_engine + ) + + +def _beat( + turn: str, *, running: bool = True, replica: str = "replica-a" +) -> SessionHeartbeatRequest: + return SessionHeartbeatRequest( + session_id=_SESSION, replica_id=replica, turn_id=turn, is_running=running + ) + + +def _cancel() -> SessionStreamCommandRequest: + return SessionStreamCommandRequest(session_id=_SESSION) + + +async def _alive(lock_engine) -> Optional[str]: + return await get_alive_owner( + lock_engine, project_id=str(_PROJECT), session_id=_SESSION + ) + + +async def _running(lock_engine) -> Optional[str]: + return await get_running_owner( + lock_engine, project_id=str(_PROJECT), session_id=_SESSION + ) + + +async def _owner(lock_engine) -> Optional[str]: + return await get_owner(lock_engine, project_id=str(_PROJECT), session_id=_SESSION) + + +async def _superseded(lock_engine, turn: str) -> bool: + return await is_turn_superseded( + lock_engine, project_id=str(_PROJECT), session_id=_SESSION, turn_id=turn + ) + + +# --------------------------------------------------------------------------- # +# Replica affinity +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_a_dead_turns_beat_does_not_reclaim_replica_affinity(lock_engine): + """`claim_owner` never steals, so an owner key that keeps getting renewed locks the + session out of every other replica for as long as the renewals continue. A tombstoned + turn's beat must read affinity, not claim it — otherwise a killed session's trailing + beats pin it to the replica that no longer runs anything.""" + dao = _FakeStreamsDAO() + svc = _service(lock_engine, dao) + + await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-a")) + assert await _owner(lock_engine) == "replica-a" + + await svc.command(project_id=_PROJECT, user_id=_USER, request=_cancel()) + assert await _superseded(lock_engine, "turn-a") is True + # What kill does to affinity, so another replica can take the session over. + await force_clear_owner(lock_engine, project_id=str(_PROJECT), session_id=_SESSION) + + late = await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-a")) + + assert late.is_current_turn is False + assert await _owner(lock_engine) is None, ( + "a dead turn's beat re-pinned the session to its replica for a full OWNER_TTL" + ) + assert late.replica_id == "replica-a", ( + "the caller still needs an owner back; reporting the beat's own replica is fine " + "because is_current_turn=False already tells it to stop" + ) + + +# --------------------------------------------------------------------------- # +# The handover's read-then-delete +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_handover_will_not_evict_a_turn_that_took_the_lock_mid_read(lock_engine): + """The handover branch reads the `alive` owner, then clears it. A real `_start_turn` can + land in that gap. Clearing unconditionally would delete the incoming turn's lock and + tombstone it — killing a turn that had just legitimately taken the session.""" + dao = _FakeStreamsDAO() + svc = _service(lock_engine, dao) + + # A parked holder: alive=turn-live, running cleared. + await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-live")) + await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-live", running=False)) + assert await _alive(lock_engine) == "turn-live" + assert await _running(lock_engine) is None + + # turn-new reads a value that is already stale by the time it writes. + with patch( + "oss.src.core.sessions.streams.service.get_alive_owner", + new=AsyncMock(return_value="turn-ghost"), + ): + result = await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-new")) + + assert await _alive(lock_engine) == "turn-live", ( + "the handover evicted the holder on the strength of a stale read" + ) + assert await _superseded(lock_engine, "turn-live") is False, ( + "worse than the eviction: the holder was tombstoned, so it can never beat back in" + ) + assert result.is_current_turn is False, ( + "losing the race is not an error — it just means this turn is not current" + ) + + +# --------------------------------------------------------------------------- # +# Displacement ordering +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_cancel_tombstones_before_it_clears_the_locks(lock_engine): + """Cancel clears `alive` and then tombstones the turn it displaced. A beat from that very + turn arriving between the two finds `alive` free, nx-acquires it back, and the cancelled + session reads as alive for a full ALIVE_TTL. Writing the tombstone first closes it.""" + dao = _FakeStreamsDAO() + svc = _service(lock_engine, dao) + await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-a")) + assert await _alive(lock_engine) == "turn-a" + + async def _beat_mid_displacement(engine, *, project_id: str, session_id: str): + await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-a")) + return await clear_running(engine, project_id=project_id, session_id=session_id) + + # `clear_running` runs after `alive` is cleared, i.e. inside the old window. + with patch( + "oss.src.core.sessions.streams.service.clear_running", + new=_beat_mid_displacement, + ): + await svc.command(project_id=_PROJECT, user_id=_USER, request=_cancel()) + + assert await _alive(lock_engine) is None, ( + "the cancelled turn's own beat re-armed `alive`; the session stays 'alive' until " + "the TTL expires and a follow-up send 409s" + ) + assert await _running(lock_engine) is None + assert await _superseded(lock_engine, "turn-a") is True diff --git a/api/oss/tests/pytest/unit/sessions/test_heartbeat_parked_zombie.py b/api/oss/tests/pytest/unit/sessions/test_heartbeat_parked_zombie.py new file mode 100644 index 0000000000..66ba40e2b7 --- /dev/null +++ b/api/oss/tests/pytest/unit/sessions/test_heartbeat_parked_zombie.py @@ -0,0 +1,346 @@ +"""The parked-session lock ambiguity (approvals plan §6) and its close. + +`alive` outlives its turn (`release_alive` has no callers; the turn-end beat clears only +`running`) and a turn parked awaiting approval also clears `running`. So the state +"`alive` held by a DIFFERENT turn + no `running`" is genuinely ambiguous between + + (a) a lapsed previous turn — the common case, which MUST be a legitimate handover or + every follow-up turn on a warm session aborts (that shipped once as a Critical + regression; `test_heartbeat_turn_handover.py` guards it), and + (b) a live-but-parked holder. + +We still resolve it as (a) — but a beat can only reach that branch if its turn has never +been displaced. Every displacement (handover, cancel, steer, kill, orphan sweep) tombstones +the turn it displaced, and a tombstoned turn's beats are refused before they touch a lock or +the row. A zombie is by definition a turn that already lost the nest, so the tombstone is +exactly the discriminator the locks alone cannot provide. + +These tests pin both horns: the zombie must not take a parked session's nest, and the +legitimate handover/steer/resume paths must keep working. +""" + +from typing import Optional +from unittest.mock import patch +from uuid import UUID, uuid4 + +import pytest +import pytest_asyncio + +from agenta.sdk.models.workflows import WorkflowServiceRequestData + +from oss.src.core.sessions.streams.dtos import ( + SessionHeartbeatRequest, + SessionStream, + SessionStreamCommandRequest, +) +from oss.src.core.sessions.streams.service import SessionStreamsService +from oss.src.dbs.redis.sessions.locks import ( + get_alive_owner, + get_running_owner, + is_turn_superseded, +) + +from unit.sessions.test_project_scoped_locks import _FakeRedis + + +_PROJECT = uuid4() +_USER = uuid4() +_SESSION = "session_parked_zombie" + + +class _FakeStreamsDAO: + def __init__(self, existing: Optional[SessionStream] = None): + self.row = existing + + async def get_by_session_id(self, *, project_id: UUID, session_id: str): + return self.row + + async def create(self, *, project_id, user_id, stream): + self.row = SessionStream( + id=uuid4(), + project_id=project_id, + session_id=stream.session_id, + flags=stream.flags, + turn_id=stream.turn_id, + ) + return self.row + + async def update(self, *, project_id, user_id, session_id, stream): + prior = self.row + self.row = SessionStream( + id=prior.id if prior else uuid4(), + project_id=project_id, + session_id=session_id, + flags=stream.flags + if stream.flags is not None + else (prior.flags if prior else None), + turn_id=stream.turn_id + if stream.turn_id is not None + else (prior.turn_id if prior else None), + ) + return self.row + + async def delete_by_session_id(self, *, project_id, session_id): + return True + + +@pytest_asyncio.fixture +async def lock_engine(): + from oss.src.dbs.redis.shared.engine import LockEngine + + eng = LockEngine() + with patch.object(eng, "_client", return_value=_FakeRedis()): + yield eng + + +def _service(lock_engine, dao=None): + return SessionStreamsService( + streams_dao=dao or _FakeStreamsDAO(), lock_engine=lock_engine + ) + + +def _beat(turn: str, *, running: bool = True) -> SessionHeartbeatRequest: + return SessionHeartbeatRequest( + session_id=_SESSION, replica_id="replica-a", turn_id=turn, is_running=running + ) + + +async def _alive(lock_engine) -> Optional[str]: + return await get_alive_owner( + lock_engine, project_id=str(_PROJECT), session_id=_SESSION + ) + + +async def _running(lock_engine) -> Optional[str]: + return await get_running_owner( + lock_engine, project_id=str(_PROJECT), session_id=_SESSION + ) + + +async def _superseded(lock_engine, turn: str) -> bool: + return await is_turn_superseded( + lock_engine, project_id=str(_PROJECT), session_id=_SESSION, turn_id=turn + ) + + +async def _park_a_session(svc, lock_engine) -> None: + """Drive the session to: turn-old ran and was handed over to turn-live, which then + parked awaiting approval (its run returned, so `release()` beat is_running=false).""" + await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-old")) + await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-old", running=False)) + await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-live")) + await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-live", running=False)) + assert await _alive(lock_engine) == "turn-live" + assert await _running(lock_engine) is None + + +# --------------------------------------------------------------------------- # +# Horn (b): the zombie +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_handover_tombstones_the_turn_it_displaced(lock_engine): + svc = _service(lock_engine) + await _park_a_session(svc, lock_engine) + + assert await _superseded(lock_engine, "turn-old") is True + assert await _superseded(lock_engine, "turn-live") is False, ( + "the turn that WON the handover must stay alive-eligible" + ) + + +@pytest.mark.asyncio +async def test_zombie_beat_cannot_take_the_nest_of_a_parked_session(lock_engine): + """The gap, precisely: a late beat from an older turn used to find the parked holder's + `alive` with no `running`, read that as a lapsed turn, and take the whole nest.""" + dao = _FakeStreamsDAO() + svc = _service(lock_engine, dao) + await _park_a_session(svc, lock_engine) + + zombie = await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-old")) + + assert zombie.is_current_turn is False, ( + "a turn that already lost the nest must be told it is not current" + ) + assert await _alive(lock_engine) == "turn-live", ( + "the zombie took the parked session's alive lock — the approval resume then " + "reports is_current_turn=false and aborts" + ) + assert await _running(lock_engine) is None, ( + "the zombie must not arm `running`: that key is the takeover discriminator the " + "resume reads, and stamping it makes the resume look superseded" + ) + assert dao.row is not None and dao.row.turn_id == "turn-live", ( + "a refused beat must not stamp its dead turn id on the durable row" + ) + + +@pytest.mark.asyncio +async def test_approval_resume_survives_a_zombie_beat(lock_engine): + """End to end: park, zombie beat, then the user approves and the resume turn starts.""" + svc = _service(lock_engine) + await _park_a_session(svc, lock_engine) + await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-old")) + + resumed = await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-resume")) + + assert resumed.is_current_turn is True, ( + "the approval resume must not be aborted — this is the user-visible failure" + ) + assert await _alive(lock_engine) == "turn-resume" + assert await _running(lock_engine) == "turn-resume" + + +@pytest.mark.asyncio +async def test_zombie_turn_end_beat_cannot_clear_the_live_turns_running(lock_engine): + """`clear_running` is unconditional, so a superseded turn's own is_running=false beat + used to end the LIVE turn's run. Refusing the beat before the branch fixes that too.""" + svc = _service(lock_engine) + await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-old")) + await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-old", running=False)) + await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-live")) + assert await _running(lock_engine) == "turn-live" + + await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-old", running=False)) + + assert await _running(lock_engine) == "turn-live" + assert await _alive(lock_engine) == "turn-live" + + +@pytest.mark.asyncio +async def test_repeated_zombie_beats_stay_refused(lock_engine): + """The tombstone is refreshed on every hit, so a zombie that keeps beating for longer + than the tombstone TTL never outlives its own death certificate.""" + svc = _service(lock_engine) + await _park_a_session(svc, lock_engine) + + for _ in range(3): + beat = await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-old")) + assert beat.is_current_turn is False + assert await _alive(lock_engine) == "turn-live" + + +# --------------------------------------------------------------------------- # +# Horn (a): the common case must keep working (the Critical regression guard) +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_a_never_displaced_turn_still_takes_a_lapsed_nest(lock_engine): + """The regression to fear: if a follow-up turn on a warm session stopped being able to + take the previous turn's stale `alive`, every follow-up turn would abort.""" + svc = _service(lock_engine) + + await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-1")) + await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-1", running=False)) + + second = await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-2")) + + assert second.is_current_turn is True + assert await _alive(lock_engine) == "turn-2" + assert await _running(lock_engine) == "turn-2" + + +@pytest.mark.asyncio +async def test_a_live_different_turn_is_still_a_real_takeover(lock_engine): + """The tombstone must not weaken the `running` discriminator: a turn superseded by a + genuinely live one still learns it lost, without any tombstone being involved.""" + svc = _service(lock_engine) + + await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-2")) + old = await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-1")) + + assert old.is_current_turn is False + assert await _alive(lock_engine) == "turn-2" + assert await _running(lock_engine) == "turn-2" + + +@pytest.mark.asyncio +async def test_two_overlapping_beats_of_one_turn_stay_current(lock_engine): + """A turn is never tombstoned by its own beats, however they interleave.""" + svc = _service(lock_engine) + + first = await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-1")) + second = await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-1")) + third = await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-1")) + + assert (first.is_current_turn, second.is_current_turn, third.is_current_turn) == ( + True, + True, + True, + ) + assert await _superseded(lock_engine, "turn-1") is False + + +# --------------------------------------------------------------------------- # +# The explicit control-plane displacements +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_cancel_tombstones_the_cancelled_turn(lock_engine): + svc = _service(lock_engine) + await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-1")) + + await svc.command( + project_id=_PROJECT, + user_id=_USER, + request=SessionStreamCommandRequest( + session_id=_SESSION, data=None, force=False + ), + ) + + assert await _superseded(lock_engine, "turn-1") is True + beat = await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-1")) + assert beat.is_current_turn is False + assert await _alive(lock_engine) is None, ( + "a cancelled turn's beat must not re-nest the session it was just cancelled out of" + ) + + +@pytest.mark.asyncio +async def test_steer_tombstones_the_displaced_turn_and_frees_the_new_one(lock_engine): + svc = _service(lock_engine) + await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-1")) + + steered = await svc.command( + project_id=_PROJECT, + user_id=_USER, + request=SessionStreamCommandRequest( + session_id=_SESSION, + data=WorkflowServiceRequestData(inputs={"messages": ["steer"]}), + force=True, + ), + ) + + assert await _superseded(lock_engine, "turn-1") is True + assert steered.turn_id is not None + assert await _superseded(lock_engine, steered.turn_id) is False + assert ( + await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-1")) + ).is_current_turn is False + # The steered-in turn owns the nest and keeps it. + assert await _alive(lock_engine) == steered.turn_id + assert await _running(lock_engine) == steered.turn_id + + +@pytest.mark.asyncio +async def test_kill_tombstones_the_killed_turn(lock_engine): + svc = _service(lock_engine) + await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-1")) + + with patch( + "oss.src.core.sessions.streams.service.kill_runner_sandbox", + new=_noop_kill, + ): + await svc.kill(project_id=_PROJECT, user_id=_USER, session_id=_SESSION) + + assert await _superseded(lock_engine, "turn-1") is True + beat = await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-1")) + assert beat.is_current_turn is False + assert await _alive(lock_engine) is None + + +async def _noop_kill(*, project_id: str, session_id: str) -> None: + return None diff --git a/api/oss/tests/pytest/unit/sessions/test_orphan_sweep_clears_redis.py b/api/oss/tests/pytest/unit/sessions/test_orphan_sweep_clears_redis.py index 3035d3080d..db97b1eb84 100644 --- a/api/oss/tests/pytest/unit/sessions/test_orphan_sweep_clears_redis.py +++ b/api/oss/tests/pytest/unit/sessions/test_orphan_sweep_clears_redis.py @@ -13,7 +13,7 @@ import pytest -from oss.src.dbs.redis.sessions.locks import get_session_liveness +from oss.src.dbs.redis.sessions.locks import get_session_liveness, is_turn_superseded from oss.src.core.sessions.streams.types import SessionTurnInUse from oss.src.tasks.asyncio.sessions.orphan_sweep import run_orphan_sweep @@ -150,6 +150,38 @@ def _send_gate(liveness): _send_gate(liveness_after) # must not raise +@pytest.mark.anyio +async def test_orphan_sweep_tombstones_the_turn_it_swept(anyio_backend): + """A swept turn is declared dead. Without a tombstone its next beat would find the nest + empty, re-acquire `alive` under its own id, and put the session straight back into the + orphaned state the sweep just cleaned up. + """ + assert anyio_backend == "asyncio" + + lock_engine = _FakeRedis() + await lock_engine.set( + f"alive:{_PROJECT_ID}:session:{_SESSION_ID}", b"turn-1", ex=3600 + ) + await lock_engine.set( + f"running:{_PROJECT_ID}:session:{_SESSION_ID}", b"turn-1", ex=3600 + ) + stale_row = _FakeRow( + session_id=_SESSION_ID, + updated_at=datetime.now(timezone.utc) - timedelta(seconds=600), + ) + + await run_orphan_sweep(_FakeTransactionsEngine([stale_row]), lock_engine) + + assert ( + await is_turn_superseded( + lock_engine, + project_id=_PROJECT_ID, + session_id=_SESSION_ID, + turn_id="turn-1", + ) + ) is True + + @pytest.mark.anyio async def test_orphan_sweep_selects_rows_never_updated_since_creation(anyio_backend): """A row whose heartbeat never wrote it has `updated_at` NULL, and `NULL < threshold` diff --git a/api/oss/tests/pytest/unit/sessions/test_orphaned_gate_reconciliation.py b/api/oss/tests/pytest/unit/sessions/test_orphaned_gate_reconciliation.py new file mode 100644 index 0000000000..5f606a43d2 --- /dev/null +++ b/api/oss/tests/pytest/unit/sessions/test_orphaned_gate_reconciliation.py @@ -0,0 +1,250 @@ +"""The orphaned-gate safety net: no HITL gate may outlive its turn. + +A `session_interactions` row is only ever created when a turn PAUSES for a human. So once a +turn reaches its terminal `done` record WITHOUT the pause marker, nothing is holding a gate for +that session and any row still `pending` is unanswerable — yet both inboxes keep offering it +(the live "orphaned gate": a stuck approval that can never be answered). + +The records worker is the choke point: it sees every turn's terminal record. Two guards keep a +legitimately parked gate alive: + + * `stopReason: "paused"` on the terminal record means THAT turn is the live park — skip it; + * the cancel is scoped to the finished turn's OWN gates, so a newer turn's park is out of + range by construction. This worker can lag arbitrarily far behind the stream without ever + cancelling underneath a turn that is parked right now — no ordering, no ledger read, and so + no window between checking and cancelling. Prior turns' leftovers belong to the runner's + turn-start sweep (`/sessions/interactions/cancel-stale`), not to this one. + +Cancelling is deliberately a DIFFERENT terminal status from a user's deny: a deny is +`resolved` + `data.resolution.verdict == "denied"`; a superseded gate is `cancelled`, which +already means "the runner abandoned the gate; no one is waiting on the token". +""" + +import zlib +from unittest.mock import AsyncMock +from uuid import uuid4 + +import pytest +from orjson import dumps + +from oss.src.core.sessions.records.dtos import SessionRecord +from oss.src.core.sessions.records.service import RecordsService +from oss.src.tasks.asyncio.sessions.records_worker import ( + RecordsWorker, + finished_turns_in_batch, +) + + +PROJECT = uuid4() +SESSION = "sess-orphan" +GATE_TURN = "11111111-1111-4111-8111-111111111111" +RESUME_TURN = "22222222-2222-4222-8222-222222222222" + + +class _Event: + """The deserialized stream message shape the worker consumes (`msg.record_event`).""" + + def __init__(self, record_event): + self.record_event = record_event + + +class _Record: + def __init__(self, *, session_id, record_type, turn_id=None, attributes=None): + self.session_id = session_id + self.record_type = record_type + self.turn_id = turn_id + self.attributes = attributes + + +def _done(turn_id, *, paused=False, session_id=SESSION): + attributes = {"type": "done", "traceId": "t"} + if paused: + attributes["stopReason"] = "paused" + return _Event( + _Record( + session_id=session_id, + record_type="done", + turn_id=turn_id, + attributes=attributes, + ) + ) + + +def _worker(*, interactions=None, interactions_wired=True): + interactions_service = None + if interactions_wired: + interactions_service = interactions or AsyncMock() + if interactions is None: + interactions_service.cancel_session_pending = AsyncMock(return_value=1) + return RecordsWorker( + service=RecordsService(records_dao=AsyncMock()), + redis_client=None, + stream_name="streams:records", + consumer_group="worker-records", + interactions_service=interactions_service, + ) + + +# --------------------------------------------------------------- the batch predicate + + +def test_finished_turns_picks_a_completed_turn(): + assert finished_turns_in_batch([_done(RESUME_TURN)]) == {SESSION: RESUME_TURN} + + +def test_finished_turns_skips_a_paused_turn(): + """The pause marker IS the live park; sweeping on it would cancel the gate the human is + being asked to answer.""" + assert finished_turns_in_batch([_done(GATE_TURN, paused=True)]) == {} + + +def test_finished_turns_ignores_non_terminal_records_and_untagged_turns(): + events = [ + _Event( + _Record( + session_id=SESSION, + record_type="message", + turn_id=RESUME_TURN, + attributes={"type": "message"}, + ) + ), + _Event(_Record(session_id=SESSION, record_type="done", attributes={})), + ] + assert finished_turns_in_batch(events) == {} + + +def test_finished_turns_tolerates_a_null_attributes_record(): + events = [ + _Event(_Record(session_id=SESSION, record_type="done", turn_id=RESUME_TURN)) + ] + assert finished_turns_in_batch(events) == {SESSION: RESUME_TURN} + + +# --------------------------------------------------------------- the reconciliation + + +@pytest.mark.asyncio +async def test_a_finished_turn_cancels_the_gate_it_orphaned(): + """The live bug: turn 1 parks a gate, the resume runs as turn 2 and finishes without ever + binding to the park, so nothing resolves the row. Turn 2's terminal record must clear it.""" + worker = _worker() + + await worker.reconcile_orphaned_gates( + project_id=PROJECT, + events=[_done(GATE_TURN, paused=True), _done(RESUME_TURN)], + ) + + worker.interactions_service.cancel_session_pending.assert_awaited_once_with( + project_id=PROJECT, + session_id=SESSION, + only_turn_id=RESUME_TURN, + ) + + +@pytest.mark.asyncio +async def test_a_parked_gate_is_not_cancelled(): + """MUST NOT cancel: the only turn in the batch paused, so its gate is live and awaiting a + human. This is the regression guard for the whole safety net.""" + worker = _worker() + + await worker.reconcile_orphaned_gates( + project_id=PROJECT, + events=[_done(GATE_TURN, paused=True)], + ) + + worker.interactions_service.cancel_session_pending.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_a_newer_turns_park_is_out_of_range_however_late_this_runs(): + """Worker lag, which used to be guarded by re-reading the turns ledger — a read that could + go stale before the cancel landed, taking a just-parked turn's gate with it. The scope makes + the question moot: whatever turn 3 is doing right now, only turn 2's rows are addressed.""" + worker = _worker() + + await worker.reconcile_orphaned_gates( + project_id=PROJECT, + events=[_done(RESUME_TURN)], + ) + + _, kwargs = worker.interactions_service.cancel_session_pending.await_args + assert kwargs["only_turn_id"] == RESUME_TURN, ( + "an unscoped cancel takes every pending row in the session, including the gate a " + "newer turn parked while this batch was queued" + ) + + +@pytest.mark.asyncio +async def test_reconciliation_is_disabled_without_the_interactions_service(): + worker = _worker(interactions_wired=False) + + await worker.reconcile_orphaned_gates( + project_id=PROJECT, + events=[_done(RESUME_TURN)], + ) + + +@pytest.mark.asyncio +async def test_a_reconciliation_failure_never_propagates(): + """The record append is already committed; a safety-net failure must not re-drive it.""" + interactions = AsyncMock() + interactions.cancel_session_pending = AsyncMock(side_effect=RuntimeError("db down")) + worker = _worker(interactions=interactions) + + await worker.reconcile_orphaned_gates( + project_id=PROJECT, + events=[_done(RESUME_TURN)], + ) + + +@pytest.mark.asyncio +async def test_process_batch_reconciles_after_the_append(): + """End-to-end through the stream loop: the cancel runs post-append (so a woken client sees + the cleared row) and the append still reports its count.""" + journal: list = [] + records_dao = AsyncMock() + + async def _append_many(*, events): + journal.append("append") + return [ + SessionRecord(record_id=uuid4(), session_id=SESSION, project_id=PROJECT) + for _ in events + ] + + records_dao.append_many = AsyncMock(side_effect=_append_many) + + interactions = AsyncMock() + + async def _cancel(*, project_id, session_id, only_turn_id): + journal.append("cancel") + return 1 + + interactions.cancel_session_pending = AsyncMock(side_effect=_cancel) + + worker = RecordsWorker( + service=RecordsService(records_dao=records_dao), + redis_client=None, + stream_name="streams:records", + consumer_group="worker-records", + interactions_service=interactions, + ) + + message = { + "organization_id": None, + "project_id": str(PROJECT), + "record_event": { + "project_id": str(PROJECT), + "session_id": SESSION, + "record_index": 0, + "record_type": "done", + "turn_id": RESUME_TURN, + "attributes": {"type": "done"}, + }, + } + appended, processed = await worker.process_batch( + [(b"1-1", {b"data": zlib.compress(dumps(message))})] + ) + + assert appended == 1 + assert processed == [b"1-1"] + assert journal == ["append", "cancel"] diff --git a/api/oss/tests/pytest/unit/sessions/test_watch_endpoint.py b/api/oss/tests/pytest/unit/sessions/test_watch_endpoint.py index 2589416214..49bf2e8a8a 100644 --- a/api/oss/tests/pytest/unit/sessions/test_watch_endpoint.py +++ b/api/oss/tests/pytest/unit/sessions/test_watch_endpoint.py @@ -18,10 +18,13 @@ from oss.src.apis.fastapi.sessions.watch import ( HEARTBEAT_FRAME, format_watch_frame, + ready_frame, + retry_frame, watch_event_stream, ) from oss.src.dbs.redis.sessions.contract import watch_channel from oss.src.dbs.redis.sessions.watch import SessionsWatchPublisher +from oss.src.utils.env import env class _FakePubSub: @@ -65,25 +68,30 @@ async def test_stream_yields_event_frames_then_heartbeats(): channel="watch:p:session:s1", pubsub_factory=lambda: pubsub, heartbeat_seconds=0.01, + retry_milliseconds=5000, ) frames = [] async for frame in stream: frames.append(frame) - if len(frames) == 4: + if len(frames) == 6: await stream.aclose() break - assert frames[0].startswith("event: records-changed\n") - assert json.loads(frames[0].split("data: ")[1]) == { + # The preamble pins the client's built-in auto-reconnect delay, then `ready` marks the + # point where the subscription is live and a client may safely revalidate. + assert frames[0] == retry_frame(5000) + assert frames[1] == ready_frame() + assert frames[2].startswith("event: records-changed\n") + assert json.loads(frames[2].split("data: ")[1]) == { "type": "records-changed", "session_id": "s1", } - assert frames[1].startswith("event: lifecycle\n") - assert '"state": "running"' in frames[1] - assert frames[2].startswith("event: interaction\n") + assert frames[3].startswith("event: lifecycle\n") + assert '"state": "running"' in frames[3] + assert frames[4].startswith("event: interaction\n") # Queue drained -> the idle path emits keep-alive comments. - assert frames[3] == HEARTBEAT_FRAME + assert frames[5] == HEARTBEAT_FRAME assert pubsub.subscribed == ["watch:p:session:s1"] @@ -94,10 +102,12 @@ async def test_stream_cleans_up_subscription_on_close(): channel="watch:p:session:s1", pubsub_factory=lambda: pubsub, heartbeat_seconds=0.01, + retry_milliseconds=5000, ) - # Take one heartbeat, then simulate the client disconnecting. - frame = await stream.__anext__() - assert frame == HEARTBEAT_FRAME + # Take the preamble + one heartbeat, then simulate the client disconnecting. + assert await stream.__anext__() == retry_frame(5000) + assert await stream.__anext__() == ready_frame() + assert await stream.__anext__() == HEARTBEAT_FRAME await stream.aclose() assert pubsub.unsubscribed == ["watch:p:session:s1"] @@ -118,13 +128,43 @@ async def test_stream_skips_malformed_and_unknown_payloads(): channel="watch:p:session:s1", pubsub_factory=lambda: pubsub, heartbeat_seconds=0.01, + retry_milliseconds=5000, ) + assert await stream.__anext__() == retry_frame(5000) + assert await stream.__anext__() == ready_frame() frame = await stream.__anext__() await stream.aclose() # The three junk messages are dropped; the first frame is the real event. assert frame.startswith("event: records-changed\n") +@pytest.mark.asyncio +async def test_stream_preamble_pins_the_clients_reconnect_delay(): + """Without a `retry:` field the browser's auto-reconnect delay is + implementation-defined — an API restart then reconnect-storms us.""" + pubsub = _FakePubSub([]) + stream = watch_event_stream( + channel="watch:p:session:s1", + pubsub_factory=lambda: pubsub, + heartbeat_seconds=0.01, + retry_milliseconds=7500, + ) + first = await stream.__anext__() + await stream.aclose() + + assert first == "retry: 7500\n\n" + # The preamble follows SUBSCRIBE, so no event can land in an unsubscribed + # window between the client's `open` and the first frame. + assert pubsub.subscribed == ["watch:p:session:s1"] + + +def test_retry_frame_is_a_field_only_sse_frame(): + frame = retry_frame(5000) + assert frame == "retry: 5000\n\n" + # A field-only frame sets the reconnect time without dispatching an event. + assert "data:" not in frame and "event:" not in frame + + def test_format_watch_frame_rejects_non_dict_and_unknown_type(): assert format_watch_frame(b"[1, 2]") is None assert format_watch_frame(b"\xff\xfe") is None @@ -150,8 +190,11 @@ async def test_stream_delivers_publisher_events_end_to_end(): channel=channel, pubsub_factory=lambda: redis.pubsub(), heartbeat_seconds=0.05, + retry_milliseconds=5000, ) - # First frame is a heartbeat — proves the subscription is live before publishing. + # Preamble, then a heartbeat — proves the subscription is live before publishing. + assert await stream.__anext__() == retry_frame(5000) + assert await stream.__anext__() == ready_frame() assert await stream.__anext__() == HEARTBEAT_FRAME publisher = SessionsWatchPublisher(redis_client=redis) @@ -232,3 +275,52 @@ async def test_watch_endpoint_returns_event_stream_response(): assert response.media_type == "text/event-stream" assert response.headers["cache-control"] == "no-cache" assert response.headers["x-accel-buffering"] == "no" + + +@pytest.mark.asyncio +async def test_watch_endpoint_streams_the_configured_retry_preamble(): + """The env-configured reconnect delay actually reaches the wire.""" + router = _router() + request = _make_authed_request(FastAPI(), uuid4(), uuid4()) + pubsub = _FakePubSub([]) + + with ( + patch( + "oss.src.apis.fastapi.sessions.router.check_action_access", + new_callable=AsyncMock, + return_value=True, + ), + patch( + "oss.src.apis.fastapi.sessions.router.get_streams_engine" + ) as streams_engine, + patch.object(env.sessions, "watch_retry_milliseconds", 9000), + ): + streams_engine.return_value.get_redis.return_value.pubsub.return_value = pubsub + response = await router.watch_session_stream(request=request, session_id="s-1") + first = await response.body_iterator.__anext__() + await response.body_iterator.aclose() + + assert first == "retry: 9000\n\n" + + +@pytest.mark.asyncio +async def test_ready_is_not_emitted_before_the_subscription_is_live(): + """The whole point of the `ready` event. A client revalidating on `onopen` races the + subscription: Starlette flushes the response headers before it iterates this generator, so a + change can land and publish in between and reach neither the refetch nor the stream. `ready` + is only reachable after `subscribe` has returned.""" + pubsub = _FakePubSub([]) + stream = watch_event_stream( + channel="watch:p:session:s1", + pubsub_factory=lambda: pubsub, + heartbeat_seconds=0.01, + retry_milliseconds=5000, + ) + + assert await stream.__anext__() == retry_frame(5000) + assert await stream.__anext__() == ready_frame() + assert pubsub.subscribed == ["watch:p:session:s1"], ( + "ready reached the client before the channel was subscribed" + ) + + await stream.aclose() diff --git a/docs/design/agenta-mobile/plans/2026-07-27-mobile-approvals-steering.md b/docs/design/agenta-mobile/plans/2026-07-27-mobile-approvals-steering.md index e92ab92323..ead77482d7 100644 --- a/docs/design/agenta-mobile/plans/2026-07-27-mobile-approvals-steering.md +++ b/docs/design/agenta-mobile/plans/2026-07-27-mobile-approvals-steering.md @@ -346,7 +346,7 @@ Execution scope now: M0 + M1 (minus M1.5 steer) + TTL bump + M2. noted slow). The M0.3/M1.3 tightened cadence must be foreground-only + only while running/pending; back off on `visibilitychange`. -## 6. Tracked residual: parked-session lock ambiguity (found 2026-07-28) +## 6. CLOSED: parked-session lock ambiguity (found 2026-07-28, closed 2026-07-30) Live QA of approvals surfaced a dead liveness mirror in `SessionStreamsService.heartbeat` (fixed: b0281c5788, 6761727847, 2174162d80, 5d2ed61e9f, 076dc41b7e — see the memory entry @@ -365,11 +365,49 @@ reports `is_current_turn=False` and aborts. Narrow today (`_start_turn` is off t path; cross-container zombies are blocked by the non-stealing `claim_owner` affinity key), but real. -**Fix options (pick when the send/steer path gets wired):** store `alive` as -`{turn_id, state}` or add a sibling `parked:` key so a parked/starting holder is -distinguishable from a lapsed one; or give `release_alive` an actual caller so `alive` stops -outliving its turn (the root cause). Either way the runner's `startAliveWatchdog` must be -updated in lockstep. +**The fix that landed (2026-07-30): supersession tombstones.** Neither option originally +sketched here survives contact with the failure: + +- *Self-describing `alive` (`{turn_id, state}` / a sibling `parked:` key)* tells a parked + holder from a lapsed one, but that is not the decision the heartbeat has to make. The + approval RESUME is itself a different turn that MUST displace the parked holder, and it is + indistinguishable from the zombie at the lock layer — so "parked ⇒ don't steal" blocks the + resume, which is the shipped-Critical failure mode ("every follow-up turn aborts") in a new + costume. Making it work needs the resume to announce itself on the heartbeat wire = a + runner wire change + restart. +- *Giving `release_alive` a caller* does not fix the zombie at all (the zombie then simply + takes an EMPTY nest and the resume still reads `is_current_turn=False`), and it makes a + parked session `is_alive=false` — which flips the SEND gate open mid-park, inviting the + very concurrent-turn state that produces zombies. It trades one gap for a worse one. + +What is actually knowable, at the exact moment it happens, is that a turn was **displaced**. +So every displacement — the heartbeat handover, cancel, steer, kill, orphan sweep — now +tombstones the turn it displaced (`superseded::session::turn:`, +TTL-refreshed on every hit so a long-lived zombie never outlives its own death certificate), +and a tombstoned turn's beats are refused before they touch any lock or the row. A zombie is +by definition a turn that already lost the nest, so `displaced ⇒ dead` is precisely the +discriminator the locks cannot provide. The ambiguous state keeps resolving as (a) — no +regression risk to the warm-session handover — but only a never-displaced turn can reach it. + +Side effects worth knowing: a superseded turn's own `is_running=false` beat no longer clears +the LIVE turn's `running` (`clear_running` is unconditional), and a cancelled turn's beat no +longer re-acquires `alive` under its dead id. + +**The runner needed no change.** The heartbeat wire is unchanged and `is_current_turn: false` +already means "you lost the session, abort" — a refused beat is exactly that. Nothing here +requires a runner restart. Runner-side tests were extended to pin its half of the contract +(no beats after `release()`; a refused beat aborts). + +Code: `api/oss/src/dbs/redis/sessions/{contract,locks}.py`, +`api/oss/src/core/sessions/streams/service.py`, +`api/oss/src/tasks/asyncio/sessions/orphan_sweep.py`, `api/oss/src/utils/env.py`. +Tests: `api/oss/tests/pytest/unit/sessions/test_heartbeat_parked_zombie.py` (+ the orphan +sweep and runner alive suites). + +**Residual, accepted and narrower:** `_start_turn` acquires `alive` then `running` in two +Redis round-trips; a never-displaced turn beating inside that sub-millisecond window would +still read the state as a handover. Only reachable with two genuinely concurrent turns on one +session — an un-gated-concurrent-`/invoke` hazard, not a lock-contract one. **Also worth knowing:** `updated_at` is bumped by non-heartbeat writers (attach/detach, rename), so watcher churn can hold an orphan's sweep clock open — now a 30-minute window for diff --git a/services/runner/src/engines/sandbox_agent/run-turn.ts b/services/runner/src/engines/sandbox_agent/run-turn.ts index 34a789fc4d..3d45e9ee2a 100644 --- a/services/runner/src/engines/sandbox_agent/run-turn.ts +++ b/services/runner/src/engines/sandbox_agent/run-turn.ts @@ -17,6 +17,7 @@ import { ConversationDecisions, extractApprovalDecisions, extractClientToolOutputs, + extractInBandApprovalAnswers, } from "../../responder.ts"; import { buildInteractionData, @@ -132,6 +133,10 @@ export async function runTurn( // stop this turn's relay on EVERY exit path (a cleared sink must never orphan it). let otel: ReturnType | undefined; let activeTurn: CurrentTurn | undefined; + // Assigned once the turn's interaction plumbing exists; called from the `finally` so EVERY exit + // path (done, paused, cancelled, error) settles the durable rows this turn's in-band answers + // consumed. Without it, a resume the harness does not re-gate leaves them `pending` forever. + let settleInBandInteractions: (() => Promise) | undefined; // Time-based run deadlines (total/idle/TTFB/per-tool-call) for THIS turn: an idle/wedged harness // has no deadline anywhere, so a silent or hung turn would hold its sandbox forever. Tripping a @@ -607,10 +612,12 @@ export async function runTurn( // interactions-plane answer already transitioned it to responded, and an in-band answer is // detected at sweep time (`inBandAnswerToken`) and exempted via the sweep's `tokens` — the // row stays pending until this resolve lands it as resolved, never cancelled. + const resolvedInteractionTokens = new Set(); const resolveInteractionToken = ( token: string, verdict?: { approved: boolean; toolCallId: string }, ): void => { + resolvedInteractionTokens.add(token); if (verdict) { run.emitEvent({ type: "interaction_response", @@ -633,6 +640,43 @@ export async function runTurn( : undefined, ); }; + // A resume's approval envelope is a CONSUMED decision even when the harness never re-raises + // the gate: on a cold replay the transcript already contains the human's answer, so the agent + // just proceeds and no reply path ever reaches `resolveInteractionToken`. That is exactly how a + // gate outlives its turn as a forever-actionable `pending` row. Settle the leftovers here, with + // the verdict the human actually gave, so the row lands `resolved` and not `cancelled`. A row + // already terminal (resolved by the reply path, or cancelled by the turn-start sweep) simply + // 404s the transition — the CAS is the arbiter, this is only a last writer. + // + // Deliberately does NOT go through `resolveInteractionToken`: that emits an + // `interaction_response` event, and this can run once the turn's event stream is closed, + // which would land a record after the turn's terminal `done`. The durable row is the only + // thing to fix. + // + // Awaited, and awaited BEFORE the terminal record is emitted. `record()` hands `done` + // straight to the sink, so the moment `finish()` runs the API's gate reconciliation may + // consume it and cancel this still-pending row — after which the transition below finds a + // terminal row and 404s, filing a decision the human actually made as an abandonment. + settleInBandInteractions = async (): Promise => { + const cred = runCredential(request); + if (!cred) return; + const settling: Promise[] = []; + for (const answer of extractInBandApprovalAnswers(request)) { + if (resolvedInteractionTokens.has(answer.token)) continue; + resolvedInteractionTokens.add(answer.token); + logger( + `[HITL] settling in-band answer with no harness gate token=${answer.token} ` + + `approved=${answer.approved}`, + ); + settling.push( + resolveInteraction(sessionId, answer.token, () => cred, { + verdict: answer.approved ? "approved" : "denied", + tool_call_id: answer.toolCallId, + }), + ); + } + await Promise.all(settling); + }; const serverPermissions = serverPermissionsFromRequest(request); // The SAME name->spec index the relay execute loop hands to the relay execution guard, so // the approval card and the guard cannot disagree about a tool's permission/readOnly. @@ -1055,6 +1099,8 @@ export async function runTurn( run.emitEvent({ type: "error", message: swallowedError }); } + // Before `finish()`, which emits the terminal `done` the API reconciles gates against. + await settleInBandInteractions?.(); const output = run.finish(stopReason); await run.flush(); const turnEndedAt = new Date().toISOString(); @@ -1127,6 +1173,8 @@ export async function runTurn( otel?.emitEvent({ type: "error", message: error }); // An aborted turn may have left a partial turn in the native transcript. invalidateContinuity(sessionId, plan.harness, deps); + // Same ordering as the happy path: settle the durable rows before the terminal record goes out. + await settleInBandInteractions?.(); // finish() must not throw uncaught — tracing must not mask the run error. try { otel?.finish(); @@ -1134,6 +1182,10 @@ export async function runTurn( await otel?.flush().catch(() => {}); return { ok: false, error }; } finally { + // Backstop for the exits that reach neither branch above (cancel, abort). Idempotent via the + // resolved-token set, so the ordered calls make this a no-op on the paths that took them, and + // never throws — a row whose gate is gone is unanswerable however the turn ended. + void settleInBandInteractions?.(); // Release every run-limits timer (idempotent, never re-arms on a late event) on EVERY path. runLimits.dispose(); // This turn owns its relay: stop it on EVERY exit path (the happy path already stopped it diff --git a/services/runner/src/responder.ts b/services/runner/src/responder.ts index 9a36ae8ddc..16f3968d08 100644 --- a/services/runner/src/responder.ts +++ b/services/runner/src/responder.ts @@ -382,6 +382,45 @@ export function extractApprovalDecisions( return decisions; } +/** One approval envelope this turn received in-band, carrying its durable row's token. */ +export type InBandApprovalAnswer = { + /** The `session_interactions.token` of the row the human answered. */ + token: string; + approved: boolean; + toolCallId: string; +}; + +/** + * The approval envelopes THIS turn received in-band, keyed by the durable interaction token the + * client echoed back. A resume delivers the human's yes/no as a `tool_result` envelope; the turn + * consumes that decision whether or not the harness re-raises the gate. When it does re-raise, + * the responder resolves the row on reply; when it does NOT (a cold replay whose transcript + * already contains the envelope, so the agent simply proceeds), nothing else ever touches the + * row — it is this list that lets the turn settle it instead of orphaning it as `pending`. + * + * Scoped to the CURRENT turn (results at/after the latest user message) so a long session does + * not re-settle every prior turn's already-terminal row on every turn. Envelopes without an + * `interactionToken` are skipped: without it there is no durable row to key on. + */ +export function extractInBandApprovalAnswers( + request: AgentRunRequest, +): InBandApprovalAnswer[] { + const answers: InBandApprovalAnswer[] = []; + const seen = new Set(); + for (const block of currentTurnToolResultBlocks(request)) { + const stored = storedApprovalDecisionOf(block); + const token = stored?.interactionToken; + if (!stored || !token || seen.has(token)) continue; + seen.add(token); + answers.push({ + token, + approved: stored.decision === "allow", + toolCallId: block.toolCallId ?? token, + }); + } + return answers; +} + /** * Build the client-tool output store from the inbound history: every NON-approval `tool_result` * is a browser-fulfilled client-tool output. Keyed by the cold-replay anchor diff --git a/services/runner/tests/unit/orphaned-gate-settle.test.ts b/services/runner/tests/unit/orphaned-gate-settle.test.ts new file mode 100644 index 0000000000..b558cf4a41 --- /dev/null +++ b/services/runner/tests/unit/orphaned-gate-settle.test.ts @@ -0,0 +1,127 @@ +/** + * The orphaned-gate settle: a resume must not leave its durable row `pending` forever. + * + * A resume delivers the human's yes/no as a `tool_result` approval envelope. When the harness + * re-raises the gate, the reply path resolves the row. When it does NOT — a cold replay whose + * transcript already contains the answer, so the agent simply proceeds — nothing else ever + * touches the row. The turn-start stale sweep can't cover it either: it EXEMPTS exactly this + * token (server.ts `staleInteractionExemptTokens`) on the promise that the resume will resolve + * it. That promise is what breaks when the resume degrades to cold, and the row is left + * forever-actionable in both inboxes. + * + * `extractInBandApprovalAnswers` is the list the turn settles from at its end. + */ +import { describe, it } from "vitest"; +import assert from "node:assert/strict"; + +import type { AgentRunRequest } from "../../src/protocol.ts"; +import { extractInBandApprovalAnswers } from "../../src/responder.ts"; + +const TOKEN = "tok-gate-1"; +const TOOL_CALL_ID = "toolu_write_1"; + +/** The live mobile resume shape: full history, decision stamped on the tail, real toolCallId. */ +function resumeRequest( + overrides: Partial = {}, +): AgentRunRequest { + return { + messages: [ + { role: "user", content: "write out.txt" }, + { + role: "assistant", + content: [ + { + type: "tool_call", + toolCallId: TOOL_CALL_ID, + toolName: "Write", + input: { path: "out.txt" }, + }, + ], + }, + { + role: "assistant", + content: [ + { + type: "tool_result", + toolCallId: TOOL_CALL_ID, + toolName: "Write", + output: { approved: true, interactionToken: TOKEN }, + }, + ], + }, + ], + ...overrides, + } as AgentRunRequest; +} + +describe("extractInBandApprovalAnswers", () => { + it("recovers the token and verdict from an approval envelope", () => { + assert.deepEqual(extractInBandApprovalAnswers(resumeRequest()), [ + { token: TOKEN, approved: true, toolCallId: TOOL_CALL_ID }, + ]); + }); + + it("carries a deny through as approved:false, not as a dropped answer", () => { + const request = resumeRequest(); + (request.messages as any)[2].content[0].output = { + approved: false, + interactionToken: TOKEN, + }; + assert.deepEqual(extractInBandApprovalAnswers(request), [ + { token: TOKEN, approved: false, toolCallId: TOOL_CALL_ID }, + ]); + }); + + it("skips an envelope with no interactionToken — there is no row to key on", () => { + const request = resumeRequest(); + (request.messages as any)[2].content[0].output = { approved: true }; + assert.deepEqual(extractInBandApprovalAnswers(request), []); + }); + + it("ignores a client-tool output — only approval envelopes carry a gate decision", () => { + const request = resumeRequest(); + (request.messages as any)[2].content[0].output = "file written"; + assert.deepEqual(extractInBandApprovalAnswers(request), []); + }); + + it("deduplicates a token repeated across the history", () => { + const request = resumeRequest(); + (request.messages as any).push((request.messages as any)[2]); + assert.deepEqual(extractInBandApprovalAnswers(request), [ + { token: TOKEN, approved: true, toolCallId: TOOL_CALL_ID }, + ]); + }); + + it("collects one answer per gate when a parallel batch is answered together", () => { + const request = resumeRequest(); + (request.messages as any)[2].content.push({ + type: "tool_result", + toolCallId: "toolu_bash_2", + toolName: "Bash", + output: { approved: false, interactionToken: "tok-gate-2" }, + }); + assert.deepEqual(extractInBandApprovalAnswers(request), [ + { token: TOKEN, approved: true, toolCallId: TOOL_CALL_ID }, + { token: "tok-gate-2", approved: false, toolCallId: "toolu_bash_2" }, + ]); + }); + + it("scopes to the current turn: a prior turn's already-terminal row is not re-settled", () => { + // A long session's history keeps every past envelope. Re-settling them on every turn would + // fire a 404 transition per approval per turn; the row is already terminal. + const request = resumeRequest(); + (request.messages as any).push({ role: "user", content: "now do something else" }); + assert.deepEqual(extractInBandApprovalAnswers(request), []); + }); + + it("returns nothing for a plain new turn", () => { + const request = { + messages: [{ role: "user", content: "hello" }], + } as AgentRunRequest; + assert.deepEqual(extractInBandApprovalAnswers(request), []); + }); + + it("returns nothing for an empty request", () => { + assert.deepEqual(extractInBandApprovalAnswers({} as AgentRunRequest), []); + }); +}); diff --git a/services/runner/tests/unit/session-alive-interrupt.test.ts b/services/runner/tests/unit/session-alive-interrupt.test.ts index 72bd2c0380..26e45881b7 100644 --- a/services/runner/tests/unit/session-alive-interrupt.test.ts +++ b/services/runner/tests/unit/session-alive-interrupt.test.ts @@ -102,6 +102,25 @@ describe("startAliveWatchdog onInterrupted", () => { assert.equal(onInterrupted.mock.calls.length, 1); }); + it("a refused beat (superseded turn) aborts the run and never throws", async () => { + // The API refuses every beat from a turn it has tombstoned as superseded — including + // the turn-end beat — by answering `is_current_turn: false`. The runner needs no new + // wire field to understand that: the existing signal already means "you lost the + // session, abort". This pins that the refusal path is the abort path. + nextIsCurrentTurn = false; + const onInterrupted = vi.fn(); + const watchdog = await startAliveWatchdog( + "sess-superseded", + "turn-superseded", + "proj-1", + onInterrupted, + ); + await flushMicrotasks(); + + assert.equal(onInterrupted.mock.calls.length, 1); + await assert.doesNotReject(() => watchdog.release()); + }); + it("treats a network/HTTP failure as NOT interrupted (fail-open)", async () => { vi.stubGlobal("fetch", async () => { throw new Error("network down"); diff --git a/services/runner/tests/unit/session-alive.test.ts b/services/runner/tests/unit/session-alive.test.ts index 1cff659c27..b4d118cd06 100644 --- a/services/runner/tests/unit/session-alive.test.ts +++ b/services/runner/tests/unit/session-alive.test.ts @@ -119,6 +119,52 @@ describe("startAliveWatchdog", () => { ); }); + it("sends no further heartbeats after release() — the zombie-beat source", async () => { + // A turn that ended or parked must stop beating. A beat that outlives its turn is what + // the API's supersession tombstone exists to refuse (approvals plan §6): it would + // otherwise find the parked holder's `alive` with no `running` and take the whole nest, + // making the user's approval resume look superseded. Stopping the interval BEFORE the + // final beat is the runner's half of that contract. + vi.useFakeTimers(); + try { + const watchdog = await startAliveWatchdog( + "sess-parked", + "turn-parked", + "proj-1", + ); + await watchdog.release(); + const afterRelease = fetchCalls.filter((c) => + c.url.includes("heartbeat"), + ).length; + + await vi.advanceTimersByTimeAsync(5 * 30_000); + + assert.equal( + fetchCalls.filter((c) => c.url.includes("heartbeat")).length, + afterRelease, + "a released turn kept heartbeating: every one of those beats is a zombie", + ); + } finally { + vi.useRealTimers(); + } + }); + + it("release()'s final beat is the turn-end signal: is_running=false under this turn's id", async () => { + // The API keys the whole nest handover on this pair. If the id drifted, the turn-end + // beat would clear a DIFFERENT turn's `running`. + const watchdog = await startAliveWatchdog("sess-end", "turn-end", "proj-1"); + fetchCalls.length = 0; + + await watchdog.release(); + + const beats = fetchCalls.filter((c) => c.url.includes("heartbeat")); + assert.equal(beats.length, 1, "exactly one turn-end beat"); + const body = beats[0].body as Record; + assert.equal(body["is_running"], false); + assert.equal(body["turn_id"], "turn-end"); + assert.equal(body["session_id"], "sess-end"); + }); + it("swallows heartbeat failures — never throws", async () => { fetchShouldFail = true; const watchdog = await startAliveWatchdog("sess-3", "run-fail", "proj-3"); diff --git a/services/runner/tests/unit/session-keepalive-approval.test.ts b/services/runner/tests/unit/session-keepalive-approval.test.ts index b668d14cc3..f185b927e5 100644 --- a/services/runner/tests/unit/session-keepalive-approval.test.ts +++ b/services/runner/tests/unit/session-keepalive-approval.test.ts @@ -1547,6 +1547,8 @@ function pausableHarness( logs: [] as string[], resolvePrompt: undefined as ((value: unknown) => void) | undefined, promptCount: 0, + /** Ordered marks for the settle-before-terminal-record invariant (see the test at the end). */ + journal: [] as string[], }; const captured = { onEvent: undefined as ((event: any) => void) | undefined, @@ -1679,6 +1681,8 @@ function pausableHarness( }, setUsage() {}, finish() { + // The real otel hands the terminal `done` straight to its sink here. + calls.journal.push("done"); return "assistant output"; }, recordError() {}, @@ -2082,6 +2086,187 @@ describe("runTurn: real approval park + respondPermission resume", () => { } }); + it("resolves an in-band answer the harness never re-gated (the orphan fix)", async () => { + // The live orphan: a mobile resume trips `approval-mismatch (history)`, evicts, and runs + // COLD. The transcript already holds the human's answer, so the agent just proceeds and no + // permission request is ever raised — nothing calls the reply path. Meanwhile the turn-start + // stale sweep EXEMPTED this token on the promise that the resume would resolve it, so the + // row is left `pending` forever, actionable in both inboxes. The turn must settle it itself. + const posted: Array<{ url: string; body: Record }> = []; + const fetchSpy = vi + .spyOn(globalThis, "fetch") + .mockImplementation(async (input, init) => { + posted.push({ + url: String(input), + body: JSON.parse(init?.body as string) as Record, + }); + return new Response(JSON.stringify({ ok: true }), { status: 200 }); + }); + + try { + const { calls, deps } = pausableHarness(); + deps.hydrateHarnessSessionFromDurable = async () => {}; + const coldResume: AgentRunRequest = { + harness: "claude", + model: "m1", + sessionId: "sess-orphan", + turnId: "turn-cold-replay", + ...auth, + messages: [ + { role: "user", content: "do X" }, + { + role: "assistant", + content: [ + { type: "tool_call", toolCallId: "tc-gate", toolName: "commit" }, + ], + }, + { + role: "assistant", + content: [ + { + type: "tool_result", + toolCallId: "tc-gate", + toolName: "commit", + output: { approved: true, interactionToken: "tok-orphan" }, + }, + ], + }, + ], + }; + const acquired = await acquireEnvironment(coldResume, deps); + assert.equal(acquired.ok, true); + if (!acquired.ok) return; + const env = acquired.env; + + // No `resume` opts and NO permission request: exactly a cold replay whose agent proceeds. + const turn = runTurn(env, coldResume, undefined, undefined, { + approvalParkMode: true, + }); + await flush(); + calls.resolvePrompt!({ + stopReason: "complete", + usage: { inputTokens: 1, outputTokens: 1 }, + }); + const result = await turn; + assert.equal(result.ok, true); + assert.notEqual(result.stopReason, "paused"); + for ( + let attempt = 0; + attempt < 10 && + !posted.some(({ url }) => + url.endsWith("/sessions/interactions/transition"), + ); + attempt += 1 + ) { + await flush(); + } + + const settled = posted.filter(({ url }) => + url.endsWith("/sessions/interactions/transition"), + ); + assert.equal(settled.length, 1, JSON.stringify(posted)); + // `resolved` with the human's real verdict — NOT `cancelled`, which would record a granted + // approval as abandoned. + assert.deepEqual(settled[0].body, { + session_id: "sess-orphan", + token: "tok-orphan", + status: "resolved", + resolution: { verdict: "approved", tool_call_id: "tc-gate" }, + }); + await env.destroy(); + } finally { + fetchSpy.mockRestore(); + } + }); + + it("does not double-settle a gate the reply path already resolved", async () => { + const posted: Array<{ url: string; body: Record }> = []; + const fetchSpy = vi + .spyOn(globalThis, "fetch") + .mockImplementation(async (input, init) => { + posted.push({ + url: String(input), + body: JSON.parse(init?.body as string) as Record, + }); + return new Response(JSON.stringify({ ok: true }), { status: 200 }); + }); + + try { + const { calls, deps, captured } = pausableHarness(); + deps.hydrateHarnessSessionFromDurable = async () => {}; + const coldResume: AgentRunRequest = { + harness: "claude", + model: "m1", + sessionId: "sess-once", + turnId: "turn-cold-regate", + ...auth, + messages: [ + { role: "user", content: "do X" }, + { + role: "assistant", + content: [ + { + type: "tool_call", + toolCallId: "tc-gate", + toolName: "commit", + input: { message: "hi" }, + }, + ], + }, + { + role: "assistant", + content: [ + { + type: "tool_result", + toolCallId: "tc-gate", + toolName: "commit", + input: { message: "hi" }, + output: { approved: true, interactionToken: "tok-once" }, + }, + ], + }, + ], + }; + const acquired = await acquireEnvironment(coldResume, deps); + assert.equal(acquired.ok, true); + if (!acquired.ok) return; + const env = acquired.env; + + const turn = runTurn(env, coldResume, undefined, undefined, { + approvalParkMode: true, + }); + await flush(); + // The harness DOES re-raise the same call: the stored decision map answers it and the reply + // path resolves the row. The turn-end settle must then be a no-op, not a second write. + captured.onPermissionRequest!({ + id: "perm-regate", + availableReplies: ["once", "reject"], + toolCall: { + toolCallId: "tc-gate", + name: "commit", + rawInput: { message: "hi" }, + }, + }); + await flush(); + calls.resolvePrompt!({ + stopReason: "complete", + usage: { inputTokens: 1, outputTokens: 1 }, + }); + const result = await turn; + assert.equal(result.ok, true); + for (let attempt = 0; attempt < 10; attempt += 1) await flush(); + + const settled = posted.filter(({ url }) => + url.endsWith("/sessions/interactions/transition"), + ); + assert.equal(settled.length, 1, JSON.stringify(settled)); + assert.equal(settled[0].body["token"], "tok-once"); + await env.destroy(); + } finally { + fetchSpy.mockRestore(); + } + }); + it("preserves a denied call failed frame while a sibling gate is carried", async () => { const { calls, deps, captured } = pausableHarness(); const acquired = await acquireEnvironment(engineReq, deps); @@ -3370,3 +3555,73 @@ describe("runTurn: real approval park + respondPermission resume", () => { await env.destroy(); }); }); + +describe("runTurn: the in-band settle lands before the terminal record", () => { + it("resolves the durable row before finish() publishes `done`", async () => { + const TOKEN = "tok-inband-order"; + const TOOL_CALL_ID = "toolu_inband_1"; + const { calls, deps } = pausableHarness(); + + // A cold replay: the transcript already carries the human's yes, so the harness never + // re-raises the gate and only the end-of-turn settle can transition the row. + const request: AgentRunRequest = { + ...engineReq, + telemetry: { + exporters: { otlp: { headers: { authorization: "ApiKey run" } } }, + }, + messages: [ + { role: "user", content: "write out.txt" }, + { + role: "assistant", + content: [ + { + type: "tool_call", + toolCallId: TOOL_CALL_ID, + toolName: "Write", + input: { path: "out.txt" }, + }, + ], + }, + { + role: "assistant", + content: [ + { + type: "tool_result", + toolCallId: TOOL_CALL_ID, + toolName: "Write", + output: { approved: true, interactionToken: TOKEN }, + }, + ], + }, + ], + } as AgentRunRequest; + + const realFetch = globalThis.fetch; + globalThis.fetch = (async (input: any) => { + if (String(input).includes("/sessions/interactions/transition")) { + calls.journal.push("resolve"); + } + return new Response("{}", { status: 200 }); + }) as typeof fetch; + + try { + const acquired = await acquireEnvironment(request, deps); + assert.equal(acquired.ok, true); + if (!acquired.ok) return; + const turn = runTurn(acquired.env, request, undefined, undefined, {}); + await flush(); + calls.resolvePrompt?.({}); + await turn; + await acquired.env.destroy(); + } finally { + globalThis.fetch = realFetch; + } + + assert.deepEqual( + calls.journal, + ["resolve", "done"], + "the API cancels a still-pending gate the moment it consumes `done`; settling after that " + + "files a decision the human actually made as an abandonment", + ); + }); +}); diff --git a/web/mobile/package.json b/web/mobile/package.json index 42ee8268d4..e9b168016c 100644 --- a/web/mobile/package.json +++ b/web/mobile/package.json @@ -37,6 +37,7 @@ "radix-ui": "^1.6.2", "react": "^19.0.0", "react-dom": "^19.0.0", + "streamdown": "^2.5.0", "supertokens-web-js": "^0.16.0", "tailwind-merge": "^3.3.1", "zod": "^4.3.6" diff --git a/web/mobile/src/features/chat/AssistantMarkdown.tsx b/web/mobile/src/features/chat/AssistantMarkdown.tsx new file mode 100644 index 0000000000..b40286ee01 --- /dev/null +++ b/web/mobile/src/features/chat/AssistantMarkdown.tsx @@ -0,0 +1,58 @@ +import {Streamdown, type Components} from "streamdown" + +/** + * Streamdown ships `rehype-raw → rehype-sanitize (GitHub's default schema) → rehype-harden` + * as its default rehype pipeline, so raw HTML in model output is parsed but stripped down to + * the safe subset (no `