From 1c72999fbd72c146a271e8c172de600953f2623c Mon Sep 17 00:00:00 2001 From: Arda Erzin Date: Mon, 27 Jul 2026 23:21:32 +0300 Subject: [PATCH 01/18] feat(api): publish per-session watch events for the M3 live relay MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the watch::session: pub/sub channel to the sessions Redis contract (durable plane) with three change-notification payloads: records-changed, lifecycle {running|ended}, and interaction {pending|resolved}. SessionsWatchPublisher is strictly fire-and-forget — a publish failure is logged and swallowed so the write path that triggered it never fails or re-drives. Publish points (one choke point per event family): - RecordsWorker.process_batch tees records-changed strictly post- append_many, once per distinct session in the project batch. - SessionStreamsService publishes lifecycle running on turn start (send/steer) and ended on cancel/kill/turn-end heartbeat (transition only, not on idle heartbeats). - SessionInteractionsService publishes interaction pending on create and resolved on transition/cancel-sweep. Wired in worker_streams (reusing the durable client), routers, and worker_queues. No behavior change when no watch publisher is injected. --- api/entrypoints/routers.py | 7 + api/entrypoints/worker_queues.py | 8 +- api/entrypoints/worker_streams.py | 4 + .../src/core/sessions/interactions/service.py | 46 +++- api/oss/src/core/sessions/streams/service.py | 51 ++++- api/oss/src/dbs/redis/sessions/contract.py | 39 ++++ api/oss/src/dbs/redis/sessions/locks.py | 24 +++ api/oss/src/dbs/redis/sessions/watch.py | 97 +++++++++ .../tasks/asyncio/sessions/records_worker.py | 25 +++ .../sessions/test_heartbeat_stale_turn_end.py | 155 +++++++++++++ .../test_watch_interactions_publish.py | 131 +++++++++++ .../sessions/test_watch_lifecycle_publish.py | 203 ++++++++++++++++++ .../unit/sessions/test_watch_publish.py | 180 ++++++++++++++++ 13 files changed, 964 insertions(+), 6 deletions(-) create mode 100644 api/oss/src/dbs/redis/sessions/watch.py create mode 100644 api/oss/tests/pytest/unit/sessions/test_heartbeat_stale_turn_end.py create mode 100644 api/oss/tests/pytest/unit/sessions/test_watch_interactions_publish.py create mode 100644 api/oss/tests/pytest/unit/sessions/test_watch_lifecycle_publish.py create mode 100644 api/oss/tests/pytest/unit/sessions/test_watch_publish.py diff --git a/api/entrypoints/routers.py b/api/entrypoints/routers.py index 34cbe49015..70efedf047 100644 --- a/api/entrypoints/routers.py +++ b/api/entrypoints/routers.py @@ -25,6 +25,7 @@ get_cache_engine, get_streams_engine, ) +from oss.src.dbs.redis.sessions.watch import SessionsWatchPublisher from oss.databases.postgres.migrations.core.utils import ( check_for_new_migrations as check_for_new_core_migrations, @@ -642,9 +643,14 @@ async def lifespan(*args, **kwargs): _lock_engine = get_lock_engine() +# M3 live relay: lifecycle/interaction change notifications for the SSE watch +# endpoint, published fire-and-forget on the durable plane. +_sessions_watch_publisher = SessionsWatchPublisher() + session_streams_service = SessionStreamsService( streams_dao=session_streams_dao, lock_engine=_lock_engine, + watch_publisher=_sessions_watch_publisher, ) session_turns_service = SessionTurnsService( @@ -819,6 +825,7 @@ async def lifespan(*args, **kwargs): interactions_service = SessionInteractionsService( interactions_dao=interactions_dao, + watch_publisher=_sessions_watch_publisher, ) triggers_service = TriggersService( diff --git a/api/entrypoints/worker_queues.py b/api/entrypoints/worker_queues.py index a397ca550b..000ca4468b 100644 --- a/api/entrypoints/worker_queues.py +++ b/api/entrypoints/worker_queues.py @@ -69,6 +69,7 @@ ) from oss.src.dbs.postgres.sessions.interactions.dao import SessionInteractionsDAO from oss.src.dbs.postgres.sessions.records.dao import RecordsDAO +from oss.src.dbs.redis.sessions.watch import SessionsWatchPublisher from oss.src.dbs.postgres.shared.engine import ( get_analytics_engine, get_transactions_engine, @@ -218,7 +219,12 @@ def _build_interactions_broker() -> tuple[AsyncBroker, int]: workflows_service.embeds_service = embeds_service environments_service.embeds_service = embeds_service - interactions_service = SessionInteractionsService(interactions_dao=interactions_dao) + interactions_service = SessionInteractionsService( + interactions_dao=interactions_dao, + # M3 live relay: approval resolutions land here (worker process), so this + # composition publishes watch notifications too. + watch_publisher=SessionsWatchPublisher(), + ) # Approval answers replay the session's durable records into the resume conversation; # records live on the analytics engine (same as the API composition in routers.py). records_service = RecordsService( diff --git a/api/entrypoints/worker_streams.py b/api/entrypoints/worker_streams.py index 6031b23ba7..a776d12270 100644 --- a/api/entrypoints/worker_streams.py +++ b/api/entrypoints/worker_streams.py @@ -32,6 +32,7 @@ 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 +from oss.src.dbs.redis.sessions.watch import SessionsWatchPublisher from oss.src.tasks.asyncio.events.worker import EventsWorker from oss.src.tasks.asyncio.sessions.records_worker import RecordsWorker from oss.src.tasks.asyncio.shared.consumer import StreamConsumer @@ -83,6 +84,9 @@ async def _build_records_worker(redis_client: Redis) -> StreamConsumer: redis_client=redis_client, stream_name="streams:records", 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), ) diff --git a/api/oss/src/core/sessions/interactions/service.py b/api/oss/src/core/sessions/interactions/service.py index 16130b252d..e15d098c9b 100644 --- a/api/oss/src/core/sessions/interactions/service.py +++ b/api/oss/src/core/sessions/interactions/service.py @@ -12,11 +12,33 @@ ) from oss.src.core.sessions.interactions.types import InteractionNotFound from oss.src.core.shared.dtos import Windowing +from oss.src.dbs.redis.sessions.contract import ( + WATCH_INTERACTION_PENDING, + WATCH_INTERACTION_RESOLVED, +) +from oss.src.dbs.redis.sessions.watch import SessionsWatchPublisher class SessionInteractionsService: - def __init__(self, *, interactions_dao: SessionInteractionsDAOInterface) -> None: + def __init__( + self, + *, + interactions_dao: SessionInteractionsDAOInterface, + watch_publisher: Optional[SessionsWatchPublisher] = None, + ) -> None: self.interactions_dao = interactions_dao + self._watch = watch_publisher + + async def _publish_interaction( + self, *, project_id: UUID, session_id: str, status: str + ) -> None: + # Fire-and-forget relay notification; the publisher never raises. + if self._watch is not None: + await self._watch.interaction( + project_id=str(project_id), + session_id=session_id, + status=status, + ) async def create_interaction( self, @@ -26,11 +48,17 @@ async def create_interaction( # interaction: SessionInteractionCreate, ) -> SessionInteraction: - return await self.interactions_dao.create_interaction( + created = await self.interactions_dao.create_interaction( project_id=project_id, user_id=user_id, interaction=interaction, ) + await self._publish_interaction( + project_id=project_id, + session_id=interaction.session_id, + status=WATCH_INTERACTION_PENDING, + ) + return created async def fetch_interaction( self, @@ -59,6 +87,11 @@ async def transition_interaction( raise InteractionNotFound( f"Interaction with token {transition.token!r} not found or already terminal" ) + await self._publish_interaction( + project_id=transition.project_id, + session_id=transition.session_id, + status=WATCH_INTERACTION_RESOLVED, + ) return result async def cancel_session_pending( @@ -69,12 +102,19 @@ async def cancel_session_pending( except_turn_id: Optional[str] = None, except_tokens: Optional[List[str]] = None, ) -> int: - return await self.interactions_dao.cancel_session_pending( + 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, ) + if cancelled: + await self._publish_interaction( + project_id=project_id, + session_id=session_id, + status=WATCH_INTERACTION_RESOLVED, + ) + return cancelled async def query_interactions( self, diff --git a/api/oss/src/core/sessions/streams/service.py b/api/oss/src/core/sessions/streams/service.py index 39298d3e32..095b856ceb 100644 --- a/api/oss/src/core/sessions/streams/service.py +++ b/api/oss/src/core/sessions/streams/service.py @@ -21,13 +21,17 @@ from oss.src.dbs.redis.shared.engine import LockEngine from oss.src.dbs.redis.sessions.contract import ( CONCURRENCY_LIMIT, + WATCH_LIFECYCLE_ENDED, + WATCH_LIFECYCLE_RUNNING, validate_session_id as _validate_session_id_fn, ) +from oss.src.dbs.redis.sessions.watch import SessionsWatchPublisher from oss.src.dbs.redis.sessions.locks import ( acquire_alive, acquire_running, claim_owner, clear_running, + release_running, force_cancel_alive, force_clear_owner, get_session_liveness, @@ -74,9 +78,22 @@ def __init__( *, streams_dao: SessionStreamsDAOInterface, lock_engine: LockEngine, + watch_publisher: Optional[SessionsWatchPublisher] = None, ) -> None: self._dao = streams_dao self._lock = lock_engine + self._watch = watch_publisher + + async def _publish_lifecycle( + self, *, project_id: UUID, session_id: str, state: str + ) -> None: + # Fire-and-forget relay notification; the publisher never raises. + if self._watch is not None: + await self._watch.lifecycle( + project_id=str(project_id), + session_id=session_id, + state=state, + ) async def command( self, @@ -149,6 +166,11 @@ async def command( user_id=user_id, session_id=session_id, ) + await self._publish_lifecycle( + project_id=project_id, + session_id=session_id, + state=WATCH_LIFECYCLE_ENDED, + ) return SessionStreamCommandResponse( mode=mode, session_id=session_id, @@ -250,6 +272,11 @@ async def kill( user_id=user_id, session_id=session_id, ) + await self._publish_lifecycle( + project_id=project_id, + session_id=session_id, + state=WATCH_LIFECYCLE_ENDED, + ) return await self._dao.delete_by_session_id( project_id=project_id, session_id=session_id, @@ -337,9 +364,24 @@ async def heartbeat( elif not request.is_running: # Turn ended: drop only `running`. `alive` outlives the turn (own TTL, cleared # only by kill) — this is what makes the session reattachable. - await clear_running( - self._lock, project_id=str(project_id), session_id=request.session_id + # + # Release only what this turn owns. The arming branch above already refuses a + # superseded turn; clearing unconditionally left the mirror hole open, where a + # stale turn's final beat deleted the LIVE turn's lock and published `ended` + # underneath it. + released = await release_running( + self._lock, + project_id=str(project_id), + session_id=request.session_id, + turn_id=request.turn_id, ) + # Publish only on the actual transition, not on every idle heartbeat. + if released: + await self._publish_lifecycle( + project_id=project_id, + session_id=request.session_id, + state=WATCH_LIFECYCLE_ENDED, + ) liveness = await get_session_liveness( self._lock, project_id=str(project_id), session_id=request.session_id @@ -610,6 +652,11 @@ async def _start_turn( user_id=user_id, session_id=session_id, ) + await self._publish_lifecycle( + project_id=project_id, + session_id=session_id, + state=WATCH_LIFECYCLE_RUNNING, + ) return turn_id async def _mirror_flags( diff --git a/api/oss/src/dbs/redis/sessions/contract.py b/api/oss/src/dbs/redis/sessions/contract.py index 6d71eec760..f736f423e3 100644 --- a/api/oss/src/dbs/redis/sessions/contract.py +++ b/api/oss/src/dbs/redis/sessions/contract.py @@ -10,6 +10,7 @@ attached::session: — attach lock (client watching live view) 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) `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: @@ -73,6 +74,44 @@ def make_displacement_payload(*, by: str) -> dict: return {"reason": DISPLACEMENT_REASON_STOLEN, "by": by} +# --------------------------------------------------------------------------- +# Watch channel (M3 live relay) — change notifications, never record payloads. +# Published on the DURABLE Redis plane (the SSE endpoint subscribes there via +# get_streams_engine(); publisher and subscriber must share one plane — the +# displaced channel above lives on the volatile plane instead). +# Payload shapes: +# {"type": "records-changed", "session_id": s} +# {"type": "lifecycle", "session_id": s, "state": "running"|"ended"} +# {"type": "interaction", "session_id": s, "status": "pending"|"resolved"} +# --------------------------------------------------------------------------- + +WATCH_EVENT_RECORDS_CHANGED = "records-changed" +WATCH_EVENT_LIFECYCLE = "lifecycle" +WATCH_EVENT_INTERACTION = "interaction" + +WATCH_LIFECYCLE_RUNNING = "running" +WATCH_LIFECYCLE_ENDED = "ended" + +WATCH_INTERACTION_PENDING = "pending" +WATCH_INTERACTION_RESOLVED = "resolved" + + +def watch_channel(project_id: str, session_id: str) -> str: + return f"watch:{project_id}:session:{session_id}" + + +def make_watch_records_changed_payload(*, session_id: str) -> dict: + return {"type": WATCH_EVENT_RECORDS_CHANGED, "session_id": session_id} + + +def make_watch_lifecycle_payload(*, session_id: str, state: str) -> dict: + return {"type": WATCH_EVENT_LIFECYCLE, "session_id": session_id, "state": state} + + +def make_watch_interaction_payload(*, session_id: str, status: str) -> dict: + return {"type": WATCH_EVENT_INTERACTION, "session_id": session_id, "status": status} + + # --------------------------------------------------------------------------- # Release-if-owner Lua scripts # These are the canonical scripts; both Python and TS implementations must diff --git a/api/oss/src/dbs/redis/sessions/locks.py b/api/oss/src/dbs/redis/sessions/locks.py index 1805c6338e..a979669ef6 100644 --- a/api/oss/src/dbs/redis/sessions/locks.py +++ b/api/oss/src/dbs/redis/sessions/locks.py @@ -145,6 +145,30 @@ async def refresh_running( return False +async def release_running( + engine: LockEngine, + *, + project_id: str, + session_id: str, + turn_id: str, +) -> bool: + """Clear the running lock only if turn_id still owns it. + + The unconditional `clear_running` is right for displacement and the orphan sweep, which + mean to evict whoever holds it. It is wrong for a turn reporting its own end: a stale + turn's final beat would delete the live turn's lock and publish `ended` underneath it. + Atomic, so the owner cannot change between the read and the delete. + """ + key = running_key(project_id, session_id) + result = await engine.eval( + RELEASE_IF_OWNER_LUA, + 1, + key.encode(), + turn_id.encode(), + ) + return result == 1 + + async def clear_running( engine: LockEngine, *, diff --git a/api/oss/src/dbs/redis/sessions/watch.py b/api/oss/src/dbs/redis/sessions/watch.py new file mode 100644 index 0000000000..f893d4207e --- /dev/null +++ b/api/oss/src/dbs/redis/sessions/watch.py @@ -0,0 +1,97 @@ +"""Per-session watch channel — fire-and-forget publishers (M3 live relay). + +Publish side of ``GET /sessions/streams/watch``. Every publish is best-effort: +a failure is logged and swallowed so the write path that triggered it (records +append, turn lifecycle, interaction create/resolve) never fails or re-drives +because of the relay. PUBLISH to zero subscribers is an O(1) Redis no-op. + +Plane: durable Redis — the SSE endpoint subscribes there via +``get_streams_engine()``; publisher and subscriber must share one plane. +""" + +import json +from typing import TYPE_CHECKING, Optional + +from oss.src.dbs.redis.sessions.contract import ( + make_watch_interaction_payload, + make_watch_lifecycle_payload, + make_watch_records_changed_payload, + watch_channel, +) +from oss.src.dbs.redis.shared.engine import get_streams_engine +from oss.src.utils.logging import get_module_logger + +if TYPE_CHECKING: + from redis.asyncio import Redis + +log = get_module_logger(__name__) + + +class SessionsWatchPublisher: + """Publishes watch events on the durable plane. Never raises.""" + + def __init__(self, *, redis_client: Optional["Redis"] = None) -> None: + # An injected client (e.g. the stream worker's durable connection) is + # reused; otherwise the shared streams engine client is opened lazily. + self._redis = redis_client + + def _client(self) -> "Redis": + if self._redis is None: + self._redis = get_streams_engine().get_redis() + return self._redis + + async def _publish( + self, + *, + project_id: str, + session_id: str, + payload: dict, + ) -> None: + try: + await self._client().publish( + watch_channel(project_id, session_id), + json.dumps(payload).encode(), + ) + except Exception: + # Relay only — the write this notifies about is already committed. + log.warning( + "[WATCH] publish failed", + project_id=project_id, + session_id=session_id, + event_type=payload.get("type"), + ) + + async def records_changed(self, *, project_id: str, session_id: str) -> None: + await self._publish( + project_id=project_id, + session_id=session_id, + payload=make_watch_records_changed_payload(session_id=session_id), + ) + + async def lifecycle( + self, + *, + project_id: str, + session_id: str, + state: str, + ) -> None: + await self._publish( + project_id=project_id, + session_id=session_id, + payload=make_watch_lifecycle_payload(session_id=session_id, state=state), + ) + + async def interaction( + self, + *, + project_id: str, + session_id: str, + status: str, + ) -> None: + await self._publish( + project_id=project_id, + session_id=session_id, + payload=make_watch_interaction_payload( + session_id=session_id, status=status + ), + ) diff --git a/api/oss/src/tasks/asyncio/sessions/records_worker.py b/api/oss/src/tasks/asyncio/sessions/records_worker.py index bba0767ca1..9bd9403f47 100644 --- a/api/oss/src/tasks/asyncio/sessions/records_worker.py +++ b/api/oss/src/tasks/asyncio/sessions/records_worker.py @@ -5,6 +5,7 @@ from oss.src.core.sessions.records.service import RecordsService from oss.src.core.sessions.records.streaming import deserialize_record +from oss.src.dbs.redis.sessions.watch import SessionsWatchPublisher from oss.src.utils.common import is_ee from oss.src.utils.logging import get_module_logger from oss.src.tasks.asyncio.shared.consumer import StreamConsumer @@ -45,6 +46,7 @@ def __init__( max_block_ms: int = 5000, max_delay_ms: int = 250, max_batch_mb: int = 50, + watch_publisher: Optional[SessionsWatchPublisher] = None, ): super().__init__( redis_client=redis_client, @@ -57,6 +59,7 @@ def __init__( max_batch_mb=max_batch_mb, ) self.service = service + self.watch_publisher = watch_publisher async def process_batch( self, @@ -156,5 +159,27 @@ async def process_batch( project_id=str(project_batch["project_id"]), exc_info=True, ) + continue + + # 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. + if self.watch_publisher is not None: + project_id = str(project_batch["project_id"]) + session_ids = { + msg.record_event.session_id for msg in project_batch["events"] + } + for session_id in sorted(session_ids): + try: + await self.watch_publisher.records_changed( + project_id=project_id, + session_id=session_id, + ) + except Exception: + log.warning( + "[RECORDS] Watch publish failed", + project_id=project_id, + session_id=session_id, + ) return total_appended, processed_ids diff --git a/api/oss/tests/pytest/unit/sessions/test_heartbeat_stale_turn_end.py b/api/oss/tests/pytest/unit/sessions/test_heartbeat_stale_turn_end.py new file mode 100644 index 0000000000..9a8f80fa45 --- /dev/null +++ b/api/oss/tests/pytest/unit/sessions/test_heartbeat_stale_turn_end.py @@ -0,0 +1,155 @@ +"""A turn ending must clear only ITS OWN `running` lock. + +The arming half of the heartbeat already refuses a turn that no longer owns the session +(`test_heartbeat_parked_zombie.py`): `acquire_running` overwrites, so a superseded turn's beat +would stamp its dead id over the live turn's. + +The clearing half had the mirror hole. `clear_running` deletes unconditionally, so a stale +turn's final `is_running=False` beat deleted whatever id was there — including a LIVE turn's — +and then published `lifecycle: ended` underneath it. Clients watching that session would see it +go idle while it was still running. +""" + +from typing import Optional +from unittest.mock import patch +from uuid import UUID, uuid4 + +import pytest +import pytest_asyncio + +from oss.src.core.sessions.streams.dtos import SessionHeartbeatRequest, SessionStream +from oss.src.core.sessions.streams.service import SessionStreamsService +from oss.src.dbs.redis.sessions.locks import acquire_running, get_running_owner + +from unit.sessions.test_project_scoped_locks import _FakeRedis + + +_PROJECT = uuid4() +_USER = uuid4() +_SESSION = "session_stale_turn_end" + + +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 + + +class _RecordingWatch: + def __init__(self): + self.lifecycle_states = [] + + async def lifecycle(self, *, project_id, session_id, state): + self.lifecycle_states.append(state) + + async def records_changed(self, **kwargs): + return None + + +def _service(lock_engine, watch=None): + return SessionStreamsService( + streams_dao=_FakeStreamsDAO(), + lock_engine=lock_engine, + watch_publisher=watch, + ) + + +@pytest.mark.asyncio +async def test_stale_turn_end_does_not_clear_the_live_turns_running_lock(lock_engine): + live_turn = str(uuid4()) + stale_turn = str(uuid4()) + + await acquire_running( + lock_engine, + project_id=str(_PROJECT), + session_id=_SESSION, + turn_id=live_turn, + ) + + watch = _RecordingWatch() + service = _service(lock_engine, watch) + + # The stale turn reports that IT has ended. + await service.heartbeat( + project_id=_PROJECT, + request=SessionHeartbeatRequest( + session_id=_SESSION, + turn_id=stale_turn, + is_running=False, + replica_id="replica-1", + ), + ) + + owner = await get_running_owner( + lock_engine, project_id=str(_PROJECT), session_id=_SESSION + ) + assert owner == live_turn, "a stale turn's end deleted the live turn's running lock" + assert "ended" not in watch.lifecycle_states, ( + "a stale turn's end published `ended` for a session that is still running" + ) + + +@pytest.mark.asyncio +async def test_a_turn_still_clears_its_own_running_lock(lock_engine): + turn = str(uuid4()) + await acquire_running( + lock_engine, project_id=str(_PROJECT), session_id=_SESSION, turn_id=turn + ) + + watch = _RecordingWatch() + service = _service(lock_engine, watch) + + await service.heartbeat( + project_id=_PROJECT, + request=SessionHeartbeatRequest( + session_id=_SESSION, + turn_id=turn, + is_running=False, + replica_id="replica-1", + ), + ) + + owner = await get_running_owner( + lock_engine, project_id=str(_PROJECT), session_id=_SESSION + ) + assert owner is None + assert watch.lifecycle_states == ["ended"] diff --git a/api/oss/tests/pytest/unit/sessions/test_watch_interactions_publish.py b/api/oss/tests/pytest/unit/sessions/test_watch_interactions_publish.py new file mode 100644 index 0000000000..9835cb9845 --- /dev/null +++ b/api/oss/tests/pytest/unit/sessions/test_watch_interactions_publish.py @@ -0,0 +1,131 @@ +"""M3 live relay — interaction publish points (decision §5-2). + +`SessionInteractionsService` is the single choke point for approval-gate state +(create / transition / cancel-stale fan-outs all go through it), so it owns the +`interaction` watch events: `pending` on create, `resolved` on any transition +away from pending, and one `resolved` when a cancel sweep actually cancelled +something (a no-op sweep publishes nothing). +""" + +from unittest.mock import AsyncMock +from uuid import uuid4 + +import pytest + +from oss.src.core.sessions.interactions.dtos import ( + SessionInteraction, + SessionInteractionCreate, + SessionInteractionKind, + SessionInteractionStatus, + SessionInteractionTransition, +) +from oss.src.core.sessions.interactions.service import SessionInteractionsService + + +_PROJECT = uuid4() + + +class _RecordingPublisher: + def __init__(self): + self.interaction_calls: list[tuple[str, str, str]] = [] + + async def interaction( + self, *, project_id: str, session_id: str, status: str + ) -> None: + self.interaction_calls.append((project_id, session_id, status)) + + +def _interaction(session_id: str) -> SessionInteraction: + return SessionInteraction( + id=uuid4(), + project_id=_PROJECT, + session_id=session_id, + token="tok-1", + kind=SessionInteractionKind.user_approval, + status=SessionInteractionStatus.pending, + ) + + +def _service(dao): + publisher = _RecordingPublisher() + return ( + SessionInteractionsService(interactions_dao=dao, watch_publisher=publisher), + publisher, + ) + + +@pytest.mark.asyncio +async def test_create_publishes_pending(): + dao = AsyncMock() + dao.create_interaction = AsyncMock(return_value=_interaction("sess-1")) + svc, publisher = _service(dao) + + await svc.create_interaction( + project_id=_PROJECT, + interaction=SessionInteractionCreate( + project_id=_PROJECT, + session_id="sess-1", + token="tok-1", + kind=SessionInteractionKind.user_approval, + ), + ) + + assert publisher.interaction_calls == [(str(_PROJECT), "sess-1", "pending")] + + +@pytest.mark.asyncio +async def test_transition_publishes_resolved(): + dao = AsyncMock() + dao.transition_interaction = AsyncMock(return_value=_interaction("sess-1")) + svc, publisher = _service(dao) + + await svc.transition_interaction( + transition=SessionInteractionTransition( + project_id=_PROJECT, + session_id="sess-1", + token="tok-1", + status=SessionInteractionStatus.responded, + ), + ) + + assert publisher.interaction_calls == [(str(_PROJECT), "sess-1", "resolved")] + + +@pytest.mark.asyncio +async def test_failed_transition_publishes_nothing(): + from oss.src.core.sessions.interactions.types import InteractionNotFound + + dao = AsyncMock() + dao.transition_interaction = AsyncMock(return_value=None) + svc, publisher = _service(dao) + + with pytest.raises(InteractionNotFound): + await svc.transition_interaction( + transition=SessionInteractionTransition( + project_id=_PROJECT, + session_id="sess-1", + token="tok-1", + status=SessionInteractionStatus.responded, + ), + ) + + assert publisher.interaction_calls == [] + + +@pytest.mark.asyncio +async def test_cancel_sweep_publishes_resolved_only_when_it_cancelled(): + dao = AsyncMock() + dao.cancel_session_pending = AsyncMock(return_value=2) + svc, publisher = _service(dao) + + cancelled = await svc.cancel_session_pending( + project_id=_PROJECT, session_id="sess-1" + ) + assert cancelled == 2 + assert publisher.interaction_calls == [(str(_PROJECT), "sess-1", "resolved")] + + # No-op sweep: nothing was pending, nothing changed, nothing to notify. + dao.cancel_session_pending = AsyncMock(return_value=0) + publisher.interaction_calls.clear() + await svc.cancel_session_pending(project_id=_PROJECT, session_id="sess-1") + assert publisher.interaction_calls == [] diff --git a/api/oss/tests/pytest/unit/sessions/test_watch_lifecycle_publish.py b/api/oss/tests/pytest/unit/sessions/test_watch_lifecycle_publish.py new file mode 100644 index 0000000000..d6776d978d --- /dev/null +++ b/api/oss/tests/pytest/unit/sessions/test_watch_lifecycle_publish.py @@ -0,0 +1,203 @@ +"""M3 live relay — lifecycle publish points (decision §5-2). + +Turn state transitions are written in exactly one place, `SessionStreamsService` +(send/steer via `_start_turn`, cancel/kill, and the runner's turn-ended +heartbeat), so that service is the single choke point for the `lifecycle` +watch events: `running` when a turn starts, `ended` when it stops. Idle +heartbeats (no running key to clear) must NOT re-publish `ended`. +""" + +from typing import Optional +from unittest.mock import AsyncMock, 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 unit.sessions.test_project_scoped_locks import _FakeRedis + + +_PROJECT = uuid4() +_USER = uuid4() + + +class _RecordingPublisher: + def __init__(self): + self.lifecycle_calls: list[tuple[str, str, str]] = [] + + async def lifecycle(self, *, project_id: str, session_id: str, state: str) -> None: + self.lifecycle_calls.append((project_id, session_id, state)) + + +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): + publisher = _RecordingPublisher() + service = SessionStreamsService( + streams_dao=_FakeStreamsDAO(), + lock_engine=lock_engine, + watch_publisher=publisher, # duck-typed fake; only .lifecycle is used + ) + return service, publisher + + +def _session_id() -> str: + return f"session_{uuid4().hex[:12]}" + + +@pytest.mark.asyncio +async def test_send_publishes_lifecycle_running(lock_engine): + svc, publisher = _service(lock_engine) + session_id = _session_id() + + await svc.command( + project_id=_PROJECT, + user_id=_USER, + request=SessionStreamCommandRequest( + session_id=session_id, + data=WorkflowServiceRequestData(inputs={"messages": ["hi"]}), + force=False, + ), + ) + + assert publisher.lifecycle_calls == [(str(_PROJECT), session_id, "running")] + + +@pytest.mark.asyncio +async def test_steer_publishes_lifecycle_running(lock_engine): + svc, publisher = _service(lock_engine) + session_id = _session_id() + + await svc.command( + project_id=_PROJECT, + user_id=_USER, + request=SessionStreamCommandRequest( + session_id=session_id, + data=WorkflowServiceRequestData(inputs={"messages": ["hi"]}), + force=True, + ), + ) + + assert (str(_PROJECT), session_id, "running") in publisher.lifecycle_calls + + +@pytest.mark.asyncio +async def test_cancel_publishes_lifecycle_ended(lock_engine): + svc, publisher = _service(lock_engine) + session_id = _session_id() + + await svc.command( + project_id=_PROJECT, + user_id=_USER, + request=SessionStreamCommandRequest(session_id=session_id, force=False), + ) + + assert publisher.lifecycle_calls == [(str(_PROJECT), session_id, "ended")] + + +@pytest.mark.asyncio +async def test_kill_publishes_lifecycle_ended(lock_engine): + svc, publisher = _service(lock_engine) + session_id = _session_id() + + with patch( + "oss.src.core.sessions.streams.service.kill_runner_sandbox", + new_callable=AsyncMock, + ): + await svc.kill(project_id=_PROJECT, user_id=_USER, session_id=session_id) + + assert publisher.lifecycle_calls == [(str(_PROJECT), session_id, "ended")] + + +@pytest.mark.asyncio +async def test_turn_end_heartbeat_publishes_ended_once(lock_engine): + svc, publisher = _service(lock_engine) + session_id = _session_id() + + # Start a turn (send), then the runner reports the turn ended. + result = await svc.command( + project_id=_PROJECT, + user_id=_USER, + request=SessionStreamCommandRequest( + session_id=session_id, + data=WorkflowServiceRequestData(inputs={"messages": ["hi"]}), + force=False, + ), + ) + publisher.lifecycle_calls.clear() + + await svc.heartbeat( + project_id=_PROJECT, + request=SessionHeartbeatRequest( + session_id=session_id, + replica_id="replica-1", + turn_id=result.turn_id, + is_running=False, + ), + ) + assert publisher.lifecycle_calls == [(str(_PROJECT), session_id, "ended")] + + # An idle heartbeat (running key already gone) must not publish again. + publisher.lifecycle_calls.clear() + await svc.heartbeat( + project_id=_PROJECT, + request=SessionHeartbeatRequest( + session_id=session_id, + replica_id="replica-1", + turn_id=result.turn_id, + is_running=False, + ), + ) + assert publisher.lifecycle_calls == [] diff --git a/api/oss/tests/pytest/unit/sessions/test_watch_publish.py b/api/oss/tests/pytest/unit/sessions/test_watch_publish.py new file mode 100644 index 0000000000..5c948c7342 --- /dev/null +++ b/api/oss/tests/pytest/unit/sessions/test_watch_publish.py @@ -0,0 +1,180 @@ +"""M3 live relay — publish side (T1). + +The records worker tees a change notification onto the per-session watch channel +strictly AFTER `append_many` commits, once per distinct (project, session) in the +batch. Publishing is fire-and-forget: an append failure publishes nothing (there +is nothing new to see), and a publish failure never fails the worker loop (the +DB write is already committed and must not be re-driven by relay errors). +""" + +import json +import zlib +from unittest.mock import AsyncMock +from uuid import uuid4 + +import pytest +from orjson import dumps + +from oss.src.core.sessions.records.service import RecordsService +from oss.src.core.sessions.records.dtos import SessionRecord +from oss.src.dbs.redis.sessions.contract import watch_channel +from oss.src.dbs.redis.sessions.watch import SessionsWatchPublisher +from oss.src.tasks.asyncio.sessions.records_worker import RecordsWorker + + +def _payload(*, project_id, session_id, record_index=0): + message = { + "organization_id": None, + "project_id": str(project_id), + "record_event": { + "project_id": str(project_id), + "session_id": session_id, + "record_index": record_index, + }, + } + return zlib.compress(dumps(message)) + + +class _RecordingPublisher: + """Watch-publisher fake that records calls; `fail=True` simulates a broken relay.""" + + def __init__(self, *, fail: bool = False, journal=None): + self.calls: list[tuple[str, str]] = [] + self.fail = fail + self.journal = journal + + async def records_changed(self, *, project_id: str, session_id: str) -> None: + if self.fail: + raise RuntimeError("relay down") + self.calls.append((project_id, session_id)) + if self.journal is not None: + self.journal.append(("publish", session_id)) + + +def _worker(records_dao, publisher): + return RecordsWorker( + service=RecordsService(records_dao=records_dao), + redis_client=None, + stream_name="streams:records", + consumer_group="worker-records", + watch_publisher=publisher, + ) + + +@pytest.mark.asyncio +async def test_worker_publishes_once_per_session_after_append(): + project_id = uuid4() + journal: list = [] + + records_dao = AsyncMock() + + async def _append_many(*, events): + journal.append(("append", len(events))) + return [ + SessionRecord( + record_id=uuid4(), session_id=e.session_id, project_id=project_id + ) + for e in events + ] + + records_dao.append_many = AsyncMock(side_effect=_append_many) + + publisher = _RecordingPublisher(journal=journal) + worker = _worker(records_dao, publisher) + + batch = [ + (b"1-0", {b"data": _payload(project_id=project_id, session_id="sess-a")}), + ( + b"2-0", + { + b"data": _payload( + project_id=project_id, session_id="sess-a", record_index=1 + ) + }, + ), + (b"3-0", {b"data": _payload(project_id=project_id, session_id="sess-b")}), + ] + + total_appended, processed_ids = await worker.process_batch(batch) + + assert total_appended == 3 + assert len(processed_ids) == 3 + # 2 distinct sessions -> exactly 2 publishes, never one per record. + assert sorted(publisher.calls) == [ + (str(project_id), "sess-a"), + (str(project_id), "sess-b"), + ] + # Strict ordering: the append committed before any publish fired. + assert journal[0] == ("append", 3) + assert all(entry[0] == "publish" for entry in journal[1:]) + + +@pytest.mark.asyncio +async def test_worker_skips_publish_when_append_fails(): + project_id = uuid4() + + records_dao = AsyncMock() + records_dao.append_many = AsyncMock(side_effect=RuntimeError("db down")) + + publisher = _RecordingPublisher() + worker = _worker(records_dao, publisher) + + batch = [(b"1-0", {b"data": _payload(project_id=project_id, session_id="s")})] + total_appended, processed_ids = await worker.process_batch(batch) + + assert total_appended == 0 + assert publisher.calls == [] + # Failed appends stay in the existing retry path; publish adds nothing to it. + assert len(processed_ids) == 1 + + +@pytest.mark.asyncio +async def test_worker_survives_publisher_failure(): + project_id = uuid4() + + records_dao = AsyncMock() + records_dao.append_many = AsyncMock(return_value=[object()]) + + worker = _worker(records_dao, _RecordingPublisher(fail=True)) + + batch = [(b"1-0", {b"data": _payload(project_id=project_id, session_id="s")})] + total_appended, processed_ids = await worker.process_batch(batch) + + # The committed append is still counted and acked; relay errors never re-drive it. + assert total_appended == 1 + assert len(processed_ids) == 1 + + +@pytest.mark.asyncio +async def test_publisher_publishes_on_watch_channel(): + import fakeredis + + redis = fakeredis.FakeAsyncRedis() + pubsub = redis.pubsub() + project_id = str(uuid4()) + channel = watch_channel(project_id, "sess-1") + await pubsub.subscribe(channel) + await pubsub.get_message(timeout=1) # drain the subscribe confirmation + + publisher = SessionsWatchPublisher(redis_client=redis) + await publisher.records_changed(project_id=project_id, session_id="sess-1") + + message = await pubsub.get_message(ignore_subscribe_messages=True, timeout=1) + assert message is not None + assert json.loads(message["data"]) == { + "type": "records-changed", + "session_id": "sess-1", + } + + +@pytest.mark.asyncio +async def test_publisher_swallows_redis_failure(): + broken = AsyncMock() + broken.publish = AsyncMock(side_effect=ConnectionError("redis gone")) + + publisher = SessionsWatchPublisher(redis_client=broken) + # Must not raise — the relay is strictly best-effort. + await publisher.records_changed(project_id="p", session_id="s") + await publisher.lifecycle(project_id="p", session_id="s", state="running") + await publisher.interaction(project_id="p", session_id="s", status="pending") + assert broken.publish.await_count == 3 From ea2fb0a39ae2aade1c4115ba8921d6c105c025a7 Mon Sep 17 00:00:00 2001 From: Arda Erzin Date: Mon, 27 Jul 2026 23:26:24 +0300 Subject: [PATCH 02/18] feat(api): SSE watch endpoint GET /sessions/streams/watch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bridges one durable-Redis pubsub subscription per connection into a text/event-stream StreamingResponse: named events (records-changed, lifecycle, interaction) with minimal metadata only — clients revalidate via the regular query endpoints; no record payloads ride the wire. Idle windows emit ': heartbeat' comments every 15s (env-tunable via AGENTA_SESSIONS_WATCH_HEARTBEAT_SECONDS) so proxies never idle out. Client disconnect cancels the generator, whose finally block tears down the redis subscription — nothing outlives its SSE connection. Multi- replica correct by construction: events originate in worker/API processes and fan out over shared Redis pub/sub, so any replica can serve the GET (no sticky sessions). Auth rides the existing middleware (cookie/ApiKey/Bearer) evaluated at connect; the handler enforces VIEW_SESSIONS like query_records and validates session_id against the contract allowlist. Spec surface: the route lands in OpenAPI (operation_id watch_session_stream) for documentation, but Fern does not model SSE — clients consume it with a native EventSource (documented in the route docstring); no client regeneration needed. --- api/oss/src/apis/fastapi/sessions/router.py | 75 +++++- api/oss/src/apis/fastapi/sessions/watch.py | 81 ++++++ api/oss/src/utils/env.py | 6 + .../unit/sessions/test_watch_endpoint.py | 234 ++++++++++++++++++ 4 files changed, 395 insertions(+), 1 deletion(-) create mode 100644 api/oss/src/apis/fastapi/sessions/watch.py create mode 100644 api/oss/tests/pytest/unit/sessions/test_watch_endpoint.py diff --git a/api/oss/src/apis/fastapi/sessions/router.py b/api/oss/src/apis/fastapi/sessions/router.py index 1fb8b12b34..e4cd7f9e70 100644 --- a/api/oss/src/apis/fastapi/sessions/router.py +++ b/api/oss/src/apis/fastapi/sessions/router.py @@ -28,16 +28,21 @@ Response, status, ) -from fastapi.responses import JSONResponse +from fastapi.responses import JSONResponse, StreamingResponse # FastAPI route params need fastapi.UploadFile; request.form() yields starlette's base class. from fastapi import UploadFile as FastAPIUploadFile from starlette.datastructures import UploadFile from typing import Any, Optional, Union +from oss.src.utils.env import env from oss.src.utils.exceptions import intercept_exceptions from oss.src.utils.logging import get_module_logger +from oss.src.dbs.redis.sessions.contract import watch_channel +from oss.src.dbs.redis.shared.engine import get_streams_engine +from oss.src.apis.fastapi.sessions.watch import watch_event_stream + from oss.src.core.access.permissions.types import Permission from oss.src.core.access.permissions.service import check_action_access from oss.src.apis.fastapi.shared.exceptions import FORBIDDEN_EXCEPTION @@ -335,6 +340,15 @@ def __init__( tags=["Sessions"], ) + self.router.add_api_route( + "/sessions/streams/watch", + self.watch_session_stream, + methods=["GET"], + operation_id="watch_session_stream", + tags=["Sessions"], + response_model=None, + ) + @intercept_exceptions() @_handle_session_exceptions() async def set_session_stream( @@ -520,6 +534,65 @@ async def set_session_stream_header( ) return SessionStreamResponse(stream=stream) + @intercept_exceptions() + @_handle_session_exceptions() + async def watch_session_stream( + self, + request: Request, + session_id: str = Query(...), + ) -> StreamingResponse: + """Server-sent events relay for one session (M3 live relay). + + Emits change notifications only — never record payloads; clients + revalidate through the regular query endpoints on each event: + + - ``event: records-changed`` — ``{"session_id"}``; new/updated rows + landed in the record log (published post-DB-commit). + - ``event: lifecycle`` — ``{"session_id", "state": "running"|"ended"}``. + - ``event: interaction`` — ``{"session_id", "status": "pending"|"resolved"}``. + - ``: heartbeat`` comment frames while idle (keep-alive). + + Auth is the standard middleware (cookie ``sAccessToken``, ApiKey, or + Bearer) evaluated once at connect; scope is the credential's project. + The stream has no replay/cursor semantics — ``EventSource`` reconnects + and clients revalidate once on every ``open``, which covers any missed + notifications. + + NOTE (spec surface): this route appears in OpenAPI for documentation, + but Fern does not model SSE — consume it with a native ``EventSource`` + (same-origin ``/api`` + cookie auth needs no custom headers), not the + generated client. + """ + _validate_session_id_http(session_id) + project_id = request.state.project_id + user_id = request.state.user_id + + has_permission = await check_action_access( + user_uid=str(user_id), + project_id=str(project_id), + permission=Permission.VIEW_SESSIONS, + ) + if not has_permission: + raise FORBIDDEN_EXCEPTION + + stream = watch_event_stream( + channel=watch_channel(str(project_id), session_id), + # One pubsub connection per SSE connection (v1 — simplest correct + # 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, + ) + return StreamingResponse( + stream, + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "Connection": "keep-alive", + # Disable proxy buffering so frames flush immediately. + "X-Accel-Buffering": "no", + }, + ) + class RecordsRouter: """Records sub-router — /sessions/records/*""" diff --git a/api/oss/src/apis/fastapi/sessions/watch.py b/api/oss/src/apis/fastapi/sessions/watch.py new file mode 100644 index 0000000000..0758b9ef05 --- /dev/null +++ b/api/oss/src/apis/fastapi/sessions/watch.py @@ -0,0 +1,81 @@ +"""SSE frame generator for ``GET /sessions/streams/watch`` (M3 live relay). + +Bridges one Redis pub/sub subscription (durable plane, one per SSE connection) +into `text/event-stream` frames. Events carry TYPE + minimal metadata only — +clients revalidate through their existing query paths; no record payloads ride +the wire. Idle periods emit ``: heartbeat`` comment frames so proxies and +clients never see a silent connection. +""" + +import json +from typing import Any, AsyncIterator, Callable, Optional + +from oss.src.dbs.redis.sessions.contract import ( + WATCH_EVENT_INTERACTION, + WATCH_EVENT_LIFECYCLE, + WATCH_EVENT_RECORDS_CHANGED, +) +from oss.src.utils.logging import get_module_logger + +log = get_module_logger(__name__) + +HEARTBEAT_FRAME = ": heartbeat\n\n" + +_KNOWN_EVENTS = { + WATCH_EVENT_RECORDS_CHANGED, + WATCH_EVENT_LIFECYCLE, + WATCH_EVENT_INTERACTION, +} + + +def format_watch_frame(raw: Any) -> Optional[str]: + """One published payload -> one SSE frame; None for anything malformed/unknown.""" + try: + payload = json.loads(raw) + except (ValueError, TypeError): + return None + if not isinstance(payload, dict): + return None + event = payload.get("type") + # `_KNOWN_EVENTS` is a set, so an unhashable `type` (a list, a dict) would raise + # TypeError out of the membership test and tear down the whole stream. A malformed + # publish must drop its own frame, never the connection. + if not isinstance(event, str) or event not in _KNOWN_EVENTS: + return None + return f"event: {event}\ndata: {json.dumps(payload)}\n\n" + + +async def watch_event_stream( + *, + channel: str, + pubsub_factory: Callable[[], Any], + heartbeat_seconds: float, +) -> AsyncIterator[str]: + """Subscribe to the session's watch channel and yield SSE frames forever. + + 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. + """ + pubsub = pubsub_factory() + try: + await pubsub.subscribe(channel) + while True: + message = await pubsub.get_message( + ignore_subscribe_messages=True, + timeout=heartbeat_seconds, + ) + if message is None: + yield HEARTBEAT_FRAME + continue + if message.get("type") != "message": + continue + frame = format_watch_frame(message.get("data")) + if frame is not None: + yield frame + finally: + try: + await pubsub.unsubscribe(channel) + await pubsub.aclose() + except Exception: # pragma: no cover — teardown is best-effort + log.warning("[WATCH] pubsub teardown failed", channel=channel) diff --git a/api/oss/src/utils/env.py b/api/oss/src/utils/env.py index 397404d032..b9d788f638 100644 --- a/api/oss/src/utils/env.py +++ b/api/oss/src/utils/env.py @@ -1387,6 +1387,12 @@ class SessionsRedisConfig(BaseModel): _parse_optional_positive_int_env("AGENTA_SESSIONS_REDIS_CONCURRENCY_LIMIT") or 1000 ) + # API-side only (SSE watch endpoint keep-alive cadence) — NOT part of the + # runner golden fixture; safe to tune without touching the TS side. + watch_heartbeat_seconds: int = ( + _parse_optional_positive_int_env("AGENTA_SESSIONS_WATCH_HEARTBEAT_SECONDS") + or 15 + ) model_config = ConfigDict(extra="ignore") diff --git a/api/oss/tests/pytest/unit/sessions/test_watch_endpoint.py b/api/oss/tests/pytest/unit/sessions/test_watch_endpoint.py new file mode 100644 index 0000000000..2589416214 --- /dev/null +++ b/api/oss/tests/pytest/unit/sessions/test_watch_endpoint.py @@ -0,0 +1,234 @@ +"""M3 live relay — SSE watch endpoint (T2). + +`GET /sessions/streams/watch` bridges one durable-Redis pubsub subscription into +`text/event-stream` frames: known payloads become named SSE events, idle windows +become `: heartbeat` comments, and closing the generator (client disconnect) +tears the subscription down. RBAC mirrors `query_records` (VIEW_SESSIONS). +""" + +import asyncio +import json +from unittest.mock import AsyncMock, patch +from uuid import uuid4 + +import pytest +from fastapi import FastAPI, HTTPException, Request + +from oss.src.apis.fastapi.sessions.router import SessionStreamsRouter +from oss.src.apis.fastapi.sessions.watch import ( + HEARTBEAT_FRAME, + format_watch_frame, + watch_event_stream, +) +from oss.src.dbs.redis.sessions.contract import watch_channel +from oss.src.dbs.redis.sessions.watch import SessionsWatchPublisher + + +class _FakePubSub: + """Scripted pubsub: returns queued messages, then None (idle) forever.""" + + def __init__(self, messages): + self.messages = list(messages) + self.subscribed: list[str] = [] + self.unsubscribed: list[str] = [] + self.closed = False + + async def subscribe(self, channel): + self.subscribed.append(channel) + + async def get_message(self, *, ignore_subscribe_messages=False, timeout=None): + if self.messages: + return self.messages.pop(0) + return None + + async def unsubscribe(self, channel): + self.unsubscribed.append(channel) + + async def aclose(self): + self.closed = True + + +def _msg(payload: dict) -> dict: + return {"type": "message", "data": json.dumps(payload).encode()} + + +@pytest.mark.asyncio +async def test_stream_yields_event_frames_then_heartbeats(): + pubsub = _FakePubSub( + [ + _msg({"type": "records-changed", "session_id": "s1"}), + _msg({"type": "lifecycle", "session_id": "s1", "state": "running"}), + _msg({"type": "interaction", "session_id": "s1", "status": "pending"}), + ] + ) + stream = watch_event_stream( + channel="watch:p:session:s1", + pubsub_factory=lambda: pubsub, + heartbeat_seconds=0.01, + ) + + frames = [] + async for frame in stream: + frames.append(frame) + if len(frames) == 4: + await stream.aclose() + break + + assert frames[0].startswith("event: records-changed\n") + assert json.loads(frames[0].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") + # Queue drained -> the idle path emits keep-alive comments. + assert frames[3] == HEARTBEAT_FRAME + assert pubsub.subscribed == ["watch:p:session:s1"] + + +@pytest.mark.asyncio +async def test_stream_cleans_up_subscription_on_close(): + pubsub = _FakePubSub([]) + stream = watch_event_stream( + channel="watch:p:session:s1", + pubsub_factory=lambda: pubsub, + heartbeat_seconds=0.01, + ) + # Take one heartbeat, then simulate the client disconnecting. + frame = await stream.__anext__() + assert frame == HEARTBEAT_FRAME + await stream.aclose() + + assert pubsub.unsubscribed == ["watch:p:session:s1"] + assert pubsub.closed is True + + +@pytest.mark.asyncio +async def test_stream_skips_malformed_and_unknown_payloads(): + pubsub = _FakePubSub( + [ + {"type": "message", "data": b"not json"}, + _msg({"type": "unknown-event", "session_id": "s1"}), + {"type": "subscribe", "data": 1}, + _msg({"type": "records-changed", "session_id": "s1"}), + ] + ) + stream = watch_event_stream( + channel="watch:p:session:s1", + pubsub_factory=lambda: pubsub, + heartbeat_seconds=0.01, + ) + 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") + + +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 + assert format_watch_frame(json.dumps({"type": "nope"}).encode()) is None + frame = format_watch_frame( + json.dumps({"type": "lifecycle", "session_id": "s", "state": "ended"}).encode() + ) + assert frame == ( + 'event: lifecycle\ndata: {"type": "lifecycle", "session_id": "s", "state": "ended"}\n\n' + ) + + +@pytest.mark.asyncio +async def test_stream_delivers_publisher_events_end_to_end(): + """Publisher (T1) -> fakeredis pub/sub -> SSE generator (T2), one plane.""" + import fakeredis + + redis = fakeredis.FakeAsyncRedis() + project_id = str(uuid4()) + channel = watch_channel(project_id, "sess-e2e") + + stream = watch_event_stream( + channel=channel, + pubsub_factory=lambda: redis.pubsub(), + heartbeat_seconds=0.05, + ) + # First frame is a heartbeat — proves the subscription is live before publishing. + assert await stream.__anext__() == HEARTBEAT_FRAME + + publisher = SessionsWatchPublisher(redis_client=redis) + await publisher.records_changed(project_id=project_id, session_id="sess-e2e") + + async def _next_event_frame(): + while True: + frame = await stream.__anext__() + if frame != HEARTBEAT_FRAME: + return frame + + frame = await asyncio.wait_for(_next_event_frame(), timeout=2) + await stream.aclose() + assert frame.startswith("event: records-changed\n") + assert json.loads(frame.split("data: ")[1])["session_id"] == "sess-e2e" + + +def _make_authed_request(app: FastAPI, project_id, user_id) -> Request: + scope = { + "type": "http", + "method": "GET", + "path": "/sessions/streams/watch", + "headers": [], + "app": app, + } + request = Request(scope) + request.state.project_id = str(project_id) + request.state.user_id = str(user_id) + return request + + +def _router() -> SessionStreamsRouter: + return SessionStreamsRouter( + service=AsyncMock(), + interactions_service=AsyncMock(), + ) + + +@pytest.mark.asyncio +async def test_watch_endpoint_rejects_without_view_sessions(): + router = _router() + request = _make_authed_request(FastAPI(), uuid4(), uuid4()) + + with patch( + "oss.src.apis.fastapi.sessions.router.check_action_access", + new_callable=AsyncMock, + return_value=False, + ): + with pytest.raises(HTTPException) as exc_info: + await router.watch_session_stream(request=request, session_id="sess-1") + + assert exc_info.value.status_code == 403 + + +@pytest.mark.asyncio +async def test_watch_endpoint_rejects_invalid_session_id(): + router = _router() + request = _make_authed_request(FastAPI(), uuid4(), uuid4()) + + with pytest.raises(HTTPException) as exc_info: + await router.watch_session_stream(request=request, session_id="bad/../id") + + assert exc_info.value.status_code == 400 + + +@pytest.mark.asyncio +async def test_watch_endpoint_returns_event_stream_response(): + router = _router() + request = _make_authed_request(FastAPI(), uuid4(), uuid4()) + + with patch( + "oss.src.apis.fastapi.sessions.router.check_action_access", + new_callable=AsyncMock, + return_value=True, + ): + response = await router.watch_session_stream(request=request, session_id="s-1") + + assert response.media_type == "text/event-stream" + assert response.headers["cache-control"] == "no-cache" + assert response.headers["x-accel-buffering"] == "no" From f7cf351423f562833213a4e3b5a8c968f3261b04 Mon Sep 17 00:00:00 2001 From: Arda Erzin Date: Mon, 27 Jul 2026 23:30:27 +0300 Subject: [PATCH 03/18] fix(mobile): mute the user bubble and fetch the session title via the query path --- web/mobile/src/features/chat/ChatHeader.tsx | 6 ++++-- web/mobile/src/features/chat/TurnRow.tsx | 6 ++++-- 2 files changed, 8 insertions(+), 4 deletions(-) diff --git a/web/mobile/src/features/chat/ChatHeader.tsx b/web/mobile/src/features/chat/ChatHeader.tsx index 85ebef2439..3c642764ab 100644 --- a/web/mobile/src/features/chat/ChatHeader.tsx +++ b/web/mobile/src/features/chat/ChatHeader.tsx @@ -1,4 +1,4 @@ -import {fetchSessionStream} from "@agenta/entities/session" +import {querySessionStreams} from "@agenta/entities/session" import {useQuery} from "@tanstack/react-query" import Link from "next/link" @@ -11,9 +11,11 @@ export const ChatHeader = ({ projectId: string workspaceId: string }) => { + // The singular GET /sessions/streams redirects with a root-path-less Location + // behind the /api prefix and lands on the web app — use the proven query POST. const query = useQuery({ queryKey: ["mobile", "session-stream", projectId, sessionId], - queryFn: () => fetchSessionStream({sessionId, projectId}), + queryFn: async () => (await querySessionStreams({sessionId, projectId}))?.[0] ?? null, enabled: Boolean(projectId && sessionId), staleTime: 30_000, }) diff --git a/web/mobile/src/features/chat/TurnRow.tsx b/web/mobile/src/features/chat/TurnRow.tsx index e45e648a23..c69b89425e 100644 --- a/web/mobile/src/features/chat/TurnRow.tsx +++ b/web/mobile/src/features/chat/TurnRow.tsx @@ -24,8 +24,10 @@ export const TurnRow = ({ return (

{item.part.text} From 1c1267664fd165364ddba308ff4fb17d06b50113 Mon Sep 17 00:00:00 2001 From: Arda Erzin Date: Mon, 27 Jul 2026 23:33:36 +0300 Subject: [PATCH 04/18] feat(mobile): consume the session live relay via useSessionWatch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit One EventSource per foregrounded chat screen against /sessions/streams/watch (same-origin /api, cookie auth — native EventSource, not the Fern client, which does not model SSE): - records-changed (and every open, as missed-event coverage) drives the transcript tick's exact body — useSessionTranscript now exposes it as a stable refresh() (revalidateSessionRecordsAtom + loadSessionMessages re-read through the shared cache). - lifecycle / interaction events invalidate the shared liveness and actionable-interactions queries instead of duplicating state; their 15s polls stay as the documented fallback. - Foreground-only: closed on visibilitychange->hidden, reopened on visible. Transient errors ride EventSource's built-in reconnect; a fatal CLOSED falls back to today's 4s/7.5s cadence and retries every 60s while visible. While the stream is open the ChatScreen records tick stretches to a 30s safety net (watchAwarePollMs); when the stream is down the cadence is byte-identical to before — the fallback IS the current behavior. The sessions LIST screen keeps its 15s polls in v1 per the plan. vitest config gains the tsconfig '@/' alias so tested modules resolve app imports. --- web/mobile/src/features/chat/ChatScreen.tsx | 10 +- .../src/features/chat/useSessionTranscript.ts | 53 +++++---- .../src/features/chat/useSessionWatch.ts | 111 ++++++++++++++++++ web/mobile/src/features/chat/watchRelay.ts | 15 +++ web/mobile/tests/unit/watchRelay.test.ts | 44 +++++++ web/mobile/vitest.config.ts | 6 + 6 files changed, 213 insertions(+), 26 deletions(-) create mode 100644 web/mobile/src/features/chat/useSessionWatch.ts create mode 100644 web/mobile/src/features/chat/watchRelay.ts create mode 100644 web/mobile/tests/unit/watchRelay.test.ts diff --git a/web/mobile/src/features/chat/ChatScreen.tsx b/web/mobile/src/features/chat/ChatScreen.tsx index 20901c2a14..d3b925128d 100644 --- a/web/mobile/src/features/chat/ChatScreen.tsx +++ b/web/mobile/src/features/chat/ChatScreen.tsx @@ -14,7 +14,9 @@ import {StopButton} from "./StopButton" import {TurnRow} from "./TurnRow" import {useApprovalActions} from "./useApprovalActions" import {useSessionTranscript} from "./useSessionTranscript" +import {useSessionWatch} from "./useSessionWatch" import {useTranscriptAutoScroll} from "./useTranscriptAutoScroll" +import {watchAwarePollMs} from "./watchRelay" /** Read-only replay screen — mount it with `key={sessionId}` so per-session state resets. */ export const ChatScreen = ({ @@ -29,7 +31,10 @@ export const ChatScreen = ({ // Tightened records cadence only while this foregrounded screen shows a running or pending // turn; derived from the previous render's messages, so it settles one render behind. const [pollMs, setPollMs] = useState(0) - const {messages, state} = useSessionTranscript(sessionId, pollMs) + const {messages, state, refresh} = useSessionTranscript(sessionId, pollMs) + // Live relay (M3): push-invalidate through the same tick body; while it is open the + // poll below is only a safety net. + const watch = useSessionWatch({sessionId, projectId, onRecordsChanged: refresh}) const liveness = useLivenessPoll(projectId) const running = Boolean( liveness.data?.find((s) => s.session_id === sessionId)?.flags?.is_running, @@ -37,8 +42,9 @@ export const ChatScreen = ({ const pendingCount = useMemo(() => getPendingApprovals(messages).length, [messages]) const approvals = useApprovalActions({sessionId, projectId, pendingCount}) // ~4s while a fired decision settles (fire-and-forget — records carry the resume). - const nextPollMs = + const basePollMs = approvals.phase === "resuming" ? 4_000 : pendingCount > 0 || running ? 7_500 : 0 + const nextPollMs = watchAwarePollMs(basePollMs, watch.connected) if (nextPollMs !== pollMs) setPollMs(nextPollMs) // One identity cache per session mount (the screen is keyed by sessionId). // eslint-disable-next-line react-hooks/exhaustive-deps diff --git a/web/mobile/src/features/chat/useSessionTranscript.ts b/web/mobile/src/features/chat/useSessionTranscript.ts index d565a9589e..ca828a0187 100644 --- a/web/mobile/src/features/chat/useSessionTranscript.ts +++ b/web/mobile/src/features/chat/useSessionTranscript.ts @@ -1,4 +1,4 @@ -import {useEffect, useState} from "react" +import {useCallback, useEffect, useRef, useState} from "react" import {loadSessionMessages} from "@agenta/chat/assets" import {revalidateSessionRecordsAtom} from "@agenta/entities/session" @@ -13,10 +13,18 @@ import {getDefaultStore} from "jotai" * `pollMs` > 0 tightens the cadence (a running turn / pending approval): each tick marks the * records stale and re-reads through the shared cache. Foreground-only — a hidden tab skips * ticks entirely (the records query is the heavy one; see the plan's cost note). + * + * The returned `refresh` is the tick's body as a stable callback, so the live relay + * (`useSessionWatch`) can drive the exact same revalidate path push-style. */ export const useSessionTranscript = (sessionId: string, pollMs = 0) => { const [messages, setMessages] = useState([]) const [state, setState] = useState<"loading" | "ready" | "empty">("loading") + // Session-switch guard: a late resolve for a previous session must never land. + const sessionRef = useRef(sessionId) + sessionRef.current = sessionId + const inFlightRef = useRef(false) + useEffect(() => { let cancelled = false let refreshed = false @@ -39,33 +47,30 @@ export const useSessionTranscript = (sessionId: string, pollMs = 0) => { } }, [sessionId]) + const refresh = useCallback(() => { + if (document.visibilityState !== "visible" || inFlightRef.current) return + inFlightRef.current = true + // Invalidate first so the shared-cache read refetches instead of serving staleTime. + getDefaultStore().set(revalidateSessionRecordsAtom, sessionId) + void loadSessionMessages(sessionId) + .then((msgs) => { + if (sessionRef.current === sessionId && msgs && msgs.length > 0) { + setMessages(msgs) + setState("ready") + } + }) + .finally(() => { + inFlightRef.current = false + }) + }, [sessionId]) + useEffect(() => { if (!pollMs) return - let cancelled = false - let inFlight = false - const store = getDefaultStore() - const tick = () => { - if (document.visibilityState !== "visible" || inFlight) return - inFlight = true - // Invalidate first so the shared-cache read refetches instead of serving staleTime. - store.set(revalidateSessionRecordsAtom, sessionId) - void loadSessionMessages(sessionId) - .then((msgs) => { - if (!cancelled && msgs && msgs.length > 0) { - setMessages(msgs) - setState("ready") - } - }) - .finally(() => { - inFlight = false - }) - } - const handle = setInterval(tick, pollMs) + const handle = setInterval(refresh, pollMs) return () => { - cancelled = true clearInterval(handle) } - }, [sessionId, pollMs]) + }, [refresh, pollMs]) - return {messages, state} + return {messages, state, refresh} } diff --git a/web/mobile/src/features/chat/useSessionWatch.ts b/web/mobile/src/features/chat/useSessionWatch.ts new file mode 100644 index 0000000000..72ca608eed --- /dev/null +++ b/web/mobile/src/features/chat/useSessionWatch.ts @@ -0,0 +1,111 @@ +import {useEffect, useRef, useState} from "react" + +import {useQueryClient} from "@tanstack/react-query" + +import {actionableInteractionsQueryKey} from "../sessions/useActionableInteractions" +import {livenessQueryKey} from "../sessions/useLivenessPoll" + +import {sessionWatchUrl} from "./watchRelay" + +/** Fatal-close retry cadence (EventSource does NOT auto-retry from CLOSED). */ +const RETRY_MS = 60_000 + +/** + * One EventSource per foregrounded chat screen (M3 live relay). Events carry no payloads — + * every handler funnels into the existing revalidate paths: + * + * - `records-changed` (and every `open`, for missed-event coverage) → `onRecordsChanged`, + * i.e. the transcript tick's body (`revalidateSessionRecordsAtom` + re-read). + * - `lifecycle` / `interaction` → invalidate the shared liveness + actionable-interactions + * queries (no duplicated state; the badges' own queries refetch). + * + * Foreground-only: the source closes on `visibilitychange → hidden` and reopens on visible. + * Transient errors ride EventSource's built-in reconnect; a fatal CLOSED (endpoint missing, + * proxy reset) drops to the callers' poll cadence and retries every 60s while visible. + */ +export const useSessionWatch = ({ + sessionId, + projectId, + onRecordsChanged, +}: { + sessionId: string + projectId: string + onRecordsChanged: () => void +}): {connected: boolean} => { + const [connected, setConnected] = useState(false) + const queryClient = useQueryClient() + const onRecordsChangedRef = useRef(onRecordsChanged) + onRecordsChangedRef.current = onRecordsChanged + + useEffect(() => { + if (!sessionId || !projectId) return + if (typeof window === "undefined" || typeof window.EventSource === "undefined") return + + let source: EventSource | null = null + let retryHandle: number | undefined + let disposed = false + + const invalidateBadges = () => { + void queryClient.invalidateQueries({queryKey: livenessQueryKey(projectId)}) + void queryClient.invalidateQueries({ + queryKey: actionableInteractionsQueryKey(projectId), + }) + } + + const close = () => { + source?.close() + source = null + setConnected(false) + } + + const scheduleRetry = () => { + if (disposed || retryHandle !== undefined) return + retryHandle = window.setTimeout(() => { + retryHandle = undefined + open() + }, RETRY_MS) + } + + const open = () => { + if (disposed || source !== null || document.visibilityState !== "visible") return + const es = new EventSource(sessionWatchUrl(sessionId, projectId), { + withCredentials: true, + }) + source = es + es.onopen = () => { + setConnected(true) + // Missed-event coverage: one revalidation per (re)connect replaces + // any replay/cursor semantics on the server. + onRecordsChangedRef.current() + invalidateBadges() + } + es.addEventListener("records-changed", () => onRecordsChangedRef.current()) + es.addEventListener("lifecycle", invalidateBadges) + es.addEventListener("interaction", invalidateBadges) + es.onerror = () => { + setConnected(false) + // CONNECTING = built-in auto-reconnect; only a fatal CLOSED needs us. + if (es.readyState === EventSource.CLOSED) { + close() + scheduleRetry() + } + } + } + + const onVisibility = () => { + if (document.visibilityState === "visible") open() + else close() + } + + document.addEventListener("visibilitychange", onVisibility) + open() + return () => { + disposed = true + document.removeEventListener("visibilitychange", onVisibility) + if (retryHandle !== undefined) window.clearTimeout(retryHandle) + close() + } + }, [sessionId, projectId, queryClient]) + + return {connected} +} diff --git a/web/mobile/src/features/chat/watchRelay.ts b/web/mobile/src/features/chat/watchRelay.ts new file mode 100644 index 0000000000..50571df0f6 --- /dev/null +++ b/web/mobile/src/features/chat/watchRelay.ts @@ -0,0 +1,15 @@ +import {getApiUrl} from "@/lib/env" + +/** Watch endpoint URL — same-origin `/api` + cookie auth, consumed via native EventSource. */ +export const sessionWatchUrl = (sessionId: string, projectId: string): string => + `${getApiUrl()}/sessions/streams/watch?session_id=${encodeURIComponent( + sessionId, + )}&project_id=${encodeURIComponent(projectId)}` + +/** + * Records-tick cadence under the relay: while the EventSource is open the tick is only a + * safety net (30s); on error/close the caller's base cadence (today's 4s/7.5s/idle-0) + * stands unchanged — the fallback IS the current behavior. An idle 0 never wakes up. + */ +export const watchAwarePollMs = (baseMs: number, connected: boolean): number => + connected && baseMs > 0 ? 30_000 : baseMs diff --git a/web/mobile/tests/unit/watchRelay.test.ts b/web/mobile/tests/unit/watchRelay.test.ts new file mode 100644 index 0000000000..dcc7390eef --- /dev/null +++ b/web/mobile/tests/unit/watchRelay.test.ts @@ -0,0 +1,44 @@ +import {afterEach, describe, expect, it} from "vitest" + +import {sessionWatchUrl, watchAwarePollMs} from "../../src/features/chat/watchRelay" + +const ENV_KEY = "NEXT_PUBLIC_AGENTA_API_URL" +const priorEnv = process.env[ENV_KEY] + +afterEach(() => { + if (priorEnv === undefined) delete process.env[ENV_KEY] + else process.env[ENV_KEY] = priorEnv +}) + +describe("sessionWatchUrl", () => { + it("targets the watch endpoint with encoded session and project ids", () => { + process.env[ENV_KEY] = "http://localhost/api" + expect(sessionWatchUrl("sess-1", "proj-1")).toBe( + "http://localhost/api/sessions/streams/watch?session_id=sess-1&project_id=proj-1", + ) + }) + + it("URL-encodes hostile ids instead of letting them extend the query", () => { + process.env[ENV_KEY] = "http://localhost/api" + expect(sessionWatchUrl("a&b=c", "p")).toBe( + "http://localhost/api/sessions/streams/watch?session_id=a%26b%3Dc&project_id=p", + ) + }) +}) + +describe("watchAwarePollMs", () => { + it("stretches an active cadence to the 30s safety net while the stream is open", () => { + expect(watchAwarePollMs(7_500, true)).toBe(30_000) + expect(watchAwarePollMs(4_000, true)).toBe(30_000) + }) + + it("keeps today's cadence untouched when the stream is down (no-regression fallback)", () => { + expect(watchAwarePollMs(7_500, false)).toBe(7_500) + expect(watchAwarePollMs(4_000, false)).toBe(4_000) + expect(watchAwarePollMs(0, false)).toBe(0) + }) + + it("never wakes an idle screen just because the stream is open", () => { + expect(watchAwarePollMs(0, true)).toBe(0) + }) +}) diff --git a/web/mobile/vitest.config.ts b/web/mobile/vitest.config.ts index 064e19c375..0d09cd460a 100644 --- a/web/mobile/vitest.config.ts +++ b/web/mobile/vitest.config.ts @@ -1,6 +1,12 @@ +import {fileURLToPath} from "node:url" + import {defineConfig} from "vitest/config" export default defineConfig({ + resolve: { + // Mirror tsconfig's `@/*` -> `src/*` so tested modules resolve app imports. + alias: {"@": fileURLToPath(new URL("./src", import.meta.url))}, + }, test: { include: ["tests/unit/**/*.test.ts"], environment: "node", From 32cf66444bafce913e5ef44d0bb705f5bef95900 Mon Sep 17 00:00:00 2001 From: Arda Erzin Date: Mon, 27 Jul 2026 23:52:31 +0300 Subject: [PATCH 05/18] fix(mobile): queue trailing refreshes and bound the watch publish --- api/oss/src/dbs/redis/sessions/watch.py | 13 ++++++++++--- .../src/features/chat/useSessionTranscript.ts | 14 +++++++++++++- 2 files changed, 23 insertions(+), 4 deletions(-) diff --git a/api/oss/src/dbs/redis/sessions/watch.py b/api/oss/src/dbs/redis/sessions/watch.py index f893d4207e..947cd5ea84 100644 --- a/api/oss/src/dbs/redis/sessions/watch.py +++ b/api/oss/src/dbs/redis/sessions/watch.py @@ -9,6 +9,7 @@ ``get_streams_engine()``; publisher and subscriber must share one plane. """ +import asyncio import json from typing import TYPE_CHECKING, Optional @@ -48,9 +49,15 @@ async def _publish( payload: dict, ) -> None: try: - await self._client().publish( - watch_channel(project_id, session_id), - json.dumps(payload).encode(), + # Bounded: the streams client has no socket timeouts, and these publishes + # sit on turn-lifecycle/interaction write paths — a black-holed Redis must + # cost at most 1s, never a TCP timeout. + await asyncio.wait_for( + self._client().publish( + watch_channel(project_id, session_id), + json.dumps(payload).encode(), + ), + timeout=1.0, ) except Exception: # Relay only — the write this notifies about is already committed. diff --git a/web/mobile/src/features/chat/useSessionTranscript.ts b/web/mobile/src/features/chat/useSessionTranscript.ts index ca828a0187..7af13f5473 100644 --- a/web/mobile/src/features/chat/useSessionTranscript.ts +++ b/web/mobile/src/features/chat/useSessionTranscript.ts @@ -24,6 +24,10 @@ export const useSessionTranscript = (sessionId: string, pollMs = 0) => { const sessionRef = useRef(sessionId) sessionRef.current = sessionId const inFlightRef = useRef(false) + // A change event that lands mid-refresh must queue a trailing refresh — dropping it + // can strand the FINAL transcript state forever (the turn's `ended` also kills the + // tightened poll, so nothing else would ever re-read). + const pendingRef = useRef(false) useEffect(() => { let cancelled = false @@ -48,7 +52,11 @@ export const useSessionTranscript = (sessionId: string, pollMs = 0) => { }, [sessionId]) const refresh = useCallback(() => { - if (document.visibilityState !== "visible" || inFlightRef.current) return + if (document.visibilityState !== "visible") return + if (inFlightRef.current) { + pendingRef.current = true + return + } inFlightRef.current = true // Invalidate first so the shared-cache read refetches instead of serving staleTime. getDefaultStore().set(revalidateSessionRecordsAtom, sessionId) @@ -61,6 +69,10 @@ export const useSessionTranscript = (sessionId: string, pollMs = 0) => { }) .finally(() => { inFlightRef.current = false + if (pendingRef.current) { + pendingRef.current = false + if (sessionRef.current === sessionId) refresh() + } }) }, [sessionId]) From 341bbf95fad61b155431e9c8eda1b8ec818dbef6 Mon Sep 17 00:00:00 2001 From: Arda Erzin Date: Tue, 28 Jul 2026 11:42:47 +0300 Subject: [PATCH 06/18] fix(mobile): drain the resume stream instead of cancelling it Cancelling the response body aborted the invoke; the agent service treated the disconnect as a stop, so the resumed run died ~200ms in and the gate stayed pending. Also stop badging rows from the lagging flags mirror. --- .../src/features/chat/useApprovalActions.ts | 19 ++++++++++++++++--- .../src/features/sessions/SessionRow.tsx | 5 +++-- 2 files changed, 19 insertions(+), 5 deletions(-) diff --git a/web/mobile/src/features/chat/useApprovalActions.ts b/web/mobile/src/features/chat/useApprovalActions.ts index 33cce6f987..cafded1a1c 100644 --- a/web/mobile/src/features/chat/useApprovalActions.ts +++ b/web/mobile/src/features/chat/useApprovalActions.ts @@ -142,9 +142,22 @@ export const useApprovalActions = ({ if (!response.ok) { throw new Error(`Resume failed (HTTP ${response.status}).`) } - // Fire-and-forget: release the stream immediately — session runs survive - // client disconnect, and holding the SSE open for the whole turn is waste. - void response.body?.cancel().catch(() => undefined) + // Fire-and-forget, but NEVER cancel: cancelling the body aborts the request, + // and the agent service treats that disconnect as "stop" — the resumed run + // dies ~200ms in and the gate stays pending (observed live). Drain instead. + void (async () => { + const reader = response.body?.getReader() + if (!reader) return + try { + for (;;) { + const {done} = await reader.read() + if (done) return + } + } catch { + // Connection dropped (screen locked, network change) — the run + // continues server-side; records polling picks the result up. + } + })() } catch (err) { setPhase("error") setErrorText(err instanceof Error ? err.message : "Resume failed.") diff --git a/web/mobile/src/features/sessions/SessionRow.tsx b/web/mobile/src/features/sessions/SessionRow.tsx index 4d0736301f..994a1b4157 100644 --- a/web/mobile/src/features/sessions/SessionRow.tsx +++ b/web/mobile/src/features/sessions/SessionRow.tsx @@ -33,8 +33,9 @@ export const SessionRow = ({ }) => { const agentLabel = session.references?.[0]?.slug ?? session.references?.[0]?.id ?? "—" const activity = timeAgo(session.updated_at ?? session.created_at) - const badge = - liveness === undefined ? (session.flags?.is_alive ? "alive" : null) : (liveness ?? null) + // Row flags are a lagging mirror of the Redis nest and go stale (rows sit at + // is_running=true for days after a crashed run) — only the live poll may badge. + const badge = liveness ?? null return ( From 383ce75a731389b90e694c38279822b51d381410 Mon Sep 17 00:00:00 2001 From: Arda Erzin Date: Tue, 28 Jul 2026 11:46:24 +0300 Subject: [PATCH 07/18] feat(mobile): move approvals into a bottom dock --- web/mobile/src/features/chat/ApprovalCard.tsx | 69 --------------- web/mobile/src/features/chat/ApprovalDock.tsx | 85 +++++++++++++++++++ web/mobile/src/features/chat/ChatScreen.tsx | 12 ++- web/mobile/src/features/chat/TurnRow.tsx | 29 ++----- .../src/features/chat/approvalInputSummary.ts | 41 +++++++++ 5 files changed, 137 insertions(+), 99 deletions(-) delete mode 100644 web/mobile/src/features/chat/ApprovalCard.tsx create mode 100644 web/mobile/src/features/chat/ApprovalDock.tsx create mode 100644 web/mobile/src/features/chat/approvalInputSummary.ts diff --git a/web/mobile/src/features/chat/ApprovalCard.tsx b/web/mobile/src/features/chat/ApprovalCard.tsx deleted file mode 100644 index 3f81b1dfa2..0000000000 --- a/web/mobile/src/features/chat/ApprovalCard.tsx +++ /dev/null @@ -1,69 +0,0 @@ -import type {ApprovalActions} from "./useApprovalActions" - -/** - * Raw highlighted pending-approval block: tool name + exact payload + Approve/Deny (and - * Approve-all when several gates are pending). Without `actions` it degrades to the read-only - * M0 card ("answer on desktop"). Raw UI on purpose — flows over polish. - */ -export const ApprovalCard = ({ - toolName, - input, - approvalId, - pendingCount = 0, - actions, -}: { - toolName: string - input: unknown - /** The gate's interaction id off the tool part (`approval.id`). */ - approvalId?: string - /** Gates pending on the paused turn — >1 surfaces the Approve-all button. */ - pendingCount?: number - actions?: ApprovalActions -}) => { - const actionable = Boolean(actions && approvalId) - const busy = actions?.phase === "resuming" - const disabled = !actionable || busy - return ( -

-

Approval pending — {toolName}

- 
-                {JSON.stringify(input, null, 2)}
-
-
- - - {actionable && pendingCount > 1 ? ( - - ) : null} -
- {busy ?

Resuming…

: null} - {actions?.phase === "error" && actions.errorText ? ( -

{actions.errorText}

- ) : null} - {!actionable ? ( -

Answer on desktop for now.

- ) : null} -
- ) -} diff --git a/web/mobile/src/features/chat/ApprovalDock.tsx b/web/mobile/src/features/chat/ApprovalDock.tsx new file mode 100644 index 0000000000..afc5c63e17 --- /dev/null +++ b/web/mobile/src/features/chat/ApprovalDock.tsx @@ -0,0 +1,85 @@ +import type {PendingApproval} from "@agenta/chat/model" + +import {summarizeApprovalInput} from "./approvalInputSummary" +import type {ApprovalActions} from "./useApprovalActions" + +/** + * Bottom-anchored human-in-the-loop dock — the mobile shape of the desktop ApprovalDock. + * It sits outside the transcript scroller (a shrink-0 sibling at the end of the screen's + * flex column) so a paused run can never scroll out of reach, and it owns the decision: + * the inline tool row is only an "Awaiting approval" marker. A turn can request several + * gates at once — we act on the first and surface the count, with Approve all for the batch. + */ +export const ApprovalDock = ({ + approvals, + actions, +}: { + /** Pending gates for the paused turn (index 0 is acted on first). */ + approvals: PendingApproval[] + actions: ApprovalActions +}) => { + const current = approvals[0] + if (!current) return null + const count = approvals.length + const busy = actions.phase === "resuming" + const summary = summarizeApprovalInput(current.input) + return ( +
+
+

Approval needed to continue

+ {count > 1 ? ( + + {count} pending + + ) : null} +
+

+ {current.toolName} +

+ {summary.text ? ( +
+

{summary.label}

+ 
+                        {summary.text}
+
+
+ ) : null} +
+ {count > 1 ? ( + + ) : null} + + +
+ {busy ?

Resuming…

: null} + {actions.phase === "error" && actions.errorText ? ( +

{actions.errorText}

+ ) : null} +
+ ) +} diff --git a/web/mobile/src/features/chat/ChatScreen.tsx b/web/mobile/src/features/chat/ChatScreen.tsx index d3b925128d..2710d1f8bf 100644 --- a/web/mobile/src/features/chat/ChatScreen.tsx +++ b/web/mobile/src/features/chat/ChatScreen.tsx @@ -8,6 +8,7 @@ import { import {useLivenessPoll} from "../sessions/useLivenessPoll" +import {ApprovalDock} from "./ApprovalDock" import {ChatHeader} from "./ChatHeader" import {ChatEmpty, ChatLoading} from "./states/ChatStates" import {StopButton} from "./StopButton" @@ -39,7 +40,8 @@ export const ChatScreen = ({ const running = Boolean( liveness.data?.find((s) => s.session_id === sessionId)?.flags?.is_running, ) - const pendingCount = useMemo(() => getPendingApprovals(messages).length, [messages]) + const pendingApprovals = useMemo(() => getPendingApprovals(messages), [messages]) + const pendingCount = pendingApprovals.length const approvals = useApprovalActions({sessionId, projectId, pendingCount}) // ~4s while a fired decision settles (fire-and-forget — records carry the resume). const basePollMs = @@ -67,12 +69,7 @@ export const ChatScreen = ({ {turns .filter((turn) => !turn.hidden) .map((turn) => ( - + ))} ) @@ -94,6 +91,7 @@ export const ChatScreen = ({ > {body} + ) } diff --git a/web/mobile/src/features/chat/TurnRow.tsx b/web/mobile/src/features/chat/TurnRow.tsx index c69b89425e..3891fe8ec5 100644 --- a/web/mobile/src/features/chat/TurnRow.tsx +++ b/web/mobile/src/features/chat/TurnRow.tsx @@ -1,19 +1,7 @@ import {partToolName, rowSummary, type TurnViewModel} from "@agenta/chat/model" -import {ApprovalCard} from "./ApprovalCard" -import type {ApprovalActions} from "./useApprovalActions" - /** One transcript turn: raw aligned text parts, one-line tool summaries, raw error line. */ -export const TurnRow = ({ - turn, - approvalActions, - pendingApprovals = 0, -}: { - turn: TurnViewModel - /** Resume actions for pending-approval cards (absent = read-only cards). */ - approvalActions?: ApprovalActions - pendingApprovals?: number -}) => ( +export const TurnRow = ({turn}: {turn: TurnViewModel}) => (
{ const key = part.toolCallId ?? `${item.index}-${i}` if (part.state === "approval-requested") { - const approvalId = (part as {approval?: {id?: string}}).approval - ?.id + // The decision lives in the bottom ApprovalDock; the row is + // just the marker (desktop parity). return ( - +

+ Awaiting approval — {partToolName(part)} +

) } const summary = rowSummary(part) diff --git a/web/mobile/src/features/chat/approvalInputSummary.ts b/web/mobile/src/features/chat/approvalInputSummary.ts new file mode 100644 index 0000000000..6827b22ff2 --- /dev/null +++ b/web/mobile/src/features/chat/approvalInputSummary.ts @@ -0,0 +1,41 @@ +/** The readable part of a gate's payload: `{label, text}`, empty text when there is nothing to show. */ +export interface ApprovalInputSummary { + /** The field the text came from ("command", "query", …) or "Input" for the JSON fallback. */ + label: string + text: string +} + +// Fields tools use for their one human-readable argument — a bash gate must read as the command +// itself, not as a JSON blob. Order is priority order. +const PRIMARY_FIELDS = [ + "command", + "cmd", + "script", + "query", + "sql", + "url", + "path", + "file_path", + "prompt", + "message", + "content", + "text", +] + +/** Pick the payload's primary field when it has one; otherwise fall back to compact JSON. */ +export const summarizeApprovalInput = (input: unknown): ApprovalInputSummary => { + if (input == null) return {label: "Input", text: ""} + if (typeof input === "string") return {label: "Input", text: input} + if (typeof input !== "object") return {label: "Input", text: String(input)} + const record = input as Record + for (const field of PRIMARY_FIELDS) { + const value = record[field] + if (typeof value === "string" && value.trim()) return {label: field, text: value} + } + try { + const json = JSON.stringify(input) + return {label: "Input", text: json === "{}" ? "" : json} + } catch { + return {label: "Input", text: String(input)} + } +} From 9852b60e451b3d9405c7f75ab140904e8667730e Mon Sep 17 00:00:00 2001 From: Arda Erzin Date: Tue, 28 Jul 2026 14:21:49 +0300 Subject: [PATCH 08/18] fix(api): write the session liveness mirror on every heartbeat MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `heartbeat` seeded its result from the row it had just read, which skipped both the create and the update branch whenever a `session_streams` row already existed. The runner's heartbeat is the only producer of `flags` for a real turn (nothing calls the send/steer command path), so the column stayed NULL on every session that had been named first, and froze at `is_running: true` on rows the first beat had created — the turn-end beat could not write either. `updated_at` stayed NULL too, which is the column the orphan sweep filters on. The mirror write is now unconditional: create on first touch, else update. That makes the row track the CURRENT turn, which the `is_current_turn` disambiguation was not written for: an old turn's beat now finds a different turn_id on the row and would read as establishment. Settle the takeover case on the signal that actually means it — a failed nx re-acquire of the alive lock, which only fails while another turn holds it. --- api/oss/src/core/sessions/streams/service.py | 28 ++-- .../test_heartbeat_is_current_turn.py | 25 +++ .../test_heartbeat_mirrors_existing_row.py | 153 ++++++++++++++++++ 3 files changed, 197 insertions(+), 9 deletions(-) create mode 100644 api/oss/tests/pytest/unit/sessions/test_heartbeat_mirrors_existing_row.py diff --git a/api/oss/src/core/sessions/streams/service.py b/api/oss/src/core/sessions/streams/service.py index 095b856ceb..a2a4b6e05c 100644 --- a/api/oss/src/core/sessions/streams/service.py +++ b/api/oss/src/core/sessions/streams/service.py @@ -319,8 +319,9 @@ async def heartbeat( # `_start_turn` acquire may not have landed yet, or this beat wins a race with it), and # that is NOT an interruption. Disambiguate with the durable row's `turn_id`: if it # already recorded THIS turn_id as established (a prior heartbeat's write), the key - # being gone now is something else's doing; if the row shows no turn yet, or a - # different one, this is establishment. + # being gone now is something else's doing. A row carrying a DIFFERENT turn_id is not + # decisive on its own (it is also what a fresh turn on a previously-run session sees) + # — the failed nx re-acquire below is what settles that case. prior_stream = await self._dao.get_by_session_id( project_id=project_id, session_id=request.session_id, @@ -333,20 +334,25 @@ async def heartbeat( if request.turn_id and request.is_running: # 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 alive acquire is the unambiguous takeover signal: nx only fails when a + # different turn holds the key right now. A successful one means the key was + # merely absent, which is an interruption only if this turn had already + # established it (`turn_was_established`) rather than establishing it here. + # (`acquire_running` is not nx — it overwrites — so it carries no such signal.) if not await refresh_alive( self._lock, project_id=str(project_id), session_id=request.session_id, turn_id=request.turn_id, ): - if turn_was_established: - is_current_turn = False - await acquire_alive( + 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( self._lock, project_id=str(project_id), @@ -392,11 +398,14 @@ async def heartbeat( is_attached=liveness["attached"], ) - # Nothing between `prior_stream`'s fetch above and here mutates the row, so it is - # still the current read — no need to re-fetch. - stream = prior_stream + # The mirror write is unconditional: create when this beat is the row's first touch, + # otherwise UPDATE. Seeding `stream` from `prior_stream` here would skip both branches + # for every session whose row already exists (renamed, or created by an earlier beat), + # leaving `flags` NULL on named sessions and frozen at is_running=true on the rest — + # and `updated_at` NULL, which is the column the orphan sweep filters on. + stream: Optional[SessionStream] = None - if stream is None: + if prior_stream is None: try: stream = await self._dao.create( project_id=project_id, @@ -418,6 +427,7 @@ async def heartbeat( session_id=request.session_id, stream=SessionStreamEdit(flags=flags, turn_id=request.turn_id), ) + return SessionHeartbeatResult( stream=stream, replica_id=owner, diff --git a/api/oss/tests/pytest/unit/sessions/test_heartbeat_is_current_turn.py b/api/oss/tests/pytest/unit/sessions/test_heartbeat_is_current_turn.py index 4f72336691..fa3595d73c 100644 --- a/api/oss/tests/pytest/unit/sessions/test_heartbeat_is_current_turn.py +++ b/api/oss/tests/pytest/unit/sessions/test_heartbeat_is_current_turn.py @@ -175,3 +175,28 @@ async def test_losing_owner_claim_reports_not_current(lock_engine): assert result.is_current_turn is False assert result.replica_id == "replica-a" + + +@pytest.mark.asyncio +async def test_new_turn_on_a_previously_run_session_is_current(lock_engine): + """The row records the LATEST turn, so a fresh turn always finds a different turn_id on + it. That alone must not read as a takeover: after the previous turn ended and its alive + lock lapsed, the new turn is simply establishing the nest — exactly the state a brand-new + turn is in. Only a lock still held by another turn (the failed nx acquire) means takeover. + """ + svc = _service(lock_engine) + + await svc.heartbeat(project_id=_PROJECT, request=_beat("replica-a", "turn-1")) + await svc.heartbeat( + project_id=_PROJECT, request=_beat("replica-a", "turn-1", running=False) + ) + # The previous turn's alive lock lapses (TTL) / is cleared before the next turn starts. + await force_cancel_alive(lock_engine, project_id=str(_PROJECT), session_id=_SESSION) + + fresh = await svc.heartbeat( + project_id=_PROJECT, request=_beat("replica-a", "turn-2") + ) + + assert fresh.is_current_turn is True, ( + "a new turn must not be aborted just because the row still named the old one" + ) diff --git a/api/oss/tests/pytest/unit/sessions/test_heartbeat_mirrors_existing_row.py b/api/oss/tests/pytest/unit/sessions/test_heartbeat_mirrors_existing_row.py new file mode 100644 index 0000000000..d9152a5421 --- /dev/null +++ b/api/oss/tests/pytest/unit/sessions/test_heartbeat_mirrors_existing_row.py @@ -0,0 +1,153 @@ +"""The heartbeat must mirror the Redis nest onto a row that ALREADY exists. + +The runner is the only component that reports liveness for a real turn (it mints its own +turn id and calls `POST /sessions/streams/heartbeat`; nothing invokes the send/steer command +path), so the heartbeat's row write is the sole producer of `session_streams.flags` — +the column the project-wide liveness query (`is_alive=true`) and the concurrency cap read. + +Seeding the result from the pre-read row skipped BOTH the create and the update branch +whenever a row already existed, which is every session that was named first (rename creates +the row) or that had beat once before. Symptoms, all from this one omission: + - a genuinely running named session kept `flags = NULL` -> no badge anywhere; + - a row created BY a first beat froze at is_running=true, because the turn-end beat + (is_running=false) could not write either -> a phantom "running" forever; + - `updated_at` stayed NULL, so the orphan sweep (which filters on it) never reclaimed them. +""" + +from typing import Optional +from unittest.mock import patch +from uuid import UUID, uuid4 + +import pytest +import pytest_asyncio + +from oss.src.core.sessions.streams.dtos import ( + SessionHeartbeatRequest, + SessionStream, + SessionStreamFlags, +) +from oss.src.core.sessions.streams.service import SessionStreamsService + +from unit.sessions.test_project_scoped_locks import _FakeRedis + + +_PROJECT = uuid4() +_SESSION = "session_named_first" + + +class _FakeStreamsDAO: + """One in-memory row; records how it was written.""" + + def __init__(self, existing: Optional[SessionStream] = None): + self.row = existing + self.creates = 0 + self.updates = 0 + + 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.creates += 1 + self.row = SessionStream( + id=uuid4(), + project_id=project_id, + session_id=stream.session_id, + name=stream.name, + flags=stream.flags, + turn_id=stream.turn_id, + ) + return self.row + + async def update(self, *, project_id, user_id, session_id, stream): + self.updates += 1 + prior = self.row + self.row = SessionStream( + id=prior.id if prior else uuid4(), + project_id=project_id, + session_id=session_id, + name=prior.name if prior else None, + flags=stream.flags if stream.flags is not None else prior.flags, + turn_id=stream.turn_id if stream.turn_id is not None else prior.turn_id, + ) + return self.row + + +@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 _named_row() -> SessionStream: + """A row created by the rename edit alone: header only, no flags, no turn.""" + return SessionStream( + id=uuid4(), + project_id=_PROJECT, + session_id=_SESSION, + name="Mobile Test", + ) + + +def _beat(turn_id: str, *, is_running: bool = True) -> SessionHeartbeatRequest: + return SessionHeartbeatRequest( + session_id=_SESSION, + replica_id="replica-a", + turn_id=turn_id, + is_running=is_running, + ) + + +@pytest.mark.asyncio +async def test_heartbeat_writes_flags_onto_a_pre_existing_row(lock_engine): + dao = _FakeStreamsDAO(_named_row()) + svc = SessionStreamsService(streams_dao=dao, lock_engine=lock_engine) + + result = await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-1")) + + assert dao.creates == 0, "the row exists; the beat must UPDATE, never re-create" + assert dao.updates == 1, "an existing row must still be mirrored" + assert result.stream is not None + assert result.stream.flags == SessionStreamFlags( + is_alive=True, is_running=True, is_attached=False + ) + assert result.stream.turn_id == "turn-1" + assert result.stream.name == "Mobile Test", ( + "the mirror write must not clobber the name" + ) + + +@pytest.mark.asyncio +async def test_turn_end_heartbeat_clears_running_on_the_row(lock_engine): + dao = _FakeStreamsDAO(_named_row()) + svc = SessionStreamsService(streams_dao=dao, lock_engine=lock_engine) + + await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-1")) + result = await svc.heartbeat( + project_id=_PROJECT, request=_beat("turn-1", is_running=False) + ) + + assert result.stream is not None + # `alive` outlives the turn (that is what makes a session reattachable); only + # `running` collapses — the phantom-running badge is what this guards. + assert result.stream.flags.is_running is False + assert result.stream.flags.is_alive is True + + +@pytest.mark.asyncio +async def test_first_beat_creates_then_later_beats_update(lock_engine): + dao = _FakeStreamsDAO() + svc = SessionStreamsService(streams_dao=dao, lock_engine=lock_engine) + + await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-1")) + assert (dao.creates, dao.updates) == (1, 0) + + result = await svc.heartbeat( + project_id=_PROJECT, request=_beat("turn-1", is_running=False) + ) + assert (dao.creates, dao.updates) == (1, 1), ( + "the turn-end beat must reach the row it created" + ) + assert result.stream.flags.is_running is False From 3a3c35811e0ef8f68cfd564a167f60c4f9d57e6c Mon Sep 17 00:00:00 2001 From: Arda Erzin Date: Tue, 28 Jul 2026 14:22:00 +0300 Subject: [PATCH 09/18] fix(api): let the orphan sweep reclaim rows never updated since creation The sweep selected on `updated_at < threshold`, and `NULL < threshold` is NULL, so a row whose heartbeat never wrote it could never be reclaimed however long it had claimed to be alive. Locally that left ~1.2k rows stuck at is_running=true for days: invisible to the sweep, counted by the per-project concurrency cap, and returned by the project-wide liveness query as phantom "running" sessions. Select on coalesce(updated_at, created_at), and bound the pass so an accumulated backlog drains over successive sweeps instead of one oversized commit. --- .../tasks/asyncio/sessions/orphan_sweep.py | 21 +++++++++---- .../test_orphan_sweep_clears_redis.py | 30 +++++++++++++++++-- 2 files changed, 43 insertions(+), 8 deletions(-) diff --git a/api/oss/src/tasks/asyncio/sessions/orphan_sweep.py b/api/oss/src/tasks/asyncio/sessions/orphan_sweep.py index 9cdf728783..ef929d8517 100644 --- a/api/oss/src/tasks/asyncio/sessions/orphan_sweep.py +++ b/api/oss/src/tasks/asyncio/sessions/orphan_sweep.py @@ -24,7 +24,7 @@ force_clear_owner, ) -from sqlalchemy import select +from sqlalchemy import func, select log = get_module_logger(__name__) @@ -34,16 +34,27 @@ # How often the sweep runs. SWEEP_INTERVAL_SECONDS: int = 60 +# Rows swept per pass. A backlog drains over successive passes instead of one huge commit. +SWEEP_BATCH_SIZE: int = 500 + async def run_orphan_sweep(engine: TransactionsEngine, lock_engine: LockEngine) -> None: """Single sweep pass: mark stale is_alive rows as ended.""" threshold = datetime.now(timezone.utc) - timedelta(seconds=ORPHAN_THRESHOLD_SECONDS) async with engine.session() as session: - stmt = select(SessionStreamDBE).where( - SessionStreamDBE.deleted_at.is_(None), - SessionStreamDBE.flags.contains({"is_alive": True}), - SessionStreamDBE.updated_at < threshold, + stmt = ( + select(SessionStreamDBE) + .where( + SessionStreamDBE.deleted_at.is_(None), + SessionStreamDBE.flags.contains({"is_alive": True}), + # coalesce, not a bare `updated_at <`: a row never updated since creation has + # updated_at NULL, and `NULL < threshold` is NULL — such a row could never be + # swept, however long it had claimed to be alive. + func.coalesce(SessionStreamDBE.updated_at, SessionStreamDBE.created_at) + < threshold, + ) + .limit(SWEEP_BATCH_SIZE) ) result = await session.execute(stmt) orphans = result.scalars().all() 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 9004831ed5..3035d3080d 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 @@ -48,10 +48,12 @@ def scalars(self): class _FakePgSession: - def __init__(self, rows): + def __init__(self, rows, seen): self._rows = rows + self._seen = seen - async def execute(self, _stmt): + async def execute(self, stmt): + self._seen.append(stmt) return _FakeResult(self._rows) async def commit(self): @@ -63,10 +65,11 @@ class _FakeTransactionsEngine: def __init__(self, rows): self._rows = rows + self.statements = [] @asynccontextmanager async def session(self): - yield _FakePgSession(self._rows) + yield _FakePgSession(self._rows, self.statements) class _FakeRedis: @@ -145,3 +148,24 @@ def _send_gate(liveness): raise SessionTurnInUse(session_id=_SESSION_ID, liveness=liveness) _send_gate(liveness_after) # must not raise + + +@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` + is NULL — so a bare `updated_at <` predicate can never reclaim it, however long it has + claimed to be alive. The sweep must compare on coalesce(updated_at, created_at), and + cap the pass so a large backlog drains over several passes. + """ + assert anyio_backend == "asyncio" + + pg_engine = _FakeTransactionsEngine([]) + await run_orphan_sweep(pg_engine, _FakeRedis()) + + assert pg_engine.statements, "the sweep must issue its select" + sql = str( + pg_engine.statements[0].compile(compile_kwargs={"literal_binds": False}) + ).lower() + assert "coalesce" in sql, "a NULL updated_at must fall back to created_at" + assert "session_streams.created_at" in sql + assert "limit" in sql, "one pass must be bounded" From 3a0d352f148fa82db5a60c9ee6d52c63d8bf9b5e Mon Sep 17 00:00:00 2001 From: Arda Erzin Date: Tue, 28 Jul 2026 14:24:40 +0300 Subject: [PATCH 10/18] feat(api): publish the `running` lifecycle event from the runner heartbeat MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `_start_turn` was the only publisher of the M3 relay's `running` lifecycle event, and it is reachable only through the send/steer command path — which no shipped client calls. A real turn goes straight to the agent invoke, and the runner mints its own turn id and only heartbeats, so watchers saw `records-changed` and `interaction` events but never `running`; `ended` already rode the turn-end beat. Publish it from the heartbeat on the transition into a turn the row has not yet recorded, so a ~30s beat does not re-announce it and a send-started turn (whose row `_start_turn` already stamped) does not announce twice. --- api/oss/src/core/sessions/streams/service.py | 11 +++ .../sessions/test_watch_lifecycle_publish.py | 75 +++++++++++++++++++ 2 files changed, 86 insertions(+) diff --git a/api/oss/src/core/sessions/streams/service.py b/api/oss/src/core/sessions/streams/service.py index a2a4b6e05c..080f875a84 100644 --- a/api/oss/src/core/sessions/streams/service.py +++ b/api/oss/src/core/sessions/streams/service.py @@ -428,6 +428,17 @@ async def heartbeat( stream=SessionStreamEdit(flags=flags, turn_id=request.turn_id), ) + # `running` lifecycle for the path that actually runs turns. `_start_turn` publishes it + # for send/steer, but the runner mints its own turn id and only ever heartbeats, so + # without this the relay's `running` event never fires for a real run. Gated on the + # turn being new to this row, so a 30s-interval beat doesn't re-publish it. + if request.turn_id and request.is_running and not turn_was_established: + await self._publish_lifecycle( + project_id=project_id, + session_id=request.session_id, + state=WATCH_LIFECYCLE_RUNNING, + ) + return SessionHeartbeatResult( stream=stream, replica_id=owner, diff --git a/api/oss/tests/pytest/unit/sessions/test_watch_lifecycle_publish.py b/api/oss/tests/pytest/unit/sessions/test_watch_lifecycle_publish.py index d6776d978d..51caf48561 100644 --- a/api/oss/tests/pytest/unit/sessions/test_watch_lifecycle_publish.py +++ b/api/oss/tests/pytest/unit/sessions/test_watch_lifecycle_publish.py @@ -201,3 +201,78 @@ async def test_turn_end_heartbeat_publishes_ended_once(lock_engine): ), ) assert publisher.lifecycle_calls == [] + + +@pytest.mark.asyncio +async def test_first_beat_of_a_runner_minted_turn_publishes_running(lock_engine): + """The path real runs take: no send/steer command, so `_start_turn` never fires — the + runner mints its own turn id and only heartbeats. Without a publish here the relay's + `running` lifecycle event never fires for any actual turn.""" + svc, publisher = _service(lock_engine) + session_id = _session_id() + + await svc.heartbeat( + project_id=_PROJECT, + request=SessionHeartbeatRequest( + session_id=session_id, + replica_id="replica-1", + turn_id="runner-turn-1", + is_running=True, + ), + ) + assert publisher.lifecycle_calls == [(str(_PROJECT), session_id, "running")] + + # Every ~30s beat of the SAME turn must stay silent; only the transition publishes. + publisher.lifecycle_calls.clear() + await svc.heartbeat( + project_id=_PROJECT, + request=SessionHeartbeatRequest( + session_id=session_id, + replica_id="replica-1", + turn_id="runner-turn-1", + is_running=True, + ), + ) + assert publisher.lifecycle_calls == [] + + # A new turn on the same session is a fresh transition. + await svc.heartbeat( + project_id=_PROJECT, + request=SessionHeartbeatRequest( + session_id=session_id, + replica_id="replica-1", + turn_id="runner-turn-2", + is_running=True, + ), + ) + assert publisher.lifecycle_calls == [(str(_PROJECT), session_id, "running")] + + +@pytest.mark.asyncio +async def test_send_does_not_double_publish_running_on_its_first_beat(lock_engine): + """`_start_turn` already recorded the turn on the row, so the runner's first beat for + that same turn must not emit a second `running`.""" + svc, publisher = _service(lock_engine) + session_id = _session_id() + + result = await svc.command( + project_id=_PROJECT, + user_id=_USER, + request=SessionStreamCommandRequest( + session_id=session_id, + data=WorkflowServiceRequestData(inputs={"messages": ["hi"]}), + force=False, + ), + ) + publisher.lifecycle_calls.clear() + + await svc.heartbeat( + project_id=_PROJECT, + request=SessionHeartbeatRequest( + session_id=session_id, + replica_id="replica-1", + turn_id=result.turn_id, + is_running=True, + ), + ) + assert publisher.lifecycle_calls == [] From b50b8db403e5b043ea312901ceaf062ee0961f31 Mon Sep 17 00:00:00 2001 From: Arda Erzin Date: Tue, 28 Jul 2026 14:25:00 +0300 Subject: [PATCH 11/18] test(mobile): cover the session-list liveness badge mapping The list badge now rests entirely on the project-wide liveness poll (the row's own flags were dropped as a lagging mirror), so the poll-to-badge mapping is the only thing between a truthful backend and a truthful row. Pin its three states: running, alive-but-idle (a turn parked awaiting approval), and unresolved. --- web/mobile/tests/unit/livenessBadge.test.ts | 41 +++++++++++++++++++++ 1 file changed, 41 insertions(+) create mode 100644 web/mobile/tests/unit/livenessBadge.test.ts diff --git a/web/mobile/tests/unit/livenessBadge.test.ts b/web/mobile/tests/unit/livenessBadge.test.ts new file mode 100644 index 0000000000..b0b6baa444 --- /dev/null +++ b/web/mobile/tests/unit/livenessBadge.test.ts @@ -0,0 +1,41 @@ +import type {SessionStream} from "@agenta/entities/session" +import {describe, expect, it} from "vitest" + +import {livenessBySession} from "../../src/features/sessions/useLivenessPoll" + +const stream = (session_id: string, flags?: SessionStream["flags"]) => + ({session_id, flags}) as SessionStream + +describe("livenessBySession", () => { + it("badges a running session from the poll's flags", () => { + const map = livenessBySession([ + stream("s-running", {is_alive: true, is_running: true, is_attached: false}), + ]) + expect(map?.get("s-running")).toBe("running") + }) + + it("badges an alive-but-idle session (parked awaiting approval) as live, not running", () => { + // A parked turn has ended, so `running` collapses while `alive` outlives it — that + // pairing is exactly what the backend heartbeat mirror writes at turn end. + const map = livenessBySession([ + stream("s-parked", {is_alive: true, is_running: false, is_attached: false}), + ]) + expect(map?.get("s-parked")).toBe("alive") + }) + + it("omits sessions the poll did not return, so a reclaimed row loses its badge", () => { + const map = livenessBySession([stream("s-alive", {is_alive: true, is_running: true})]) + expect(map?.has("s-swept")).toBe(false) + expect(map?.get("s-swept")).toBeUndefined() + }) + + it("stays undefined until the poll resolves, so rows render unbadged rather than wrong", () => { + expect(livenessBySession(undefined)).toBeUndefined() + expect(livenessBySession(null)).toBeUndefined() + }) + + it("treats a flagless row as not running", () => { + const map = livenessBySession([stream("s-no-flags")]) + expect(map?.get("s-no-flags")).toBe("alive") + }) +}) From 5ed70a896f6091e8457e81b9ebe25bb5baa94802 Mon Sep 17 00:00:00 2001 From: Arda Erzin Date: Tue, 28 Jul 2026 17:01:00 +0300 Subject: [PATCH 12/18] fix(api): treat a stale alive lock as handover, not turn takeover A failed nx acquire is not by itself a takeover: alive outlives its turn, so every follow-up turn on a warm session saw the previous turn's key and was reported interrupted. The running lock is the discriminator. Also gives alive-but-idle rows a 30-minute sweep grace (matching the approval park TTL) and stops announcing running for a killed tombstone. --- api/oss/src/core/sessions/streams/service.py | 51 ++++++++++++++++--- .../tasks/asyncio/sessions/orphan_sweep.py | 28 +++++++--- 2 files changed, 65 insertions(+), 14 deletions(-) diff --git a/api/oss/src/core/sessions/streams/service.py b/api/oss/src/core/sessions/streams/service.py index 080f875a84..c1e3352608 100644 --- a/api/oss/src/core/sessions/streams/service.py +++ b/api/oss/src/core/sessions/streams/service.py @@ -34,6 +34,8 @@ release_running, force_cancel_alive, force_clear_owner, + get_alive_owner, + get_running_owner, get_session_liveness, refresh_alive, refresh_running, @@ -334,11 +336,11 @@ async def heartbeat( if request.turn_id and request.is_running: # 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 alive acquire is the unambiguous takeover signal: nx only fails when a - # different turn holds the key right now. A successful one means the key was - # merely absent, which is an interruption only if this turn had already - # established it (`turn_was_established`) rather than establishing it here. - # (`acquire_running` is not nx — it overwrites — so it carries no such signal.) + # 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 + # 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. if not await refresh_alive( self._lock, project_id=str(project_id), @@ -351,6 +353,36 @@ async def heartbeat( session_id=request.session_id, turn_id=request.turn_id, ) + if not acquired: + alive_owner = await get_alive_owner( + self._lock, + project_id=str(project_id), + session_id=request.session_id, + ) + running_owner = await get_running_owner( + self._lock, + project_id=str(project_id), + session_id=request.session_id, + ) + if alive_owner == request.turn_id: + # Two overlapping beats of this same turn raced; we still own it. + acquired = True + elif running_owner is not None and running_owner != request.turn_id: + 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( + 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( @@ -432,7 +464,14 @@ async def heartbeat( # for send/steer, but the runner mints its own turn id and only ever heartbeats, so # without this the relay's `running` event never fires for a real run. Gated on the # turn being new to this row, so a 30s-interval beat doesn't re-publish it. - if request.turn_id and request.is_running and not turn_was_established: + # `stream is None` = the row is a killed tombstone; announcing a dead session as + # running would light every watcher's badge for a run that cannot exist. + if ( + request.turn_id + and request.is_running + and not turn_was_established + and stream is not None + ): await self._publish_lifecycle( project_id=project_id, session_id=request.session_id, diff --git a/api/oss/src/tasks/asyncio/sessions/orphan_sweep.py b/api/oss/src/tasks/asyncio/sessions/orphan_sweep.py index ef929d8517..4acebe7045 100644 --- a/api/oss/src/tasks/asyncio/sessions/orphan_sweep.py +++ b/api/oss/src/tasks/asyncio/sessions/orphan_sweep.py @@ -24,13 +24,19 @@ force_clear_owner, ) -from sqlalchemy import func, select +from sqlalchemy import and_, func, not_, or_, select log = get_module_logger(__name__) -# A stream whose heartbeat (updated_at) is older than this is considered orphaned. +# A RUNNING stream whose heartbeat (updated_at) is older than this is orphaned: a live turn +# beats every 30s, so this much silence means the owning runner died. ORPHAN_THRESHOLD_SECONDS: int = 300 # 5 minutes +# Alive-but-idle rows (between turns, or parked awaiting approval) get a longer grace: the +# runner stops beating while a turn is parked, and it keeps that sandbox warm for the +# approval TTL (30 min). Sweeping those at 5 min would declare a resumable session dead. +IDLE_THRESHOLD_SECONDS: int = 1800 # 30 minutes + # How often the sweep runs. SWEEP_INTERVAL_SECONDS: int = 60 @@ -40,7 +46,14 @@ async def run_orphan_sweep(engine: TransactionsEngine, lock_engine: LockEngine) -> None: """Single sweep pass: mark stale is_alive rows as ended.""" - threshold = datetime.now(timezone.utc) - timedelta(seconds=ORPHAN_THRESHOLD_SECONDS) + now_utc = datetime.now(timezone.utc) + threshold = now_utc - timedelta(seconds=ORPHAN_THRESHOLD_SECONDS) + idle_threshold = now_utc - timedelta(seconds=IDLE_THRESHOLD_SECONDS) + # coalesce, not a bare `updated_at`: a row never updated since creation has updated_at + # NULL, and `NULL < threshold` is NULL — such a row could never be swept, however long + # it had claimed to be alive. + last_beat = func.coalesce(SessionStreamDBE.updated_at, SessionStreamDBE.created_at) + is_running = SessionStreamDBE.flags.contains({"is_running": True}) async with engine.session() as session: stmt = ( @@ -48,11 +61,10 @@ async def run_orphan_sweep(engine: TransactionsEngine, lock_engine: LockEngine) .where( SessionStreamDBE.deleted_at.is_(None), SessionStreamDBE.flags.contains({"is_alive": True}), - # coalesce, not a bare `updated_at <`: a row never updated since creation has - # updated_at NULL, and `NULL < threshold` is NULL — such a row could never be - # swept, however long it had claimed to be alive. - func.coalesce(SessionStreamDBE.updated_at, SessionStreamDBE.created_at) - < threshold, + or_( + and_(is_running, last_beat < threshold), + and_(not_(is_running), last_beat < idle_threshold), + ), ) .limit(SWEEP_BATCH_SIZE) ) From 40adc696c64691d41557c7d06cb58a56afcf1a05 Mon Sep 17 00:00:00 2001 From: Arda Erzin Date: Tue, 28 Jul 2026 17:25:50 +0300 Subject: [PATCH 13/18] test(api): cover the second turn on a warm session and the sweep's two thresholds MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The failed-acquire branch had no test for the state every follow-up turn is in — a stale alive lock left by this session's own previous turn — so treating it as a takeover shipped unnoticed. Pins the four states that branch must tell apart (warm handover, live takeover, same-turn self-race, resume after a park), the killed-tombstone publish gate, and the sweep's running/idle thresholds by evaluating the real WHERE expression against fake rows. --- .../sessions/test_heartbeat_turn_handover.py | 220 +++++++++++++ .../sessions/test_orphan_sweep_thresholds.py | 303 ++++++++++++++++++ .../sessions/test_watch_lifecycle_publish.py | 55 ++++ 3 files changed, 578 insertions(+) create mode 100644 api/oss/tests/pytest/unit/sessions/test_heartbeat_turn_handover.py create mode 100644 api/oss/tests/pytest/unit/sessions/test_orphan_sweep_thresholds.py diff --git a/api/oss/tests/pytest/unit/sessions/test_heartbeat_turn_handover.py b/api/oss/tests/pytest/unit/sessions/test_heartbeat_turn_handover.py new file mode 100644 index 0000000000..d9949696e9 --- /dev/null +++ b/api/oss/tests/pytest/unit/sessions/test_heartbeat_turn_handover.py @@ -0,0 +1,220 @@ +"""A second turn on a WARM session is a handover, not a takeover. + +`alive` outlives the turn that took it — `release_alive` has no callers and the turn-end beat +clears only `running` — and the runner mints a fresh uuid turn id for every message. So the +FIRST beat of turn 2 (and 3, and 4...) always finds turn 1's `alive` key still there and its +nx acquire always fails. Treating that failed acquire as the takeover signal reported +`is_current_turn: false` to a turn that nothing had interrupted, and the runner's watchdog +(`services/runner/src/sessions/alive.ts` -> `onInterrupted` -> `controller.abort()`) aborted +every follow-up turn before it emitted a token, for the whole ~5-minute window before the +orphan sweep cleared the stale key. + +`running` is the discriminator: a real takeover (steer / `_start_turn`) holds it under the +usurper's turn id, whereas a session that is merely warm holds no `running` at all. These +tests pin the four states the failed-acquire branch has to tell apart. +""" + +from typing import Optional +from unittest.mock import AsyncMock, patch +from uuid import UUID, uuid4 + +import pytest +import pytest_asyncio + +from oss.src.core.sessions.streams.dtos import ( + SessionHeartbeatRequest, + SessionStream, +) +from oss.src.core.sessions.streams.service import SessionStreamsService +from oss.src.dbs.redis.sessions.locks import ( + acquire_alive, + get_alive_owner, + get_running_owner, +) + +from unit.sessions.test_project_scoped_locks import _FakeRedis + + +_PROJECT = uuid4() +_SESSION = "session_warm_handover" + + +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 + ) + + +@pytest.mark.asyncio +async def test_second_turn_on_a_warm_session_is_current(lock_engine): + """The bug that shipped: turn 2's first beat, with turn 1's `alive` still in place.""" + svc = _service(lock_engine) + + await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-1")) + await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-1")) + # Turn end: `release()` beats once with is_running=false. It clears ONLY `running` — + # `alive` is what keeps the session warm/reattachable, so it survives by design. + await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-1", running=False)) + assert await _alive(lock_engine) == "turn-1" + assert await _running(lock_engine) is None + + second = await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-2")) + + assert second.is_current_turn is True, ( + "turn 2 was aborted before its first token: a stale `alive` from this session's own " + "previous turn is a handover, not a takeover" + ) + assert await _alive(lock_engine) == "turn-2", ( + "the handover must actually transfer the lock, or turn 2 never owns the nest" + ) + assert await _running(lock_engine) == "turn-2" + + +@pytest.mark.asyncio +async def test_a_live_different_turn_holding_running_is_still_a_takeover(lock_engine): + """The signal `is_current_turn` exists for: another turn owns the session RIGHT NOW.""" + svc = _service(lock_engine) + + # turn-2 nests first and stays running (both locks held under its id). + await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-2")) + assert await _alive(lock_engine) == "turn-2" + assert await _running(lock_engine) == "turn-2" + + # turn-1's watchdog beat, still in flight on the runner, must learn it lost the session. + old_turn = await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-1")) + + assert old_turn.is_current_turn is False + assert await _alive(lock_engine) == "turn-2", ( + "a superseded turn must not force-cancel the live turn's alive lock" + ) + + +@pytest.mark.asyncio +async def test_overlapping_beats_of_the_same_turn_stay_current(lock_engine): + """Self-race: two beats of ONE turn overlap, so the second sees its own id on `alive` + between its refresh (which read the key as absent/foreign) and its acquire. Same-turn + ownership is not a takeover, and it must not force-cancel a lock we already hold. + """ + svc = _service(lock_engine) + + # `_start_turn` (or this turn's own in-flight first beat) already nested turn-1. + await acquire_alive( + lock_engine, project_id=str(_PROJECT), session_id=_SESSION, turn_id="turn-1" + ) + + cancels: list[str] = [] + + async def _spy_force_cancel(engine, *, project_id, session_id): + cancels.append(session_id) + return None + + # refresh_alive returning False while the key holds OUR id is exactly the interleaving: + # the GET raced the concurrent beat's write. + with ( + patch( + "oss.src.core.sessions.streams.service.refresh_alive", + new=AsyncMock(return_value=False), + ), + patch( + "oss.src.core.sessions.streams.service.force_cancel_alive", + new=_spy_force_cancel, + ), + ): + result = await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-1")) + + assert result.is_current_turn is True + assert cancels == [], "we already own `alive`; there is nothing to hand over" + assert await _alive(lock_engine) == "turn-1" + + +@pytest.mark.asyncio +async def test_resume_after_a_parked_turn_is_current(lock_engine): + """A turn parked awaiting approval ends its run (so `running` clears) but keeps the + sandbox — and the `alive` key — warm for the approval TTL. The resume mints a NEW turn id, + whose first beat therefore finds the parked turn's `alive` still held. + """ + svc = _service(lock_engine) + + await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-parked")) + # Park: the run returns, `release()` beats is_running=false, the runner stops beating. + await svc.heartbeat( + project_id=_PROJECT, request=_beat("turn-parked", running=False) + ) + assert await _alive(lock_engine) == "turn-parked" + + resumed = await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-resume")) + + assert resumed.is_current_turn is True, ( + "an approval resume must not be aborted by the parked turn's own alive key" + ) + assert await _running(lock_engine) == "turn-resume", ( + "the resumed turn must own `running` — it is the takeover discriminator for the " + "next turn, and the mirror the watch relay reads" + ) + assert await _alive(lock_engine) == "turn-resume" + assert resumed.stream is not None + assert resumed.stream.turn_id == "turn-resume" + assert resumed.stream.flags.is_running is True diff --git a/api/oss/tests/pytest/unit/sessions/test_orphan_sweep_thresholds.py b/api/oss/tests/pytest/unit/sessions/test_orphan_sweep_thresholds.py new file mode 100644 index 0000000000..20f197cadf --- /dev/null +++ b/api/oss/tests/pytest/unit/sessions/test_orphan_sweep_thresholds.py @@ -0,0 +1,303 @@ +"""The orphan sweep needs TWO staleness thresholds, not one. + +A live turn beats every 30s, so 5 minutes of silence from a RUNNING row means the owning +runner died. An alive-but-idle row is a different animal: between turns, and while a turn is +parked awaiting approval, the runner stops beating entirely but keeps the sandbox warm for the +30-minute approval TTL. Sweeping those at 5 minutes collapsed the flags and force-cancelled +the Redis nest of a session the user was about to approve and resume. + +No Postgres here (this is the unit suite; the sibling `test_orphan_sweep_clears_redis.py` +stands in for both stores with in-memory fakes). Rather than assert on the compiled SQL +string, `_FakePgSession` below EVALUATES the sweep's real `WHERE` expression tree against +in-memory rows using SQL three-valued logic, so these tests exercise the predicate the sweep +actually builds — including `@>` containment semantics on absent keys and NULL `flags`. +""" + +from contextlib import asynccontextmanager +from datetime import datetime, timezone, timedelta +from typing import Optional + +import pytest +from sqlalchemy.sql import operators +from sqlalchemy.sql.elements import ( + AsBoolean, + BinaryExpression, + BindParameter, + BooleanClauseList, + ColumnClause, + Grouping, + Null, + UnaryExpression, +) +from sqlalchemy.sql.functions import Function + +from oss.src.tasks.asyncio.sessions.orphan_sweep import ( + IDLE_THRESHOLD_SECONDS, + ORPHAN_THRESHOLD_SECONDS, + run_orphan_sweep, +) + +_PROJECT_ID = "proj-sweep-1" + + +# --------------------------------------------------------------------------- # +# Minimal SQL evaluator for the predicate shapes the sweep uses +# --------------------------------------------------------------------------- # + + +def _sql_and(values): + if any(v is False for v in values): + return False + return None if any(v is None for v in values) else True + + +def _sql_or(values): + if any(v is True for v in values): + return True + return None if any(v is None for v in values) else False + + +def _contains(left, right) -> Optional[bool]: + """Postgres `jsonb @> jsonb`: NULL propagates; a missing key is FALSE, not NULL.""" + if left is None: + return None + return all(key in left and left[key] == value for key, value in right.items()) + + +def _evaluate(node, row) -> Optional[bool]: + if isinstance(node, Grouping): + return _evaluate(node.element, row) + if isinstance(node, BooleanClauseList): + parts = [_evaluate(clause, row) for clause in node.clauses] + return (_sql_and if node.operator is operators.and_ else _sql_or)(parts) + if isinstance(node, AsBoolean) and node.operator is operators.is_false: + inner = _evaluate(node.element, row) # `not_(...)` on a boolean expression + return None if inner is None else not inner + if isinstance(node, UnaryExpression) and node.operator is operators.inv: + inner = _evaluate(node.element, row) + return None if inner is None else not inner + if isinstance(node, BinaryExpression): + left, right = _value(node.left, row), _value(node.right, row) + if node.operator is operators.is_: + return left is right + if node.operator is operators.lt: + return None if left is None or right is None else left < right + if getattr(node.operator, "opstring", None) == "@>": + return _contains(left, right) + raise AssertionError( + f"the sweep grew a predicate this evaluator cannot read: {node!r}" + ) + + +def _value(node, row): + if isinstance(node, Grouping): + return _value(node.element, row) + if isinstance(node, BindParameter): + return node.value + if isinstance(node, Null): + return None + if isinstance(node, Function) and node.name == "coalesce": + for clause in node.clauses: + candidate = _value(clause, row) + if candidate is not None: + return candidate + return None + if isinstance(node, ColumnClause): + return getattr(row, node.key) + raise AssertionError(f"unreadable operand: {node!r}") + + +# --------------------------------------------------------------------------- # +# Fakes (same shape as test_orphan_sweep_clears_redis.py, plus real filtering) +# --------------------------------------------------------------------------- # + + +class _FakeRow: + def __init__(self, *, session_id: str, flags: Optional[dict], age_seconds: int): + self.session_id = session_id + self.project_id = _PROJECT_ID + self.id = session_id + self.deleted_at = None + self.flags = flags + self.created_at = datetime.now(timezone.utc) - timedelta(days=1) + self.updated_at = datetime.now(timezone.utc) - timedelta(seconds=age_seconds) + + +class _FakeScalars: + def __init__(self, rows): + self._rows = rows + + def all(self): + return self._rows + + +class _FakeResult: + def __init__(self, rows): + self._rows = rows + + def scalars(self): + return _FakeScalars(self._rows) + + +class _FakePgSession: + def __init__(self, rows): + self._rows = rows + + async def execute(self, stmt): + matched = [ + row for row in self._rows if _evaluate(stmt.whereclause, row) is True + ] + return _FakeResult(matched) + + async def commit(self): + pass + + +class _FakeTransactionsEngine: + def __init__(self, rows): + self._rows = rows + + @asynccontextmanager + async def session(self): + yield _FakePgSession(self._rows) + + +class _FakeRedis: + def __init__(self): + self._store: dict[str, bytes] = {} + + async def get(self, key): + return self._store.get(key) + + async def set(self, key, value, nx=False, ex=None): + if nx and key in self._store: + return None + self._store[key] = value + return True + + async def delete(self, key): + self._store.pop(key, None) + return 1 + + async def expire(self, key, ttl): + return True + + +def _swept(row: _FakeRow) -> bool: + return row.flags == {"is_alive": False, "is_running": False, "is_attached": False} + + +@pytest.fixture +def anyio_backend(): + return "asyncio" + + +@pytest.mark.anyio +async def test_running_row_is_swept_at_the_short_threshold(anyio_backend): + row = _FakeRow( + session_id="sess-running-stale", + flags={"is_alive": True, "is_running": True, "is_attached": False}, + age_seconds=360, + ) + + await run_orphan_sweep(_FakeTransactionsEngine([row]), _FakeRedis()) + + assert _swept(row), ( + "6 minutes of silence from a turn that beats every 30s means the runner died" + ) + + +@pytest.mark.anyio +async def test_idle_row_survives_the_short_threshold(anyio_backend): + """The regression: a turn parked awaiting approval stops beating but stays resumable.""" + row = _FakeRow( + session_id="sess-parked", + flags={"is_alive": True, "is_running": False, "is_attached": False}, + age_seconds=360, + ) + + await run_orphan_sweep(_FakeTransactionsEngine([row]), _FakeRedis()) + + assert not _swept(row), ( + "an alive-but-idle session was declared dead 5 minutes in, while the user still had " + "25 minutes of approval TTL to resume it" + ) + assert row.flags["is_alive"] is True + + +@pytest.mark.anyio +async def test_idle_row_is_swept_at_the_long_threshold(anyio_backend): + row = _FakeRow( + session_id="sess-idle-dead", + flags={"is_alive": True, "is_running": False, "is_attached": False}, + age_seconds=IDLE_THRESHOLD_SECONDS + 60, + ) + + await run_orphan_sweep(_FakeTransactionsEngine([row]), _FakeRedis()) + + assert _swept(row), ( + "past the approval TTL the sandbox is gone; the row must not stay alive forever" + ) + + +@pytest.mark.anyio +async def test_thresholds_are_five_and_thirty_minutes(anyio_backend): + assert (ORPHAN_THRESHOLD_SECONDS, IDLE_THRESHOLD_SECONDS) == (300, 1800) + + +@pytest.mark.anyio +async def test_partial_flags_are_treated_as_idle_not_running(anyio_backend): + """`flags @> '{"is_running": true}'` is FALSE (not NULL) for a row whose JSON simply lacks + the key, so `not_(...)` puts it on the idle branch rather than dropping it from both.""" + young = _FakeRow( + session_id="sess-partial-young", flags={"is_alive": True}, age_seconds=360 + ) + old = _FakeRow( + session_id="sess-partial-old", + flags={"is_alive": True}, + age_seconds=IDLE_THRESHOLD_SECONDS + 60, + ) + + await run_orphan_sweep(_FakeTransactionsEngine([young, old]), _FakeRedis()) + + assert not _swept(young) + assert _swept(old), "a row lacking is_running must still be reclaimable" + + +@pytest.mark.anyio +async def test_rows_that_never_claimed_alive_are_never_swept(anyio_backend): + """NULL flags (a row created by rename alone) make every `@>` test NULL, and a row that + says is_alive=false was never the sweep's business.""" + null_flags = _FakeRow(session_id="sess-null", flags=None, age_seconds=99_999) + not_alive = _FakeRow( + session_id="sess-ended", + flags={"is_alive": False, "is_running": False, "is_attached": False}, + age_seconds=99_999, + ) + + await run_orphan_sweep( + _FakeTransactionsEngine([null_flags, not_alive]), _FakeRedis() + ) + + assert null_flags.flags is None + assert not_alive.flags["is_alive"] is False + + +@pytest.mark.anyio +async def test_sweep_clears_redis_for_the_long_threshold_branch(anyio_backend): + """Whichever branch selected a row, the Redis nest must follow the row — otherwise the + SEND gate keeps reading `alive` from a session the sweep just declared dead.""" + session_id = "sess-idle-dead-redis" + redis = _FakeRedis() + await redis.set(f"alive:{_PROJECT_ID}:session:{session_id}", b"turn-1", ex=3600) + await redis.set(f"owner:{_PROJECT_ID}:session:{session_id}", b"replica-1", ex=3600) + row = _FakeRow( + session_id=session_id, + flags={"is_alive": True, "is_running": False, "is_attached": False}, + age_seconds=IDLE_THRESHOLD_SECONDS + 60, + ) + + await run_orphan_sweep(_FakeTransactionsEngine([row]), redis) + + assert await redis.get(f"alive:{_PROJECT_ID}:session:{session_id}") is None + assert await redis.get(f"owner:{_PROJECT_ID}:session:{session_id}") is None diff --git a/api/oss/tests/pytest/unit/sessions/test_watch_lifecycle_publish.py b/api/oss/tests/pytest/unit/sessions/test_watch_lifecycle_publish.py index 51caf48561..1859c655f3 100644 --- a/api/oss/tests/pytest/unit/sessions/test_watch_lifecycle_publish.py +++ b/api/oss/tests/pytest/unit/sessions/test_watch_lifecycle_publish.py @@ -22,6 +22,7 @@ SessionStreamCommandRequest, ) from oss.src.core.sessions.streams.service import SessionStreamsService +from oss.src.core.sessions.streams.types import SessionStreamAlreadyExists from unit.sessions.test_project_scoped_locks import _FakeRedis @@ -248,6 +249,60 @@ async def test_first_beat_of_a_runner_minted_turn_publishes_running(lock_engine) assert publisher.lifecycle_calls == [(str(_PROJECT), session_id, "running")] +class _TombstoneDAO: + """A killed session: the row is soft-deleted, so `get` and `update` see nothing and + `create` loses the (project_id, session_id) unique slot to the tombstone.""" + + def __init__(self): + self.creates = 0 + self.updates = 0 + + async def get_by_session_id(self, *, project_id: UUID, session_id: str): + return None + + async def create(self, *, project_id, user_id, stream): + self.creates += 1 + raise SessionStreamAlreadyExists(session_id=stream.session_id) + + async def update(self, *, project_id, user_id, session_id, stream): + self.updates += 1 + return None + + async def delete_by_session_id(self, *, project_id, session_id): + return True + + +@pytest.mark.asyncio +async def test_beat_on_a_killed_tombstone_publishes_nothing(lock_engine): + """A late beat from a killed turn's watchdog resolves to no row at all (create loses the + slot to the tombstone, update matches nothing). Announcing `running` for it would light + every watcher's badge for a session that cannot run — and nothing ever clears it, because + the turn-end beat's `ended` publish is gated on a `running` key that this beat is the only + thing to have armed.""" + publisher = _RecordingPublisher() + dao = _TombstoneDAO() + svc = SessionStreamsService( + streams_dao=dao, lock_engine=lock_engine, watch_publisher=publisher + ) + session_id = _session_id() + + result = await svc.heartbeat( + project_id=_PROJECT, + request=SessionHeartbeatRequest( + session_id=session_id, + replica_id="replica-1", + turn_id="killed-turn", + is_running=True, + ), + ) + + assert (dao.creates, dao.updates) == (1, 1), "both row paths must have been tried" + assert result.stream is None, "a killed session resolves to no row" + assert publisher.lifecycle_calls == [], ( + "a dead tombstone must never be announced as running" + ) + + @pytest.mark.asyncio async def test_send_does_not_double_publish_running_on_its_first_beat(lock_engine): """`_start_turn` already recorded the turn on the row, so the runner's first beat for From 4d22cd9028682c201790989392e88911e7d1e80c Mon Sep 17 00:00:00 2001 From: Arda Erzin Date: Tue, 28 Jul 2026 17:39:19 +0300 Subject: [PATCH 14/18] fix(api): a superseded turn's beat must not re-arm the running lock MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit acquire_running overwrites, so the beat that was just told is_current_turn=false stamped its own dead turn id over the live turn's running key — the exact key the takeover check now reads to tell a warm handover from a real steer. One zombie beat was enough to make the live turn look superseded to the next one, and to resurrect running for a turn a cancel had just cleared. --- api/oss/src/core/sessions/streams/service.py | 18 ++++++++++++------ .../sessions/test_heartbeat_turn_handover.py | 9 +++++++++ 2 files changed, 21 insertions(+), 6 deletions(-) diff --git a/api/oss/src/core/sessions/streams/service.py b/api/oss/src/core/sessions/streams/service.py index c1e3352608..fafdfafe6c 100644 --- a/api/oss/src/core/sessions/streams/service.py +++ b/api/oss/src/core/sessions/streams/service.py @@ -393,12 +393,18 @@ async def heartbeat( ): if turn_was_established: is_current_turn = False - await acquire_running( - self._lock, - project_id=str(project_id), - session_id=request.session_id, - turn_id=request.turn_id, - ) + # Only a turn that still believes it owns the session may (re-)arm `running`: + # acquire_running overwrites, so a superseded turn's beat would otherwise + # stamp its own dead id over the live turn's — corrupting the very key the + # takeover check above reads — and a cancelled turn would resurrect the + # `running` its cancel just cleared. + if is_current_turn: + await acquire_running( + self._lock, + project_id=str(project_id), + session_id=request.session_id, + turn_id=request.turn_id, + ) elif not request.is_running: # Turn ended: drop only `running`. `alive` outlives the turn (own TTL, cleared # only by kill) — this is what makes the session reattachable. diff --git a/api/oss/tests/pytest/unit/sessions/test_heartbeat_turn_handover.py b/api/oss/tests/pytest/unit/sessions/test_heartbeat_turn_handover.py index d9949696e9..68fc63e457 100644 --- a/api/oss/tests/pytest/unit/sessions/test_heartbeat_turn_handover.py +++ b/api/oss/tests/pytest/unit/sessions/test_heartbeat_turn_handover.py @@ -150,6 +150,15 @@ async def test_a_live_different_turn_holding_running_is_still_a_takeover(lock_en assert await _alive(lock_engine) == "turn-2", ( "a superseded turn must not force-cancel the live turn's alive lock" ) + assert await _running(lock_engine) == "turn-2", ( + "`acquire_running` overwrites, so a beat that was just told it is not the current " + "turn must not stamp its dead id over the live turn's — that key is what the " + "takeover check reads, and clobbering it makes the LIVE turn look superseded" + ) + + # And the live turn keeps running: the discriminator survived the zombie's beat. + live = await svc.heartbeat(project_id=_PROJECT, request=_beat("turn-2")) + assert live.is_current_turn is True @pytest.mark.asyncio From 30900cdd47c46469f6d97099c2001dfe0085566b Mon Sep 17 00:00:00 2001 From: Arda Erzin Date: Tue, 28 Jul 2026 17:42:14 +0300 Subject: [PATCH 15/18] docs(mobile): track the parked-session lock ambiguity --- .../2026-07-27-mobile-approvals-steering.md | 29 +++++++++++++++++++ 1 file changed, 29 insertions(+) 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 905c415845..06269824d8 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 @@ -335,3 +335,32 @@ Execution scope now: M0 + M1 (minus M1.5 steer) + TTL bump + M2. - **Records-poll cost:** the records query is the heavy one (~200KB on long sessions, backend 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) + +Live QA of approvals surfaced a dead liveness mirror in `SessionStreamsService.heartbeat` +(fixed: b0281c5788, 6761727847, 2174162d80, 5d2ed61e9f, 076dc41b7e — see the memory entry +for the full chain, incl. 1181 phantom `is_running` rows and a project sitting at 984/1000 +`CONCURRENCY_LIMIT`). Two defects the review chain caught before they could bite are fixed; +ONE residual correctness gap remains and is **deliberately not hotfixed** because closing it +needs a lock-contract change mirrored on the runner side: + +**The gap.** `alive` outlives its turn (`release_alive` has no callers) and a parked turn +clears `running`, so the state "`alive` held by another turn + no `running`" is genuinely +ambiguous between (a) a lapsed previous turn — the common case, which MUST be treated as a +legitimate handover or every follow-up turn aborts — and (b) a live-but-parked or +just-starting turn. We resolve it as (a). Consequence: a zombie beat from an older turn can +take the nest of a session parked awaiting approval, and the user's approval resume then +reports `is_current_turn=False` and aborts. Narrow today (`_start_turn` is off the product +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. + +**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 +alive-but-idle rows rather than 5. From 588c604e0900d8697b88c77b7f25d662310059b3 Mon Sep 17 00:00:00 2001 From: Arda Erzin Date: Tue, 28 Jul 2026 21:22:26 +0300 Subject: [PATCH 16/18] fix(mobile): send the bearer token on the approval resume The cookie authenticates the invoke, but the SDK resolves the model connection by fetching the vault with the caller's Authorization header only. Without it the resumed run got no injected credential and the model rejected it. --- .../src/features/chat/useApprovalActions.ts | 9 ++++++++- web/mobile/src/lib/auth.ts | 19 +++++++++++++++++++ 2 files changed, 27 insertions(+), 1 deletion(-) diff --git a/web/mobile/src/features/chat/useApprovalActions.ts b/web/mobile/src/features/chat/useApprovalActions.ts index cafded1a1c..8296d58905 100644 --- a/web/mobile/src/features/chat/useApprovalActions.ts +++ b/web/mobile/src/features/chat/useApprovalActions.ts @@ -9,6 +9,8 @@ import { } from "@agenta/chat/transport" import {queryInteractions} from "@agenta/entities/session" +import {getAuthorizationHeader} from "@/lib/auth" + import {stampApprovalResponses} from "./approvalStamp" export type ResumePhase = "idle" | "resuming" | "error" @@ -133,9 +135,14 @@ export const useApprovalActions = ({ projectId, applicationId: references.application?.id ?? undefined, }) + const authHeader = await getAuthorizationHeader() const response = await fetch(request.invocationUrl, { method: "POST", - headers: {...request.headers, "Content-Type": "application/json"}, + headers: { + ...request.headers, + ...authHeader, + "Content-Type": "application/json", + }, body: JSON.stringify(request.requestBody), credentials: "include", }) diff --git a/web/mobile/src/lib/auth.ts b/web/mobile/src/lib/auth.ts index 8486519c61..280d571367 100644 --- a/web/mobile/src/lib/auth.ts +++ b/web/mobile/src/lib/auth.ts @@ -87,6 +87,25 @@ export async function signInWithEmailPassword( } } +/** + * `Authorization` for an invoke, mirroring the desktop's `getJWT()` + * (web/oss/src/services/api.ts). The cookie alone authenticates the invoke, but the SDK + * resolves the model connection by fetching the vault with the caller's Authorization + * header ONLY — without it the run proceeds with no injected credential and the model + * rejects it ("no connection resolved for provider …"). + */ +export async function getAuthorizationHeader(): Promise> { + if (typeof window === "undefined") return {} + ensureAuthInit() + try { + if (!(await Session.doesSessionExist())) return {} + const jwt = await Session.getAccessToken() + return jwt ? {Authorization: `Bearer ${jwt}`} : {} + } catch { + return {} + } +} + /** * Attempt a cookie-based session refresh. Resolves false when there is no * refresh token or the backend rejects it — the caller's signed-out verdict From 65b72033c03d94688a8d3bb6be6995e8f81e37fa Mon Sep 17 00:00:00 2001 From: Arda Erzin Date: Mon, 3 Aug 2026 22:35:49 +0300 Subject: [PATCH 17/18] fix(api): say why a watch pubsub teardown failed The teardown warning carried only the channel, and the except block dropped the exception object, so a genuine Redis failure in production logged nothing to diagnose it with. Teardown stays best-effort; it just says what went wrong. --- api/oss/src/apis/fastapi/sessions/watch.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/api/oss/src/apis/fastapi/sessions/watch.py b/api/oss/src/apis/fastapi/sessions/watch.py index 0758b9ef05..9f98d61daf 100644 --- a/api/oss/src/apis/fastapi/sessions/watch.py +++ b/api/oss/src/apis/fastapi/sessions/watch.py @@ -77,5 +77,9 @@ async def watch_event_stream( try: await pubsub.unsubscribe(channel) await pubsub.aclose() - except Exception: # pragma: no cover — teardown is best-effort - log.warning("[WATCH] pubsub teardown failed", channel=channel) + except Exception as exc: # pragma: no cover — teardown is best-effort + log.warning( + "[WATCH] pubsub teardown failed", + channel=channel, + error=repr(exc), + ) From e062f59911a2f31252b85fef7b40bea96d46c180 Mon Sep 17 00:00:00 2001 From: Arda Erzin Date: Tue, 4 Aug 2026 00:53:59 +0300 Subject: [PATCH 18/18] fix(api): close the watch pubsub even when unsubscribe fails Teardown ran unsubscribe and aclose in one try block, so an exception from the first skipped the second and the Redis connection outlived the disconnected SSE client. They are separate attempts now, each logging its own failure, both still best-effort. Also corrects a comment on the publish test that claimed failed appends "stay in the existing retry path". There is no such path: process_batch acknowledges at parse time, before the append, and the shared consumer loop deletes what it returns. That predates this work and is shared by every worker on BaseStreamConsumer; the relay tee neither causes it nor repairs it. The comment now says so instead of implying a retry that does not exist. --- api/oss/src/apis/fastapi/sessions/watch.py | 11 ++++++++++- .../tests/pytest/unit/sessions/test_watch_publish.py | 5 ++++- 2 files changed, 14 insertions(+), 2 deletions(-) diff --git a/api/oss/src/apis/fastapi/sessions/watch.py b/api/oss/src/apis/fastapi/sessions/watch.py index 9f98d61daf..0a9e056e63 100644 --- a/api/oss/src/apis/fastapi/sessions/watch.py +++ b/api/oss/src/apis/fastapi/sessions/watch.py @@ -74,12 +74,21 @@ async def watch_event_stream( if frame is not None: yield frame finally: + # Two independent attempts: a failing unsubscribe must not skip the close, or the + # connection outlives the disconnected client. try: await pubsub.unsubscribe(channel) + except Exception as exc: # pragma: no cover — teardown is best-effort + log.warning( + "[WATCH] pubsub unsubscribe failed", + channel=channel, + error=repr(exc), + ) + try: await pubsub.aclose() except Exception as exc: # pragma: no cover — teardown is best-effort log.warning( - "[WATCH] pubsub teardown failed", + "[WATCH] pubsub close failed", channel=channel, error=repr(exc), ) diff --git a/api/oss/tests/pytest/unit/sessions/test_watch_publish.py b/api/oss/tests/pytest/unit/sessions/test_watch_publish.py index 5c948c7342..117a3522fb 100644 --- a/api/oss/tests/pytest/unit/sessions/test_watch_publish.py +++ b/api/oss/tests/pytest/unit/sessions/test_watch_publish.py @@ -124,7 +124,10 @@ async def test_worker_skips_publish_when_append_fails(): assert total_appended == 0 assert publisher.calls == [] - # Failed appends stay in the existing retry path; publish adds nothing to it. + # `process_batch` acknowledges at parse time, before the append, so a failed append is still + # acked and dropped by the shared consumer loop. That predates this change and is shared by + # every worker on `BaseStreamConsumer`; the relay tee neither causes it nor repairs it. This + # assertion pins the tee's scope, not an endorsement of the acknowledgement rule. assert len(processed_ids) == 1