[feat] Session control milestone 2: live events - #6572
Draft
mmabrouk wants to merge 81 commits into
Draft
Conversation
Publish temporary session frames to a dedicated bounded Redis Stream. Restore the durable records queue and both consumers to independent ACK and delete behavior, with count and age retention tests. Claude-Session: https://claude.ai/code/session_0164kzT6ttwpBtzvcDC6YzYk
Reject live frames above 64 KiB during request validation and immediately before Redis publication. Apply the same limit to declared HTTP request bodies and cover each boundary with regression tests. Claude-Session: https://claude.ai/code/session_0164kzT6ttwpBtzvcDC6YzYk
Mark the identity-scoped session event stream as no-store so intermediaries and browsers do not retain live session content. Pin the response header in a route regression test. Claude-Session: https://claude.ai/code/session_0164kzT6ttwpBtzvcDC6YzYk
Document one HTTP ingress with separate durable-record and disposable-frame Redis Streams. Record the deployment-wide frame bounds and clarify that clients receive the global shared-reader capability. Claude-Session: https://claude.ai/code/session_0164kzT6ttwpBtzvcDC6YzYk
Validate session live events with the shared diagnostic helper and cover invalid-frame logging. Shorten the three reviewed live-relay comments without changing behavior. Claude-Session: https://claude.ai/code/session_0164kzT6ttwpBtzvcDC6YzYk
Require each live execution to start at frame zero and advance contiguously. Suppress the preview after a gap and refresh durable session records. Claude-Session: https://claude.ai/code/session_0164kzT6ttwpBtzvcDC6YzYk
Restore the base branch's 100,000-entry approximate trim for durable record publication. Pin the flag-off Redis stream arguments in a regression test. Claude-Session: https://claude.ai/code/session_0164kzT6ttwpBtzvcDC6YzYk
Initialize durable consumers before the optional live relay. Log relay setup failures and continue running the durable stream consumers. Claude-Session: https://claude.ai/code/session_0164kzT6ttwpBtzvcDC6YzYk
Document the live-frame relay and gap handling as the shipped contract. Mark sender handoff, durable event envelopes, sequences, watermarks, and Postgres replay as later target behavior. Claude-Session: https://claude.ai/code/session_0164kzT6ttwpBtzvcDC6YzYk
The two Postgres-backed records tests declared an async autouse fixture with the plain pytest decorator, which pytest-asyncio 1.x does not collect, so both files errored at setup before any query ran. With the pytest_asyncio decorator both files run: 3 passed against the proof stack database. Claude-Session: https://claude.ai/code/session_01GAqSs7fw6QRi2n1ZJ2tmAV
Await replay queue capacity so a finite durable backlog cannot be mistaken for a live producer outrunning its reader. Keep the existing drop-and-reconnect policy for live frames and cover replay larger than the configured buffer. Claude-Session: https://claude.ai/code/session_0164kzT6ttwpBtzvcDC6YzYk
Route live frames and committed durable events through one bounded relay-stream append path while leaving record ingest retention unchanged. Claude-Session: https://claude.ai/code/session_0164kzT6ttwpBtzvcDC6YzYk
Acquire per-session sequence cursors in a stable order while preserving record order inside each session. Leave every project batch message pending when its transactional append fails. Claude-Session: https://claude.ai/code/session_0164kzT6ttwpBtzvcDC6YzYk
Abort shared-sender invocations that neither return nor acknowledge the turn within fifteen seconds. Pre-acceptance hangs now enter the existing retryable failure-card path while accepted disconnects and legacy sends keep their prior behavior. Claude-Session: https://claude.ai/code/session_0164kzT6ttwpBtzvcDC6YzYk
Coalesce shared live-frame envelopes by time, count, and serialized size while preserving frame order. Drain the independent live publisher at turn completion and keep the flag-off path inert. Claude-Session: https://claude.ai/code/session_0164kzT6ttwpBtzvcDC6YzYk
Accept ordered live-frame batches while retaining the single durable-record contract. Exempt only platform-runner record ingest from the organisation plan bucket and keep browser and other runner routes throttled. Claude-Session: https://claude.ai/code/session_0164kzT6ttwpBtzvcDC6YzYk
Refresh session-stream liveness from record notifications for legacy observers without raising the idle polling cadence. Keep flag-off presentation independent of the shared-reader snapshot and preserve connected-reader activity on desktop and mobile. Claude-Session: https://claude.ai/code/session_0164kzT6ttwpBtzvcDC6YzYk
Allow the already-known loopback OTLP fixture under its rebased commit fingerprint. The old fingerprint remains for any history that still references the pre-rebase commit. Claude-Session: https://claude.ai/code/session_0164kzT6ttwpBtzvcDC6YzYk
State the effective age cap separately from count capacity, and document the split between per-event deduplication order and the reconnect watermark. Claude-Session: https://claude.ai/code/session_0164kzT6ttwpBtzvcDC6YzYk
feat(sessions): relay live frames to every authorized reader behind AGENTA_SESSIONS_SHARED_READER
feat(sessions): durable reconnect with a per-session sequence, snapshot, and replay
feat(sessions): let the sender read from the shared path and detach the run from its request
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
Contributor
|
Important Draft PR not reviewedDraft PRs are not automatically reviewed by default.
To automatically review draft PRs, update your CodeRabbit configuration: reviews:
auto_review:
drafts: true📝 SummarySummary by CodeRabbit
WalkthroughChangesThe pull request adds a shared session live-preview flow. It introduces sequenced records, durable replay, live-frame ingestion, Redis relay workers, SSE events, detached runner execution, and web client preview state. Tests and deployment configuration cover the new behavior. Shared session backend
Web shared-reader flow
Sequence Diagram(s)sequenceDiagram
participant Runner
participant SessionAPI
participant Redis
participant RelayWorker
participant Browser
Runner->>SessionAPI: POST live frames
SessionAPI->>Redis: Append live-frame messages
RelayWorker->>Redis: Consume live-frame stream
RelayWorker->>Browser: Publish session event
Browser->>SessionAPI: Fetch snapshot and replay
SessionAPI-->>Browser: Return bounded transcript and events
Estimated code review effort: 5 (Critical) | ~120 minutes Merge Risk: 🟠 High · up to Accepted turns can be misclassified, stalled relay requests can delay completion, malformed records can repeatedly block processing, live-frame authorization may be insufficient, and deployment can block record writes. These should be resolved before merge. 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
Full details: Docstring CoverageExplanation Docstring coverage is 8.67% which is insufficient. The required threshold is 60.00%. Docstring coverage is scoped to functions touched by this diff. Analyzed 173 functions across 50 files. (82 skipped: 19 unsupported, 63 over the file limit.) ✨ Finishing Touches 💡 1📝 Generate docstrings 💡
🧪 Generate unit tests (beta)
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. Comment |
Member
Author
|
@coderabbitai review |
Contributor
✅ Action performedReview finished.
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 13
🧹 Nitpick comments (11)
api/oss/src/core/sessions/records/service.py (1)
338-338: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAnnotate the
get_events_afterreturn type.
get_events_afterreturnsSessionDurableEventsReplay, but the signature declares no return type. Every other method inRecordsServicedeclares its DTO. Add the annotation so the service contract is explicit for callers such as the live relay.♻️ Proposed change
- ): + ) -> SessionDurableEventsReplay:As per coding guidelines: "Service methods must return typed DTOs (Pydantic
BaseModelsubclasses), not raw dicts, tuples, orAny."Source: Coding guidelines
api/oss/tests/pytest/integration/sessions/test_records_replay_postgres.py (1)
26-26: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winThe duplicated
_fresh_analytics_enginefixture drops the engine without closing it. Both files copy the same fixture. Each one setsengine_module._analytics_engine = Noneat setup, so an engine created by an earlier test keeps its connection pool open for the rest of the run. Extract one shared fixture that closes the previous engine before it clears the global.
api/oss/tests/pytest/integration/sessions/test_records_replay_postgres.py#L26-L26: closeengine_module._analytics_enginewhen it is not None, then set it to None.api/oss/tests/pytest/integration/sessions/test_records_snapshot_postgres.py#L24-L24: replace the local fixture with the shared one instead of repeating the reset.api/oss/tests/pytest/unit/migrations/test_tracing_session_sequence_chain.py (1)
20-31: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winScan the versions directory so the head check can actually fail.
The test reads only the two named files, so
headsis computed over a hand-picked subset. If a later migration also setsdown_revision = "oss000000005", the chain gains a second head and Alembic refuses to upgrade, but this test still passes. Buildlinksfrom every file inVERSIONS_DIRand keep the two explicit link assertions.♻️ Proposed change
def test_tracing_chain_has_one_head_with_watchdog_migration(): - watchdog_migration = VERSIONS_DIR / "oss000000005_add_records_quarantined_at.py" - session_migration = VERSIONS_DIR / "oss000000006_add_session_sequence_cursors.py" - links = dict(map(_revision_link, (watchdog_migration, session_migration))) + links = dict(map(_revision_link, sorted(VERSIONS_DIR.glob("*.py")))) heads = set(links) - set(links.values()) - assert links == { - "oss000000005": "oss000000004", - "oss000000006": "oss000000005", - } + assert links["oss000000005"] == "oss000000004" + assert links["oss000000006"] == "oss000000005" assert heads == {"oss000000006"}api/oss/tests/pytest/unit/sessions/test_durable_events.py (1)
58-58: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAlign the test name with what it asserts.
The name claims the six direct event types are accepted, but the test supplies only
execution.startedand one unknown type. Either add records for the remaining direct types or rename the test to describe the unknown-type behavior it verifies.api/oss/src/apis/fastapi/sessions/live_events.py (1)
155-157: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winReuse the parsed payload instead of parsing each frame twice.
Line 145 already decodes
message.get("data")intopayload.format_live_framedecodes the same bytes again for every live frame. Accept the decoded object so the hot relay path parses once.♻️ Proposed change
-def format_live_frame(raw: Any) -> Optional[str]: - try: - payload = json.loads(raw) - except (ValueError, TypeError): - return None - if not isinstance(payload, dict) or payload.get("kind") not in {"frame", "event"}: +def format_live_frame(payload: Any) -> Optional[str]: + if not isinstance(payload, dict) or payload.get("kind") not in {"frame", "event"}: return None return f"data: {json.dumps(payload)}\n\n"- frame = format_live_frame(message.get("data")) + frame = format_live_frame(payload) if frame is not None: enqueue(frame)api/oss/databases/postgres/migrations/tracing_oss/versions/oss000000006_add_session_sequence_cursors.py (1)
45-50: 🩺 Stability & Availability | 🔵 TrivialUse a concurrent index build for large
recordstables.
op.create_indexemits standardCREATE UNIQUE INDEX, which takes aShareLockand can block record ingest during the build.tracing_ossruns migrations in transactions, soCREATE INDEX CONCURRENTLYrequiresautocommit_block().♻️ Concurrent index build
- op.create_index( - "ux_records_session_id_sequence", - "records", - ["project_id", "session_id", "sequence"], - unique=True, - ) + with op.get_context().autocommit_block(): + op.create_index( + "ux_records_session_id_sequence", + "records", + ["project_id", "session_id", "sequence"], + unique=True, + postgresql_concurrently=True, + )api/oss/src/dbs/postgres/sessions/records/dao.py (1)
133-151: 🚀 Performance & Scalability | 🔵 Trivial | 🏗️ Heavy liftConsider allocating a sequence block per session instead of three statements per record.
With
sequence_writesenabled, this loop replaces the single batched upsert with three statements per record: the conditional insert, the cursor upsert, and the sequence update. A 50-record batch becomes about 150 round trips inside one transaction, and this is the durable transcript ingest path that the docstring at Line 123 was written to keep batched.An alternative keeps one cursor upsert per session: group
ordered_valuesbysession_id, advancelatest_sequenceby the group size in oneon_conflict_do_update, then assign the returned block to the group's records in the batched insert. Records that conflict keep their existing sequence, as they do today.This is deferable while the flag is off by default, but it is worth doing before the flag is enabled broadly.
web/packages/agenta-entities/src/session/core/schema.ts (1)
213-218: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winReduce new code comments to one short line. Preserve only the non-obvious invariant.
web/packages/agenta-entities/src/session/core/schema.ts#L213-L218: reduce the durable-event explanation to one short line.web/packages/agenta-entities/src/session/state/livePreview.ts#L14-L18: reduce the live-preview ownership explanation to one short line.web/oss/src/components/AgentChatSlice/hooks/useSessionHydration.ts#L208-L211: reduce the run-error rationale to one short line.web/oss/src/components/AgentChatSlice/hooks/useSessionRecordsWatch.ts#L71-L76: reduce the lifecycle rationale to one short line.
As per coding guidelines: “Hard rule. At most ONE short line per comment.”Source: Coding guidelines
web/packages/agenta-chat/src/hooks/useAgentConversation.ts (2)
821-822: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winUpdate the declared
revalidatesignature to match the implementation.
revalidatenow accepts an optionalSessionTranscriptand resolves to a boolean. The publicAgentConversation.revalidatetype still declares() => void(Line 196). TypeScript accepts the assignment, so consumers that go through the interface silently lose the parameter and the result.♻️ Proposed interface change (outside the selected range)
/** Re-fetch the durable records and adopt the server transcript under the same guards as * revalidate-on-open. Returns whether the local transcript now covers the candidate. */ revalidate: (transcript?: SessionTranscript) => Promise<boolean>
278-281: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueShorten the added in-code comments.
The coding guidelines set a hard rule of at most one short line per comment for
web/**/*.{ts,tsx}. The added comments at Lines 278-281, 500-503, and 765-769 each span several lines. Move the extended rationale to the design docs and keep one line here.As per coding guidelines: "Hard rule. At most ONE short line per comment."
Source: Coding guidelines
web/packages/agenta-chat/src/model/livePreview.ts (1)
61-63: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueShorten these comments to one line each.
The coding guidelines set a hard rule of at most one short line per comment for
web/**/*.{ts,tsx}. The block at Lines 61-63 spans three lines, and the block at Lines 73-74 spans two lines. Reduce each to a single short line.As per coding guidelines: "Hard rule. At most ONE short line per comment."
Also applies to: 73-74
Source: Coding guidelines
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository YAML (base), Organization UI (inherited)
Review profile: CHILL
Plan: Team
Run ID: c01b783a-7db7-4362-9762-74faa9425bef
⛔ Files ignored due to path filters (11)
web/packages/agenta-api-client/src/generated/api/resources/sessions/client/Client.tsis excluded by!**/generated/**web/packages/agenta-api-client/src/generated/api/resources/sessions/client/requests/GetSessionSnapshotRequest.tsis excluded by!**/generated/**web/packages/agenta-api-client/src/generated/api/resources/sessions/client/requests/SessionRecordQueryRequest.tsis excluded by!**/generated/**web/packages/agenta-api-client/src/generated/api/resources/sessions/client/requests/index.tsis excluded by!**/generated/**web/packages/agenta-api-client/src/generated/api/types/SessionRecord.tsis excluded by!**/generated/**web/packages/agenta-api-client/src/generated/api/types/SessionRecordsQueryResponse.tsis excluded by!**/generated/**web/packages/agenta-api-client/src/generated/api/types/SessionRecordsReadState.tsis excluded by!**/generated/**web/packages/agenta-api-client/src/generated/api/types/SessionSnapshotPending.tsis excluded by!**/generated/**web/packages/agenta-api-client/src/generated/api/types/SessionSnapshotResponse.tsis excluded by!**/generated/**web/packages/agenta-api-client/src/generated/api/types/SessionTranscriptWindowing.tsis excluded by!**/generated/**web/packages/agenta-api-client/src/generated/api/types/index.tsis excluded by!**/generated/**
📒 Files selected for processing (132)
.gitleaksignoreapi/ee/src/middlewares/throttling.pyapi/ee/tests/pytest/unit/test_throttling.pyapi/entrypoints/worker_streams.pyapi/oss/databases/postgres/migrations/tracing_oss/versions/oss000000006_add_session_sequence_cursors.pyapi/oss/src/apis/fastapi/sessions/live_events.pyapi/oss/src/apis/fastapi/sessions/models.pyapi/oss/src/apis/fastapi/sessions/router.pyapi/oss/src/apis/fastapi/sessions/utils.pyapi/oss/src/core/sessions/records/dtos.pyapi/oss/src/core/sessions/records/events.pyapi/oss/src/core/sessions/records/interfaces.pyapi/oss/src/core/sessions/records/service.pyapi/oss/src/core/sessions/records/streaming.pyapi/oss/src/core/sessions/streams/dtos.pyapi/oss/src/dbs/postgres/sessions/records/dao.pyapi/oss/src/dbs/postgres/sessions/records/dbas.pyapi/oss/src/dbs/postgres/sessions/records/dbes.pyapi/oss/src/dbs/postgres/sessions/records/mappings.pyapi/oss/src/dbs/redis/sessions/contract.pyapi/oss/src/tasks/asyncio/sessions/live_relay_worker.pyapi/oss/src/tasks/asyncio/sessions/records_worker.pyapi/oss/src/utils/env.pyapi/oss/tests/pytest/integration/sessions/test_records_replay_postgres.pyapi/oss/tests/pytest/integration/sessions/test_records_sequence_postgres.pyapi/oss/tests/pytest/integration/sessions/test_records_snapshot_postgres.pyapi/oss/tests/pytest/unit/migrations/test_tracing_session_sequence_chain.pyapi/oss/tests/pytest/unit/sessions/test_durable_events.pyapi/oss/tests/pytest/unit/sessions/test_live_frame_ingest.pyapi/oss/tests/pytest/unit/sessions/test_live_relay.pyapi/oss/tests/pytest/unit/sessions/test_records_mapping_upsert.pyapi/oss/tests/pytest/unit/sessions/test_records_worker_batching.pyapi/oss/tests/pytest/unit/sessions/test_session_snapshot.pyapi/oss/tests/pytest/unit/sessions/test_worker_streams_startup.pydocs/design/session-control-and-live-events/contracts/events.mddocs/design/session-control-and-live-events/decisions.mddocs/design/session-control-and-live-events/live-frame-envelope.mdhosting/docker-compose/ee/docker-compose.dev.ymlhosting/docker-compose/ee/docker-compose.gh.local.ymlhosting/docker-compose/ee/docker-compose.gh.ymlhosting/docker-compose/ee/env.ee.dev.examplehosting/docker-compose/ee/env.ee.gh.examplehosting/docker-compose/oss/docker-compose.dev.ymlhosting/docker-compose/oss/docker-compose.gh.local.ymlhosting/docker-compose/oss/docker-compose.gh.ssl.ymlhosting/docker-compose/oss/docker-compose.gh.ymlhosting/docker-compose/oss/env.oss.dev.examplehosting/docker-compose/oss/env.oss.gh.examplehosting/kubernetes/helm/templates/runner-deployment.yamlhosting/kubernetes/helm/values.schema.jsonhosting/kubernetes/helm/values.yamlhosting/railway/oss/scripts/configure.shhosting/railway/oss/template/template.jsonsdks/python/agenta/sdk/agents/adapters/local.pysdks/python/agenta/sdk/agents/adapters/sandbox_agent.pysdks/python/agenta/sdk/agents/dtos.pysdks/python/agenta/sdk/agents/handler.pysdks/python/agenta/sdk/agents/interfaces.pysdks/python/agenta/sdk/agents/utils/ts_runner.pysdks/python/agenta/sdk/agents/utils/wire.pysdks/python/agenta/sdk/agents/wire_models.pysdks/python/agenta/sdk/models/workflows.pysdks/python/oss/tests/pytest/integration/agents/_fake_runner_backend.pysdks/python/oss/tests/pytest/unit/agents/conftest.pysdks/python/oss/tests/pytest/unit/agents/test_agent_composition_seam.pysdks/python/oss/tests/pytest/unit/agents/test_redaction_scope.pysdks/python/oss/tests/pytest/unit/agents/test_wire_contract.pysdks/python/oss/tests/pytest/unit/test_batch_fold_stream_contract_routing.pysdks/python/oss/tests/pytest/unit/test_invoke_real_handlers_negotiation_routing.pysdks/python/oss/tests/pytest/unit/test_workflow_request_flags_running.pyservices/oss/tests/pytest/unit/agent/conftest.pyservices/runner/src/protocol.tsservices/runner/src/server.tsservices/runner/src/sessions/live-frames.tsservices/runner/src/sessions/persist.tsservices/runner/tests/unit/live-frames.test.tsservices/runner/tests/unit/server.test.tsservices/runner/tests/unit/wire-contract.test.tsweb/mobile/src/features/chat/ChatScreen.tsxweb/mobile/src/features/chat/LiveConversation.tsxweb/mobile/src/features/chat/transcriptAdoption.tsweb/mobile/src/features/chat/turnStatus.tsweb/mobile/src/features/chat/useSessionTranscript.tsweb/mobile/src/features/chat/useSessionWatch.tsweb/mobile/tests/unit/transcriptAdoption.test.tsweb/mobile/tests/unit/turnStatus.test.tsweb/oss/src/components/AgentChatSlice/AgentConversation.tsxweb/oss/src/components/AgentChatSlice/components/AgentComposerDock.tsxweb/oss/src/components/AgentChatSlice/components/AgentMessage.runError.test.tsxweb/oss/src/components/AgentChatSlice/hooks/useAgentChatSession.test.tsweb/oss/src/components/AgentChatSlice/hooks/useAgentChatSession.tsweb/oss/src/components/AgentChatSlice/hooks/useSessionHydration.livePreview.test.tsxweb/oss/src/components/AgentChatSlice/hooks/useSessionHydration.tsweb/oss/src/components/AgentChatSlice/hooks/useSessionRecordsWatch.tsweb/oss/src/components/AgentChatSlice/state/liveness.test.tsweb/oss/src/components/AgentChatSlice/state/liveness.tsweb/packages/agenta-chat/src/assets/loadSession.tsweb/packages/agenta-chat/src/components/ConnectionWarningStrip.tsxweb/packages/agenta-chat/src/components/RunningElsewhereStrip.tsxweb/packages/agenta-chat/src/components/SessionHistoryNotice.tsxweb/packages/agenta-chat/src/components/index.tsweb/packages/agenta-chat/src/hooks/index.tsweb/packages/agenta-chat/src/hooks/useAgentChatQueue.tsweb/packages/agenta-chat/src/hooks/useAgentConversation.tsweb/packages/agenta-chat/src/hooks/useSessionLivePreview.tsweb/packages/agenta-chat/src/model/durableEvents.tsweb/packages/agenta-chat/src/model/error.tsweb/packages/agenta-chat/src/model/index.tsweb/packages/agenta-chat/src/model/livePreview.tsweb/packages/agenta-chat/src/state/sessionEphemera.tsweb/packages/agenta-chat/src/transport/AgentChatTransport.tsweb/packages/agenta-chat/src/transport/index.tsweb/packages/agenta-chat/src/transport/sessionLiveEvents.tsweb/packages/agenta-chat/tests/unit/assets/loadSession.test.tsweb/packages/agenta-chat/tests/unit/hooks/useAgentChatQueue.test.tsweb/packages/agenta-chat/tests/unit/hooks/useAgentConversation.test.tsweb/packages/agenta-chat/tests/unit/hooks/useSessionLivePreview.test.tsweb/packages/agenta-chat/tests/unit/hooks/useSessionLivePreview.test.tsxweb/packages/agenta-chat/tests/unit/model/durableEvents.test.tsweb/packages/agenta-chat/tests/unit/model/error.test.tsweb/packages/agenta-chat/tests/unit/model/livePreview.test.tsweb/packages/agenta-chat/tests/unit/transport/AgentChatTransport.test.tsweb/packages/agenta-chat/tests/unit/transport/sessionLiveEvents.test.tsweb/packages/agenta-entities/src/session/api/api.tsweb/packages/agenta-entities/src/session/core/schema.tsweb/packages/agenta-entities/src/session/index.tsweb/packages/agenta-entities/src/session/state/livePreview.tsweb/packages/agenta-playground/src/agentChat.tsweb/packages/agenta-playground/src/state/execution/agentRequest.tsweb/packages/agenta-playground/src/state/execution/index.tsweb/packages/agenta-playground/tests/unit/agentRequest.test.tsweb/storybook/stories/domain/SessionHistoryNotice.stories.tsx
Included review availability: Your plan provides up to 8 included reviews per hour; 7 remain after this review.
`_direct_event` read `payload` straight out of `record.attributes`, an open dict filled from the ingest wire. A record whose `payload` was a string, a list or a number reached `payload.get(...)` and raised `AttributeError` outside the projection's try block. That propagated into `RecordsWorker.process_batch` after the batch's rows were already committed, so the message ids were never acknowledged and the same poison record came back on every redelivery. A non-dict payload now reads exactly like a missing one. Reported by CodeRabbit on #6572. Claude-Session: https://claude.ai/code/session_0164kzT6ttwpBtzvcDC6YzYk
`XADD` ran an exact `MAXLEN` trim on every live relay message, and `trim_live_stream` ran an exact `XTRIM MINID`. Exact trimming is O(N) in the evicted entries and sits on the producer's hot path. The stream carries disposable frames, so a few extra entries per listpack cost nothing. The record stream already used approximate trimming. Reported by CodeRabbit on #6572. Claude-Session: https://claude.ai/code/session_0164kzT6ttwpBtzvcDC6YzYk
The fixture replaced `_analytics_engine` with `None` without closing it, so an engine an earlier test module created kept its connection pool open for the rest of the run. Teardown only closed the engine the fixture itself created. Reported by CodeRabbit on #6572. Claude-Session: https://claude.ai/code/session_0164kzT6ttwpBtzvcDC6YzYk
The coding guidelines call for cursor pagination through the shared `Windowing` type. `SessionTranscriptWindowing` is a deliberate exception, and the reason now sits next to the model instead of in a review thread. Reported by CodeRabbit on #6572. Claude-Session: https://claude.ai/code/session_0164kzT6ttwpBtzvcDC6YzYk
`session-accepted` switches the client to shared delivery: live text stops arriving on the invoke stream and starts arriving on the session events route. The runner emitted it at the top of the request, before the attachment-count check and before the first heartbeat answered the admission question. A turn the runner was about to refuse therefore moved the client onto a channel that would serve it nothing. Acceptance now rides the same admission path as the `turn` event, so a refused turn tells the client only what it already told it: a `session_turn_in_use` error and a failed terminal result. Reported by CodeRabbit on #6572. Claude-Session: https://claude.ai/code/session_0164kzT6ttwpBtzvcDC6YzYk
`postFrames` had no application-level timeout. An ingest endpoint that stalled before response headers left `pump()` pending, and `persist.ts` waits on `whenIdle()` in `flush()`, so a stalled socket delayed turn completion for as long as it stayed open. The POST now carries a 5 s abort signal and a timed-out batch counts as dropped, like any other send failure. `postTimeoutMs` is the test seam. Reported by CodeRabbit on #6572. Claude-Session: https://claude.ai/code/session_0164kzT6ttwpBtzvcDC6YzYk
`busyRef` folds `acceptedRunPending` into `busy`, and the transcript adoption guard read it. A shared-delivered turn renders from the live frames, not from a local stream, so this made `shouldAdoptServerTranscript` refuse for the whole accepted window. `revalidate` could then neither adopt nor confirm a candidate ahead of the client's watermark, and `hydrateAndOpen` only opens the events stream once one of the two succeeds. A shared turn whose stream dropped mid-run therefore reconnected on a growing backoff and never came back until it settled. The adoption guard now reads a ref that tracks only a stream this client renders. `busyRef` is unchanged for the preserve-at-unmount verdict, regenerate and rewind, which do want the accepted window. Reported by CodeRabbit on #6572. Claude-Session: https://claude.ai/code/session_0164kzT6ttwpBtzvcDC6YzYk
The schema transform always emits `sequence`, so both fixtures were annotated as `SessionRecord` while missing a required property. The package's `types:check` covers `src/` only, so this never failed a build, but the annotation was wrong. Reported by CodeRabbit on #6572. Claude-Session: https://claude.ai/code/session_0164kzT6ttwpBtzvcDC6YzYk
Three doubles accepted the flag and dropped it. The fake runner backend now hands it to `request_to_wire`, so the integration transport test drives the real detached payload, and both unit conftests record it. The service handler gains a direct assertion: nothing else in that package pinned `flags.detached` at the backend boundary, so a handler that stopped forwarding it passed every other invoke test. Reported by CodeRabbit on #6572. Claude-Session: https://claude.ai/code/session_0164kzT6ttwpBtzvcDC6YzYk
The events contract still told clients that the route sends no durable events, sequences,
watermarks or replayed rows, and kept the whole durable-event section under a "not shipped"
heading in the future tense. Milestone 2 shipped replay over `GET /sessions/{id}/events`
and browser fan-out. The decisions log now marks fan-out delivered and records the
approximate trimming, leaving redaction as the open item under O-004.
Reported by CodeRabbit on #6572.
Claude-Session: https://claude.ai/code/session_0164kzT6ttwpBtzvcDC6YzYk
…abbit [fix] Address CodeRabbit review for milestone 2 live events
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Context
Milestone 1 is already part of
release/v0.115.0. This branch assembles milestone 2 on that release baseline: live session event relay, durable reconnect and replay, and shared sender recovery across tabs.The new behavior remains dark by default.
AGENTA_SESSIONS_SHARED_READERandAGENTA_SESSIONS_SEQUENCE_WRITESboth default to off.What this branch contains
In merge order:
Tests
pnpm lint-fixpassed all 25 tasks.~/agenta-qa-evidence/2026-09-05-m2-proofs/.What to QA
https://claude.ai/code/session_0164kzT6ttwpBtzvcDC6YzYk