[feat] Session control milestone 2: live events - #6572
Merged
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
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
Create and drop the records sequence index concurrently from Alembic autocommit blocks so populated tracing tables remain writable during the index operation. Cover empty and populated-table upgrades and downgrades in a disposable tracing database. Claude-Session: https://claude.ai/code/session_0164kzT6ttwpBtzvcDC6YzYk
[fix] Build the records sequence index concurrently
mmabrouk
marked this pull request as ready for review
September 5, 2026 17:16
Contributor
📘 Docs preview
This comment updates in place on every push. |
Contributor
There was a problem hiding this comment.
Actionable comments posted: 3
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository YAML (base), Organization UI (inherited)
Review profile: CHILL
Plan: Team
Run ID: 97793966-7233-44d4-9573-f1c46989d36c
📒 Files selected for processing (22)
api/oss/databases/postgres/migrations/tracing_oss/versions/oss000000006_add_session_sequence_cursors.pyapi/oss/src/apis/fastapi/sessions/models.pyapi/oss/src/core/sessions/records/events.pyapi/oss/src/core/sessions/records/streaming.pyapi/oss/tests/pytest/integration/sessions/test_records_sequence_postgres.pyapi/oss/tests/pytest/unit/migrations/test_tracing_session_sequence_migration.pyapi/oss/tests/pytest/unit/sessions/test_durable_events.pyapi/oss/tests/pytest/unit/sessions/test_live_frame_ingest.pydocs/design/session-control-and-live-events/contracts/events.mddocs/design/session-control-and-live-events/decisions.mdsdks/python/oss/tests/pytest/integration/agents/_fake_runner_backend.pysdks/python/oss/tests/pytest/unit/agents/conftest.pyservices/oss/tests/pytest/unit/agent/conftest.pyservices/oss/tests/pytest/unit/agent/test_invoke_handler.pyservices/runner/src/server.tsservices/runner/src/sessions/live-frames.tsservices/runner/tests/unit/live-frames.test.tsservices/runner/tests/unit/session-admission.test.tsweb/packages/agenta-chat/src/hooks/useAgentConversation.tsweb/packages/agenta-chat/tests/unit/assets/loadSession.test.tsweb/packages/agenta-chat/tests/unit/hooks/useAgentConversation.test.tsweb/packages/agenta-chat/tests/unit/hooks/useSessionLivePreview.test.tsx
🚧 Files skipped from review as they are similar to previous changes (2)
- sdks/python/oss/tests/pytest/unit/agents/conftest.py
- api/oss/src/apis/fastapi/sessions/models.py
Included review availability: Your plan provides up to 8 included reviews per hour; 6 remain after this review.
Comment on lines
+66
to
+68
| const response = await fetch(`${apiBase()}/sessions/records/ingest`, { | ||
| method: "POST", | ||
| signal: AbortSignal.timeout(timeoutMs), |
Contributor
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -eu
printf '%s\n' '--- live-frames.ts ---'
sed -n '1,95p' services/runner/src/sessions/live-frames.ts
printf '%s\n' '--- apiBase.ts ---'
sed -n '1,55p' services/runner/src/apiBase.ts
printf '%s\n' '--- runner URL/TLS configuration references ---'
rg -n --glob '!node_modules' --glob '!dist' 'AGENTA_API_INTERNAL_URL|AGENTA_API_URL|api:8000|https://|mTLS|mtls|rejectUnauthorized' services/runner docker-compose.yml docker-compose.yaml .github 2>/dev/null | head -160Repository: Agenta-AI/agenta
Length of output: 23205
🏁 Script executed:
#!/bin/bash
set -eu
printf '%s\n' '--- server dispatch and transport guidance ---'
sed -n '1325,1365p' services/runner/src/server.ts
sed -n '1,90p' services/runner/src/engines/sandbox_agent/runtime-policy.ts
printf '%s\n' '--- runner configuration ---'
sed -n '235,290p' services/runner/src/config/runner-config.ts
printf '%s\n' '--- live-frame callers and auth construction ---'
rg -n -C 4 'new LiveFramePublisher|LiveFramePublisher\\(|auth:|postFrames\\(' services/runner/src services/runner/tests/unit/live-frames.test.ts | head -220Repository: Agenta-AI/agenta
Length of output: 8219
Sensitive Data Exposure (CWE-319): Cleartext Transmission of Sensitive Information
Exploitability: Difficult
Require encrypted transport or an explicitly trusted network before sending the API credential.
apiBase() falls back to the documented cleartext http://api:8000, and this request sends authorization: auth(). Require HTTPS or mTLS when the runner-to-API network is not fully isolated and trusted.
Comment on lines
+294
to
+295
| // Only a stream THIS client renders. A shared-delivered turn renders from the live frames, | ||
| // so the durable snapshot behind them stays adoptable — see the adoption guard below. |
Contributor
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Shorten the added comments.
The Web rule permits at most one short line per comment.
web/packages/agenta-chat/src/hooks/useAgentConversation.ts#L294-L295: reduce this rationale to one short line.web/packages/agenta-chat/src/hooks/useAgentConversation.ts#L491-L494: reduce this rationale to one short line.web/packages/agenta-chat/tests/unit/hooks/useAgentConversation.test.ts#L722-L728: replace this block with one short test rationale.
As per coding guidelines: “Hard rule. At most ONE short line per comment.”
📍 Affects 2 files
web/packages/agenta-chat/src/hooks/useAgentConversation.ts#L294-L295(this comment)web/packages/agenta-chat/src/hooks/useAgentConversation.ts#L491-L494web/packages/agenta-chat/tests/unit/hooks/useAgentConversation.test.ts#L722-L728
Source: Coding guidelines
Comment on lines
+722
to
+728
| /** | ||
| * An accepted shared turn is NOT a local stream. Its content arrives on the session events | ||
| * channel, so the durable snapshot behind those frames stays adoptable — and it has to be, | ||
| * because `hydrateAndOpen` only opens the events stream once `revalidate` adopts or confirms | ||
| * the bounded transcript. Treating `acceptedRunPending` as busy refused both, so a shared turn | ||
| * whose stream dropped mid-run reconnected forever and never came back. | ||
| */ |
Contributor
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Reduce this block comment to one short line.
Lines 722-728 exceed the one-line comment limit. Keep only the durable-transcript invariant.
As per coding guidelines: “Hard rule. At most ONE short line per comment.”
Source: Coding guidelines
Contributor
Railway Preview Environment
|
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