[feat] Stream session changes and fix session liveness (9/12)

[ardaerzin]

Context

Mobile polled to notice that something changed, which is the wrong shape for a phone: too slow when you are watching a turn, too chatty when you are not. Separately, session liveness was unreliable — rows sat at is_running=true for days after a crashed run, so any badge derived from them lied.

Changes

A metadata-only SSE relay, GET /sessions/streams/watch, publishes per-session change notifications on the durable Redis plane. It carries no payload: the client is told that something changed and re-reads through its normal cache, so the stream can never be a second source of truth. Mobile consumes it via useSessionWatch, and the client reconnect delay is pinned server-side.

The liveness fixes are the subtler half. The heartbeat writes its mirror on every beat rather than only the first. A stale alive lock is treated as handover, not as a turn takeover. A superseded turn's beat can no longer re-arm the running lock — a displaced turn is dead, and its late beats must not resurrect it. The orphan sweep gets two thresholds, one for running turns and a longer one for alive-but-idle sessions, which is what an approval parked for 30 minutes looks like.

Tests / notes

• New pytest coverage for the watch publishes, the endpoint, the heartbeat handover/zombie cases, and the sweep's two thresholds.
• The relay is metadata-only by design. If it ever carries payload, it becomes a second transcript source that can disagree with the records.

What to QA

• Open a session on mobile while a turn runs on desktop. New content appears without a manual refresh.
• Kill a runner mid-turn. The session stops showing as running within the sweep threshold rather than staying "running" indefinitely.

[vercel]

The latest updates on your projects. Learn more about Vercel for GitHub.

Project	Deployment	Actions	Updated (UTC)
agenta-documentation	Ready	Preview	Aug 3, 2026 10:30pm

[coderabbitai]

📝 Walkthrough

Summary by CodeRabbit

• New Features
  • Added live session updates for transcript changes, lifecycle status, and approval activity.
  • Added a mobile approval dock with approve, deny, and approve-all actions.
  • Added readable summaries for approval inputs.
• Bug Fixes
  • Improved approval resume handling and session heartbeat reliability.
  • Improved session liveness indicators and orphaned-session cleanup.
  • Added automatic reconnection and visibility-aware updates for live sessions.
• Tests
  • Expanded coverage for live updates, approvals, liveness, heartbeats, and cleanup behavior.

Walkthrough

Changes

Session watch and approval flow

Layer / File(s)	Summary
Watch channel and publisher contract api/oss/src/dbs/redis/sessions/*, api/oss/src/utils/env.py	Adds per-session Redis channels, event payload builders, best-effort publishing, and configurable SSE heartbeats.
Lifecycle, interaction, and record event producers api/oss/src/core/sessions/*, api/oss/src/tasks/asyncio/sessions/*, api/entrypoints/*	Publishes session lifecycle, interaction, and record changes from API services and workers.
Heartbeat ownership and orphan cleanup api/oss/src/core/sessions/streams/service.py, api/oss/src/dbs/redis/sessions/locks.py, api/oss/src/tasks/asyncio/sessions/orphan_sweep.py, api/oss/tests/pytest/unit/sessions/*heartbeat*, *orphan*, docs/design/...	Updates heartbeat handover, owner-checked lock release, row mirroring, and bounded running and idle orphan sweeps.
Authenticated SSE session relay api/oss/src/apis/fastapi/sessions/*, api/oss/tests/pytest/unit/sessions/test_watch_endpoint.py	Adds GET /sessions/streams/watch, validates access and identifiers, relays recognized Redis events, emits heartbeats, and cleans up subscriptions.
Mobile watch and transcript revalidation web/mobile/src/features/chat/*, web/mobile/src/features/sessions/SessionRow.tsx, web/mobile/src/lib/auth.ts, web/mobile/tests/unit/*, web/mobile/vitest.config.ts	Adds foreground EventSource handling, watch-triggered transcript refresh, watch-aware polling, authenticated requests, and liveness tests.
Mobile approval dock and resume handling web/mobile/src/features/chat/ApprovalDock.tsx, approvalInputSummary.ts, TurnRow.tsx, useApprovalActions.ts	Moves approval controls into ApprovalDock, summarizes inputs, replaces row controls with an awaiting marker, and drains authenticated resume responses.

Estimated code review effort: 5 (Critical) | ~120 minutes

Sequence Diagram(s)

sequenceDiagram
  participant SessionService
  participant Redis
  participant SSEEndpoint
  participant MobileChat
  SessionService->>Redis: publish session watch event
  MobileChat->>SSEEndpoint: open EventSource
  SSEEndpoint->>Redis: subscribe to session channel
  Redis-->>SSEEndpoint: event or heartbeat
  SSEEndpoint-->>MobileChat: SSE frame
  MobileChat->>MobileChat: refresh transcript or approval state

Loading

Possibly related issues

• [runner] Concurrent turns on one session are not gated at start: parallel sandboxes per session, loser reaped only at the next 30s heartbeat, disconnected turns run headless #5539: The heartbeat and turn-handover changes address concurrent-turn behavior and stale-turn safeguards.

Possibly related PRs

• Agenta-AI/agenta#5041: Related approval and session interaction workflow.
• Agenta-AI/agenta#5197: Related session heartbeat and lifecycle ownership handling.
• Agenta-AI/agenta#5470: Related mobile ApprovalDock behavior.

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 14.37% which is insufficient. The required threshold is 60.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title clearly summarizes the two primary changes: session change streaming and session liveness fixes.
Description check	✅ Passed	The description directly explains the SSE relay, mobile integration, liveness fixes, orphan sweep changes, and related tests.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

✨ Finishing Touches 💡 1

📝 Generate docstrings 💡

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch feat/sessions-watch-and-liveness

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[ardaerzin]

@coderabbitai review

[coderabbitai]

✅ Action performed

Review finished.

Note: CodeRabbit is an incremental review system and does not re-review already reviewed commits. This command is applicable only when automatic reviews are paused.

[coderabbitai]

Actionable comments posted: 3

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (2)

web/mobile/src/features/chat/useApprovalActions.ts (1)

94-115: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift

Do not resume unrelated approvals with one invocation.

When target.all is set, Line 94 selects every pending approval. Line 115 then selects references from withRefs[0]. The code documents that parked runs can use different revisions. The resume request can therefore apply approvals from multiple runs with one run's workflow configuration.

Partition approvals by their interaction references and resume each group separately. Otherwise, hide Approve all unless all pending approvals have the same references.

api/oss/src/tasks/asyncio/sessions/orphan_sweep.py (1)

90-103: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick win

Isolate per-row Redis cleanup failures so one bad row does not strand the rest.

The Postgres commit at Line 88 happens before this loop. If force_cancel_alive, clear_running, or force_clear_owner raises for one row, the loop stops and later rows keep stale Redis locks. Since the WHERE clause at Lines 62-63 requires flags @> {"is_alive": True}, a row already committed as is_alive=False will never be reselected by a later sweep pass. This means its stale Redis lock is never retried and persists until its own TTL expires. The new idle branch increases the number of rows that pass through this loop each pass, so this gap now affects more sessions.

The three awaited calls per row also run sequentially, adding up to SWEEP_BATCH_SIZE × 3 round trips per pass.

Wrap each row's cleanup so a failure on one row does not block cleanup of the rest, and run the three calls concurrently.

🛡️ Proposed fix for per-row failure isolation

         # Bring the Redis locks the SEND gate reads in sync with the rows just written.
         for row in orphans:
             project_id = str(row.project_id)
-            await force_cancel_alive(
-                lock_engine, project_id=project_id, session_id=row.session_id
-            )
-            await clear_running(
-                lock_engine, project_id=project_id, session_id=row.session_id
-            )
-            # A swept session is dead; free its affinity like kill does.
-            await force_clear_owner(
-                lock_engine, project_id=project_id, session_id=row.session_id
-            )
+            try:
+                await asyncio.gather(
+                    force_cancel_alive(
+                        lock_engine, project_id=project_id, session_id=row.session_id
+                    ),
+                    clear_running(
+                        lock_engine, project_id=project_id, session_id=row.session_id
+                    ),
+                    # A swept session is dead; free its affinity like kill does.
+                    force_clear_owner(
+                        lock_engine, project_id=project_id, session_id=row.session_id
+                    ),
+                )
+            except Exception:
+                log.exception(
+                    "orphan_sweep: failed to clear redis locks for a swept session",
+                    extra={"session_id": row.session_id, "stream_id": str(row.id)},
+                )

🧹 Nitpick comments (4)

web/mobile/src/features/chat/approvalInputSummary.ts (1)

8-9: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Reduce non-exception multi-line comments.

These comments describe UI intent. Replace them with one short line or remove them.

• web/mobile/src/features/chat/approvalInputSummary.ts#L8-L9: reduce the primary-field explanation to one short line.
• web/mobile/src/features/chat/TurnRow.tsx#L16-L17: remove or reduce the message-style explanation.
• web/mobile/src/features/chat/TurnRow.tsx#L45-L46: reduce the approval-marker explanation to one short line.

As per coding guidelines, keep in-code comments to at most one short line unless a bug, race, or ordering requirement needs more detail.

Source: Coding guidelines

web/mobile/src/features/sessions/SessionRow.tsx (1)

36-38: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Update the liveness prop documentation.

The row no longer falls back to session.flags.is_alive. The prop doc above still states that undefined means "fall back to the list row's own flags". Correct it so callers do not expect the removed behavior.

♻️ Suggested doc fix (lines 28-30)

-    /** Fresh badge from the shared liveness poll; `undefined` = poll unresolved (fall back to
-     * the list row's own flags), `null` = poll resolved and this session is idle. */
+    /** Fresh badge from the shared liveness poll; `undefined` = poll unresolved (no badge),
+     * `null` = poll resolved and this session is idle. */

api/oss/src/apis/fastapi/sessions/router.py (1)

537-595: 🚀 Performance & Scalability | 🔵 Trivial

Operational note: align watch_heartbeat_seconds with proxy/load-balancer idle timeouts.

This handler relies on HEARTBEAT_FRAME comments to keep the SSE connection alive during idle periods, at the interval configured by env.sessions.watch_heartbeat_seconds. Many reverse proxies and load balancers close idle HTTP connections after a fixed timeout, commonly 30-60 seconds. Confirm the configured heartbeat interval stays comfortably below the idle timeout of every proxy in the request path (nginx, ALB, Cloudflare, etc.) to avoid clients experiencing frequent unnecessary reconnects.

The rest of this handler — validation ordering, permission check, lazy stream construction, and response headers — matches the behavior exercised by test_watch_endpoint_rejects_without_view_sessions, test_watch_endpoint_rejects_invalid_session_id, and test_watch_endpoint_returns_event_stream_response.

api/oss/tests/pytest/unit/sessions/test_watch_endpoint.py (1)

128-137: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick win

Add a regression test for non-string type values.

test_format_watch_frame_rejects_non_dict_and_unknown_type covers a non-dict payload and an unknown string type, but not a payload like {"type": ["nope"]} where "type" is itself a list or dict. As noted in the review of api/oss/src/apis/fastapi/sessions/watch.py (Lines 31-42), that input currently raises an uncaught TypeError instead of returning None. Add a case exercising this input once the fix lands, to prevent regression.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: 0e3b76ee-1cc2-4e66-afa0-c4f915a0c484

📥 Commits

Reviewing files that changed from the base of the PR and between 1d70a28 and c4025ac.

📒 Files selected for processing (37)

• api/entrypoints/routers.py
• api/entrypoints/worker_queues.py
• api/entrypoints/worker_streams.py
• api/oss/src/apis/fastapi/sessions/router.py
• api/oss/src/apis/fastapi/sessions/watch.py
• api/oss/src/core/sessions/interactions/service.py
• api/oss/src/core/sessions/streams/service.py
• api/oss/src/dbs/redis/sessions/contract.py
• api/oss/src/dbs/redis/sessions/watch.py
• api/oss/src/tasks/asyncio/sessions/orphan_sweep.py
• api/oss/src/tasks/asyncio/sessions/records_worker.py
• api/oss/src/utils/env.py
• api/oss/tests/pytest/unit/sessions/test_heartbeat_is_current_turn.py
• api/oss/tests/pytest/unit/sessions/test_heartbeat_mirrors_existing_row.py
• api/oss/tests/pytest/unit/sessions/test_heartbeat_turn_handover.py
• api/oss/tests/pytest/unit/sessions/test_orphan_sweep_clears_redis.py
• api/oss/tests/pytest/unit/sessions/test_orphan_sweep_thresholds.py
• api/oss/tests/pytest/unit/sessions/test_watch_endpoint.py
• api/oss/tests/pytest/unit/sessions/test_watch_interactions_publish.py
• api/oss/tests/pytest/unit/sessions/test_watch_lifecycle_publish.py
• api/oss/tests/pytest/unit/sessions/test_watch_publish.py
• docs/design/agenta-mobile/plans/2026-07-27-mobile-approvals-steering.md
• web/mobile/src/features/chat/ApprovalCard.tsx
• web/mobile/src/features/chat/ApprovalDock.tsx
• web/mobile/src/features/chat/ChatHeader.tsx
• web/mobile/src/features/chat/ChatScreen.tsx
• web/mobile/src/features/chat/TurnRow.tsx
• web/mobile/src/features/chat/approvalInputSummary.ts
• web/mobile/src/features/chat/useApprovalActions.ts
• web/mobile/src/features/chat/useSessionTranscript.ts
• web/mobile/src/features/chat/useSessionWatch.ts
• web/mobile/src/features/chat/watchRelay.ts
• web/mobile/src/features/sessions/SessionRow.tsx
• web/mobile/src/lib/auth.ts
• web/mobile/tests/unit/livenessBadge.test.ts
• web/mobile/tests/unit/watchRelay.test.ts
• web/mobile/vitest.config.ts

💤 Files with no reviewable changes (1)

• web/mobile/src/features/chat/ApprovalCard.tsx

[ardaerzin]

All three verified against the code. Two were real; one I am not taking.

format_watch_frame crash — real, fixed. Reproduced it directly: _KNOWN_EVENTS is a set, so {"type": ["nope"]} raises TypeError: unhashable type: 'list' out of the membership test instead of being skipped, which tears down the SSE stream for every watcher on that channel. A malformed publish must drop its own frame, never the connection. Now gated on isinstance(event, str).

End-of-life running deletion — real, fixed, and a good catch. This is the exact mirror of a bug already fixed on the arming side: that branch refuses a turn that no longer owns the session, because acquire_running overwrites. The clearing branch called clear_running unconditionally, so a stale turn's final is_running=False beat deleted the LIVE turn's lock and then published lifecycle: ended underneath it. A watcher would see the session go idle while it was still running.

Fixed with a release_running that compares and deletes atomically (Lua, reusing the RELEASE_IF_OWNER_LUA the alive lock already uses) rather than the get-then-delete the suggestion implies, which can still race between the read and the delete. clear_running stays unconditional for the paths that legitimately evict whoever holds it: displacement (steer/cancel) and the orphan sweep.

Two regression tests added, and I verified they fail against the old code: a stale turn's end must leave the live turn's lock and publish nothing, and a turn's own end must still clear and publish.

Teardown log — not taking. The logger already carries the exception context at that call site; repeating it in the message duplicates what the handler prints.

[ardaerzin]

@coderabbitai review

[coderabbitai]

✅ Action performed

Review finished.

Note: CodeRabbit is an incremental review system and does not re-review already reviewed commits. This command is applicable only when automatic reviews are paused.

[coderabbitai]

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)

docs/design/agenta-mobile/plans/2026-07-27-mobile-approvals-steering.md (1)

14-31: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win

Update the watch-relay design status and contract.

This PR delivers GET /sessions/streams/watch as a metadata-only SSE relay. The document still states that no session watch SSE exists and schedules a cursor-based record-replay endpoint as future M3 work.

Mark the delivered relay as implemented. Describe its metadata-only revalidation contract. Keep any cursor-based record replay as a separate future enhancement.

Also applies to: 275-279

🧹 Nitpick comments (1)

web/mobile/src/features/chat/useSessionWatch.ts (1)

13-25: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value

Shorten the block comment.

This block documents routine hook behavior across 13 lines. Keep one short in-code comment. Put detailed relay behavior in external documentation if needed.

As per coding guidelines, “Keep in-code comments to at most one short line; use longer comments only for genuinely surprising constraints.”

Source: Coding guidelines

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: 15d03900-a4d3-4cdf-a2b6-a7558723c153

📥 Commits

Reviewing files that changed from the base of the PR and between c4025ac and 1c4c784.

📒 Files selected for processing (38)

• api/entrypoints/routers.py
• api/entrypoints/worker_queues.py
• api/entrypoints/worker_streams.py
• api/oss/src/apis/fastapi/sessions/router.py
• api/oss/src/apis/fastapi/sessions/watch.py
• api/oss/src/core/sessions/interactions/service.py
• api/oss/src/core/sessions/streams/service.py
• api/oss/src/dbs/redis/sessions/contract.py
• api/oss/src/dbs/redis/sessions/locks.py
• api/oss/src/dbs/redis/sessions/watch.py
• api/oss/src/tasks/asyncio/sessions/orphan_sweep.py
• api/oss/src/tasks/asyncio/sessions/records_worker.py
• api/oss/src/utils/env.py
• api/oss/tests/pytest/unit/sessions/test_heartbeat_is_current_turn.py
• api/oss/tests/pytest/unit/sessions/test_heartbeat_mirrors_existing_row.py
• api/oss/tests/pytest/unit/sessions/test_heartbeat_stale_turn_end.py
• api/oss/tests/pytest/unit/sessions/test_heartbeat_turn_handover.py
• api/oss/tests/pytest/unit/sessions/test_orphan_sweep_clears_redis.py
• api/oss/tests/pytest/unit/sessions/test_orphan_sweep_thresholds.py
• api/oss/tests/pytest/unit/sessions/test_watch_endpoint.py
• api/oss/tests/pytest/unit/sessions/test_watch_interactions_publish.py
• api/oss/tests/pytest/unit/sessions/test_watch_lifecycle_publish.py
• api/oss/tests/pytest/unit/sessions/test_watch_publish.py
• docs/design/agenta-mobile/plans/2026-07-27-mobile-approvals-steering.md
• web/mobile/src/features/chat/ApprovalDock.tsx
• web/mobile/src/features/chat/ChatHeader.tsx
• web/mobile/src/features/chat/ChatScreen.tsx
• web/mobile/src/features/chat/TurnRow.tsx
• web/mobile/src/features/chat/approvalInputSummary.ts
• web/mobile/src/features/chat/useApprovalActions.ts
• web/mobile/src/features/chat/useSessionTranscript.ts
• web/mobile/src/features/chat/useSessionWatch.ts
• web/mobile/src/features/chat/watchRelay.ts
• web/mobile/src/features/sessions/SessionRow.tsx
• web/mobile/src/lib/auth.ts
• web/mobile/tests/unit/livenessBadge.test.ts
• web/mobile/tests/unit/watchRelay.test.ts
• web/mobile/vitest.config.ts

🚧 Files skipped from review as they are similar to previous changes (31)

• api/entrypoints/routers.py
• web/mobile/src/features/chat/ChatScreen.tsx
• web/mobile/src/features/sessions/SessionRow.tsx
• web/mobile/vitest.config.ts
• web/mobile/src/features/chat/useApprovalActions.ts
• api/oss/tests/pytest/unit/sessions/test_heartbeat_mirrors_existing_row.py
• web/mobile/src/features/chat/ApprovalDock.tsx
• web/mobile/tests/unit/livenessBadge.test.ts
• web/mobile/src/features/chat/approvalInputSummary.ts
• web/mobile/src/features/chat/TurnRow.tsx
• web/mobile/src/lib/auth.ts
• api/oss/tests/pytest/unit/sessions/test_watch_interactions_publish.py
• api/oss/tests/pytest/unit/sessions/test_heartbeat_turn_handover.py
• web/mobile/src/features/chat/watchRelay.ts
• api/entrypoints/worker_queues.py
• api/oss/tests/pytest/unit/sessions/test_watch_lifecycle_publish.py
• api/oss/tests/pytest/unit/sessions/test_orphan_sweep_thresholds.py
• web/mobile/src/features/chat/useSessionTranscript.ts
• web/mobile/tests/unit/watchRelay.test.ts
• api/entrypoints/worker_streams.py
• api/oss/tests/pytest/unit/sessions/test_heartbeat_is_current_turn.py
• api/oss/src/apis/fastapi/sessions/router.py
• web/mobile/src/features/chat/ChatHeader.tsx
• api/oss/tests/pytest/unit/sessions/test_orphan_sweep_clears_redis.py
• api/oss/src/tasks/asyncio/sessions/orphan_sweep.py
• api/oss/src/tasks/asyncio/sessions/records_worker.py
• api/oss/src/dbs/redis/sessions/contract.py
• api/oss/src/core/sessions/interactions/service.py
• api/oss/tests/pytest/unit/sessions/test_watch_publish.py
• api/oss/src/utils/env.py
• api/oss/src/core/sessions/streams/service.py

[ardaerzin]

@coderabbitai review

[coderabbitai]

✅ Action performed

Review finished.

Note: CodeRabbit is an incremental review system and does not re-review already reviewed commits. This command is applicable only when automatic reviews are paused.

[ardaerzin]

@coderabbitai review

[coderabbitai]

✅ Action performed

Review finished.

Note: CodeRabbit is an incremental review system and does not re-review already reviewed commits. This command is applicable only when automatic reviews are paused.

[ardaerzin]

Log the exception on a failed teardown — real, fixed. The except block dropped the exception object, so a genuine Redis failure logged only the channel. Teardown stays best-effort; it now says what went wrong.

[coderabbitai]

Actionable comments posted: 2

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)

docs/design/agenta-mobile/plans/2026-07-27-mobile-approvals-steering.md (1)

14-21: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win

Update the plan for the implemented metadata watch relay.

GET /sessions/streams/watch now exists. It emits session-change metadata through durable Redis, and clients revalidate records through existing query paths. Update these sections to mark that relay as implemented. Keep any future cursor replay or incremental record reduction as separate work.

Also applies to: 275-279

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: 7db79f78-2543-4fb0-9a26-bf682cd0abd2

📥 Commits

Reviewing files that changed from the base of the PR and between 1c4c784 and 28906c3.

📒 Files selected for processing (38)

• api/entrypoints/routers.py
• api/entrypoints/worker_queues.py
• api/entrypoints/worker_streams.py
• api/oss/src/apis/fastapi/sessions/router.py
• api/oss/src/apis/fastapi/sessions/watch.py
• api/oss/src/core/sessions/interactions/service.py
• api/oss/src/core/sessions/streams/service.py
• api/oss/src/dbs/redis/sessions/contract.py
• api/oss/src/dbs/redis/sessions/locks.py
• api/oss/src/dbs/redis/sessions/watch.py
• api/oss/src/tasks/asyncio/sessions/orphan_sweep.py
• api/oss/src/tasks/asyncio/sessions/records_worker.py
• api/oss/src/utils/env.py
• api/oss/tests/pytest/unit/sessions/test_heartbeat_is_current_turn.py
• api/oss/tests/pytest/unit/sessions/test_heartbeat_mirrors_existing_row.py
• api/oss/tests/pytest/unit/sessions/test_heartbeat_stale_turn_end.py
• api/oss/tests/pytest/unit/sessions/test_heartbeat_turn_handover.py
• api/oss/tests/pytest/unit/sessions/test_orphan_sweep_clears_redis.py
• api/oss/tests/pytest/unit/sessions/test_orphan_sweep_thresholds.py
• api/oss/tests/pytest/unit/sessions/test_watch_endpoint.py
• api/oss/tests/pytest/unit/sessions/test_watch_interactions_publish.py
• api/oss/tests/pytest/unit/sessions/test_watch_lifecycle_publish.py
• api/oss/tests/pytest/unit/sessions/test_watch_publish.py
• docs/design/agenta-mobile/plans/2026-07-27-mobile-approvals-steering.md
• web/mobile/src/features/chat/ApprovalDock.tsx
• web/mobile/src/features/chat/ChatHeader.tsx
• web/mobile/src/features/chat/ChatScreen.tsx
• web/mobile/src/features/chat/TurnRow.tsx
• web/mobile/src/features/chat/approvalInputSummary.ts
• web/mobile/src/features/chat/useApprovalActions.ts
• web/mobile/src/features/chat/useSessionTranscript.ts
• web/mobile/src/features/chat/useSessionWatch.ts
• web/mobile/src/features/chat/watchRelay.ts
• web/mobile/src/features/sessions/SessionRow.tsx
• web/mobile/src/lib/auth.ts
• web/mobile/tests/unit/livenessBadge.test.ts
• web/mobile/tests/unit/watchRelay.test.ts
• web/mobile/vitest.config.ts

🚧 Files skipped from review as they are similar to previous changes (32)

• web/mobile/vitest.config.ts
• web/mobile/src/lib/auth.ts
• web/mobile/src/features/chat/useSessionWatch.ts
• web/mobile/src/features/chat/watchRelay.ts
• web/mobile/src/features/chat/ApprovalDock.tsx
• web/mobile/src/features/chat/TurnRow.tsx
• web/mobile/src/features/chat/ChatHeader.tsx
• api/oss/src/tasks/asyncio/sessions/records_worker.py
• api/entrypoints/worker_queues.py
• web/mobile/tests/unit/livenessBadge.test.ts
• api/oss/src/dbs/redis/sessions/contract.py
• web/mobile/src/features/chat/ChatScreen.tsx
• api/oss/tests/pytest/unit/sessions/test_orphan_sweep_thresholds.py
• api/oss/src/dbs/redis/sessions/locks.py
• web/mobile/src/features/chat/useApprovalActions.ts
• api/oss/tests/pytest/unit/sessions/test_heartbeat_stale_turn_end.py
• web/mobile/src/features/chat/approvalInputSummary.ts
• api/entrypoints/routers.py
• api/oss/src/utils/env.py
• api/oss/tests/pytest/unit/sessions/test_watch_interactions_publish.py
• api/oss/tests/pytest/unit/sessions/test_watch_lifecycle_publish.py
• web/mobile/src/features/sessions/SessionRow.tsx
• web/mobile/tests/unit/watchRelay.test.ts
• api/oss/src/tasks/asyncio/sessions/orphan_sweep.py
• api/oss/tests/pytest/unit/sessions/test_orphan_sweep_clears_redis.py
• api/oss/src/core/sessions/streams/service.py
• api/oss/tests/pytest/unit/sessions/test_heartbeat_is_current_turn.py
• api/oss/src/apis/fastapi/sessions/router.py
• api/entrypoints/worker_streams.py
• api/oss/tests/pytest/unit/sessions/test_heartbeat_mirrors_existing_row.py
• api/oss/src/core/sessions/interactions/service.py
• web/mobile/src/features/chat/useSessionTranscript.ts

[coderabbitai]

🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win

Keep failed append messages unacknowledged.

processed_ids is the acknowledgment set. This assertion accepts the failed record as processed. The caller can then acknowledge it and lose the record instead of retrying the append.

Expect no processed IDs for this batch. Update RecordsWorker.process_batch so it returns IDs only after a successful append.

Proposed test correction

-    # Failed appends stay in the existing retry path; publish adds nothing to it.
-    assert len(processed_ids) == 1
+    # Failed appends must remain unacknowledged for retry.
+    assert processed_ids == []

[ardaerzin]

@coderabbitai review

[coderabbitai]

✅ Action performed

Review finished.

Note: CodeRabbit is an incremental review system and does not re-review already reviewed commits. This command is applicable only when automatic reviews are paused.

[coderabbitai]

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)

docs/design/agenta-mobile/plans/2026-07-27-mobile-approvals-steering.md (1)

14-31: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win

Update the plan for the delivered session-watch relay.

This PR already adds GET /sessions/streams/watch. It emits metadata notifications, and clients revalidate their existing cache.

Lines 14-31 state that the endpoint does not exist. Lines 174-188 and 275-279 schedule a different cursor-and-replay endpoint. Lines 311-315 also mark the relay as pending. Update the status and contract. Keep any future cursor/replay capability as separate work. Otherwise later work can duplicate or replace this API with incompatible behavior.

Also applies to: 174-188, 275-279, 311-315

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: 79ba048b-8eb4-4db0-9576-d1fb651850ef

📥 Commits

Reviewing files that changed from the base of the PR and between 28906c3 and 090ed5d.

📒 Files selected for processing (38)

• api/entrypoints/routers.py
• api/entrypoints/worker_queues.py
• api/entrypoints/worker_streams.py
• api/oss/src/apis/fastapi/sessions/router.py
• api/oss/src/apis/fastapi/sessions/watch.py
• api/oss/src/core/sessions/interactions/service.py
• api/oss/src/core/sessions/streams/service.py
• api/oss/src/dbs/redis/sessions/contract.py
• api/oss/src/dbs/redis/sessions/locks.py
• api/oss/src/dbs/redis/sessions/watch.py
• api/oss/src/tasks/asyncio/sessions/orphan_sweep.py
• api/oss/src/tasks/asyncio/sessions/records_worker.py
• api/oss/src/utils/env.py
• api/oss/tests/pytest/unit/sessions/test_heartbeat_is_current_turn.py
• api/oss/tests/pytest/unit/sessions/test_heartbeat_mirrors_existing_row.py
• api/oss/tests/pytest/unit/sessions/test_heartbeat_stale_turn_end.py
• api/oss/tests/pytest/unit/sessions/test_heartbeat_turn_handover.py
• api/oss/tests/pytest/unit/sessions/test_orphan_sweep_clears_redis.py
• api/oss/tests/pytest/unit/sessions/test_orphan_sweep_thresholds.py
• api/oss/tests/pytest/unit/sessions/test_watch_endpoint.py
• api/oss/tests/pytest/unit/sessions/test_watch_interactions_publish.py
• api/oss/tests/pytest/unit/sessions/test_watch_lifecycle_publish.py
• api/oss/tests/pytest/unit/sessions/test_watch_publish.py
• docs/design/agenta-mobile/plans/2026-07-27-mobile-approvals-steering.md
• web/mobile/src/features/chat/ApprovalDock.tsx
• web/mobile/src/features/chat/ChatHeader.tsx
• web/mobile/src/features/chat/ChatScreen.tsx
• web/mobile/src/features/chat/TurnRow.tsx
• web/mobile/src/features/chat/approvalInputSummary.ts
• web/mobile/src/features/chat/useApprovalActions.ts
• web/mobile/src/features/chat/useSessionTranscript.ts
• web/mobile/src/features/chat/useSessionWatch.ts
• web/mobile/src/features/chat/watchRelay.ts
• web/mobile/src/features/sessions/SessionRow.tsx
• web/mobile/src/lib/auth.ts
• web/mobile/tests/unit/livenessBadge.test.ts
• web/mobile/tests/unit/watchRelay.test.ts
• web/mobile/vitest.config.ts

🚧 Files skipped from review as they are similar to previous changes (34)

• api/oss/src/utils/env.py
• api/entrypoints/worker_streams.py
• web/mobile/src/features/chat/useApprovalActions.ts
• api/oss/src/dbs/redis/sessions/contract.py
• web/mobile/src/lib/auth.ts
• web/mobile/tests/unit/livenessBadge.test.ts
• api/oss/tests/pytest/unit/sessions/test_heartbeat_is_current_turn.py
• web/mobile/vitest.config.ts
• api/entrypoints/routers.py
• api/oss/src/tasks/asyncio/sessions/records_worker.py
• web/mobile/tests/unit/watchRelay.test.ts
• api/oss/src/dbs/redis/sessions/locks.py
• api/oss/src/core/sessions/interactions/service.py
• api/entrypoints/worker_queues.py
• web/mobile/src/features/chat/TurnRow.tsx
• api/oss/src/apis/fastapi/sessions/router.py
• api/oss/src/core/sessions/streams/service.py
• api/oss/tests/pytest/unit/sessions/test_orphan_sweep_clears_redis.py
• web/mobile/src/features/chat/ChatHeader.tsx
• api/oss/src/tasks/asyncio/sessions/orphan_sweep.py
• web/mobile/src/features/chat/ApprovalDock.tsx
• api/oss/tests/pytest/unit/sessions/test_watch_interactions_publish.py
• web/mobile/src/features/chat/useSessionTranscript.ts
• api/oss/tests/pytest/unit/sessions/test_watch_lifecycle_publish.py
• web/mobile/src/features/chat/watchRelay.ts
• api/oss/tests/pytest/unit/sessions/test_heartbeat_stale_turn_end.py
• web/mobile/src/features/chat/useSessionWatch.ts
• api/oss/tests/pytest/unit/sessions/test_heartbeat_mirrors_existing_row.py
• api/oss/tests/pytest/unit/sessions/test_orphan_sweep_thresholds.py
• web/mobile/src/features/chat/ChatScreen.tsx
• api/oss/tests/pytest/unit/sessions/test_heartbeat_turn_handover.py
• api/oss/tests/pytest/unit/sessions/test_watch_publish.py
• web/mobile/src/features/chat/approvalInputSummary.ts
• web/mobile/src/features/sessions/SessionRow.tsx