Repository navigation
feat(chat-runtime): standalone session runtime — journal, live sessions, subscriptions - #6167
Merged
Merged
Conversation
Contributor
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughThe PR adds the standalone ChangesChat runtime
Estimated code review effort: 5 (Critical) | ~120 minutes Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ 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 |
Contributor
Greptile SummaryThe PR introduces a standalone chat runtime with SQLite-backed journals, live harness sessions, command deduplication, replay subscriptions, delta coalescing, and a fake integration harness.
Confidence Score: 4/5The session-creation rollback and adapter-pump failure paths need to be fixed before merging because they can leave unusable persisted sessions and unhandled asynchronous failures. Session persistence currently precedes fallible harness initialization without cleanup, while async adapter failures reject an unobserved pump and leave registered session state stale. Files Needing Attention: packages/chat-runtime/src/commands/commands.ts, packages/chat-runtime/src/sessions/liveSession.ts, packages/chat-runtime/src/sessions/registry.ts
|
| Filename | Overview |
|---|---|
| packages/chat-runtime/src/commands/commands.ts | Adds validated, deduplicated command APIs, but session creation can persist orphan rows when live harness creation fails. |
| packages/chat-runtime/src/sessions/liveSession.ts | Implements adapter pumping and FIFO prompts, but adapter-stream rejection is unobserved and leaves failed sessions stale. |
| packages/chat-runtime/src/sessions/registry.ts | Owns live adapters and disposal; its creation order contributes to partial session initialization. |
| packages/chat-runtime/src/journal/journal.ts | Adds transactional durable appends and projection rebuilding with coherent sequence and queue tracking. |
| packages/chat-runtime/src/journal/replay.ts | Adds epoch-qualified replay and backward pagination with explicit reset results. |
| packages/chat-runtime/src/stream/subscriptions.ts | Adds replay/bootstrap subscriptions, delta filtering, and scheduled coalescing; no independently publishable defect was established. |
| packages/chat-runtime/src/db/queries.ts | Adds bound SQLite queries for journal and projection persistence, but exposes no cleanup primitive for partially created sessions. |
| packages/chat-runtime/src/index.ts | Composes database, journal, live sessions, subscriptions, commands, and disposal into the public runtime. |
Sequence Diagram
sequenceDiagram
participant Client
participant Commands
participant Journal
participant Registry
participant Harness
participant Hub
Client->>Commands: createSession
Commands->>Journal: open and persist session
Commands->>Registry: create live session
Registry->>Harness: start adapter
Client->>Commands: prompt
Commands->>Registry: deliver or queue prompt
Harness-->>Registry: async events
Registry->>Journal: append durable event
Registry->>Hub: publish envelope
Hub-->>Client: replay/live envelopes
Prompt To Fix All With AI
### Issue 1
packages/chat-runtime/src/commands/commands.ts:94-101
**Failed creation persists ghost sessions**
When the harness is unregistered or its factory/start method throws, `journal.open` has already persisted the session and no rollback removes it, causing an unusable `starting` session to remain visible through `listSessions` and `getSession`; retries can persist additional orphans.
### Issue 2
packages/chat-runtime/src/sessions/liveSession.ts:75
**Adapter pump failures go unhandled**
When the harness event iterator or event handling throws, `run` rejects without an attached handler or failure-state transition, causing an unhandled rejection while the registered session retains stale running or queued state; disposal later rethrows the same rejected pump.
---
For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.Reviews (1): Last reviewed commit: "feat(chat-runtime): live layer — session..." | Re-trigger Greptile
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
Comment on lines
+94
to
+101
| options.live.create({ | ||
| sessionId, | ||
| workspaceId: parsed.workspaceId, | ||
| harness: parsed.harness, | ||
| cwd: parsed.cwd, | ||
| modeId: parsed.modeId, | ||
| modelId: parsed.modelId, | ||
| }); |
Contributor
There was a problem hiding this comment.
When the harness is unregistered or its factory/start method throws, journal.open has already persisted the session and no rollback removes it, causing an unusable starting session to remain visible through listSessions and getSession; retries can persist additional orphans.
Prompt To Fix With AI
This is a comment left during a code review.
Path: packages/chat-runtime/src/commands/commands.ts
Line: 94-101
Comment:
**Failed creation persists ghost sessions**
When the harness is unregistered or its factory/start method throws, `journal.open` has already persisted the session and no rollback removes it, causing an unusable `starting` session to remain visible through `listSessions` and `getSession`; retries can persist additional orphans.
---
For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.|
|
||
| start(startOptions: HarnessStartOptions): void { | ||
| this.emitSession({ status: "starting" }); | ||
| this.pump = this.run(this.options.adapter.start(startOptions)); |
Contributor
There was a problem hiding this comment.
When the harness event iterator or event handling throws, run rejects without an attached handler or failure-state transition, causing an unhandled rejection while the registered session retains stale running or queued state; disposal later rethrows the same rejected pump.
Prompt To Fix With AI
This is a comment left during a code review.
Path: packages/chat-runtime/src/sessions/liveSession.ts
Line: 75
Comment:
**Adapter pump failures go unhandled**
When the harness event iterator or event handling throws, `run` rejects without an attached handler or failure-state transition, causing an unhandled rejection while the registered session retains stale running or queued state; disposal later rethrows the same rejected pump.
---
For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.
Contributor
🧹 Preview Cleanup CompleteThe following preview resources have been cleaned up:
Thank you for your contribution! 🎉 |
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
19 issues found across 33 files
Prompt for AI agents (unresolved issues)
Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.
<file name="packages/chat-runtime/src/journal/replay.ts">
<violation number="1" location="packages/chat-runtime/src/journal/replay.ts:54">
P1: Reconnects can silently accept a partial transcript when an interior journal row is missing: `readSince` only reports `journal_missing` when the entire epoch is empty, then returns sequences such as 1 and 3 as a successful replay. Validate that the returned rows are contiguous through the fetched head, or return `journal_missing` before exposing the replay.</violation>
<violation number="2" location="packages/chat-runtime/src/journal/replay.ts:68">
P2: History pagination accepts a current-epoch `before` cursor beyond the journal head and returns the newest page instead of `invalid_cursor`, leaving a corrupted client state looking valid. Compare `before.seq` with `latestSeq` and return the reset result before querying the page.</violation>
</file>
<file name="plans/chat-ship-plan.md">
<violation number="1" location="plans/chat-ship-plan.md:78">
P3: The amended §2 header says `packages/chat-runtime` already landed as its own standalone package, but line 129 still describes the extraction as a future `git mv` of the host-service subtree — a stale, mutually contradicting statement now that the package ships directly. Update line 129 to reflect that the runtime is a standalone package that does not need the `git mv` step.</violation>
<violation number="2" location="plans/chat-ship-plan.md:195">
P3: The build-notes count understates what actually shipped: this PR's `packages/chat-runtime` contains 50 `test()` cases across its test files (and the PR description itself says "50 tests"), not the "29 tests" this note records. Since this note is meant to be the authoritative record for the M5 ship, fix the count so it matches the merged package — otherwise anyone reading the plan for the CDP pass will be working from a wrong baseline.</violation>
</file>
<file name="packages/chat-runtime/src/commands/commands.ts">
<violation number="1" location="packages/chat-runtime/src/commands/commands.ts:89">
P2: A failed session start leaves a persisted session behind because `journal.open` runs before `live.create`; for example, an unknown harness throws after the row is inserted, so callers see a non-running ghost session and retries create more rows. Validating/reserving the harness before opening the journal or rolling back the row when startup fails would keep creation atomic.</violation>
<violation number="2" location="packages/chat-runtime/src/commands/commands.ts:94">
P1: A harness stream failure can become an unhandled rejection after `createSession` has returned, leaving the session stuck in its last state and potentially terminating the host process. The live-session startup path should observe the pump rejection and publish a failure state (and make disposal consume the failure) instead of letting it escape unhandled.</violation>
<violation number="3" location="packages/chat-runtime/src/commands/commands.ts:117">
P2: The dedupe cache is a single shared instance keyed only by the raw `commandId` string, but it is used by every mutation command (`createSession`, `prompt`, `cancelTurn`, `respondToApproval`, `setMode`). That means the same `commandId` issued for a different command kind (or for the same kind on a different session) hits the same cache entry and replays a result of the wrong shape — e.g. `prompt` could return a cached `{ sessionId, epoch }` from an earlier `createSession`, or `createSession` could silently replay a stale session while never re-creating the live session. With random UUIDs collisions are rare, but commandId is a free-form string, so a client using stable/deterministic ids makes this a real correctness bug. Consider scoping the dedupe key by command kind (plus sessionId where relevant), e.g. `createSession:${commandId}` / `prompt:${sessionId}:${commandId}`, so different command types never share a cache slot.</violation>
</file>
<file name="plans/chat-harness-adapters.md">
<violation number="1" location="plans/chat-harness-adapters.md:43">
P3: The §2 directory tree above this amendment still labels the journal as host.db, which now directly contradicts the new '(<dataDir>/chat.db — never host.db)' directive — the amendment remaps paths but not this DB-name comment. Update the tree's '(host.db)' annotation to '(chat.db)' so the superseded placement can't be misread as authoritative.</violation>
</file>
<file name="packages/chat-runtime/src/commands/commands.test.ts">
<violation number="1" location="packages/chat-runtime/src/commands/commands.test.ts:216">
P3: The second page read's failure is silently swallowed: `if (!older.ok) return;` at line 216 means that if `getItems` returns `ok: false` (e.g. a reset/session-not-found), the test returns early and still passes, so the backwards-pagination assertion at line 217 never runs and a pagination regression goes undetected. Unlike the previous page, `older.ok` was never asserted true first. Assert it (e.g. `expect(older.ok).toBe(true); if (!older.ok) return;`) so the test fails rather than silently succeeding.</violation>
</file>
<file name="packages/chat-runtime/src/sessions/registry.ts">
<violation number="1" location="packages/chat-runtime/src/sessions/registry.ts:47">
P1: Creating another live session with an existing `sessionId` silently abandons the first adapter, so its pump can continue emitting while commands target the replacement. Reject duplicate IDs or dispose the existing session before replacing it.</violation>
<violation number="2" location="packages/chat-runtime/src/sessions/registry.ts:48">
P1: A synchronous startup failure leaves this session registered as running even though its pump was never started, and leaks any partially initialized adapter. Startup should remove the entry and dispose the session on failure before rethrowing.</violation>
<violation number="3" location="packages/chat-runtime/src/sessions/registry.ts:77">
P1: If one adapter fails during shutdown, teardown stops after `disposeAll()` and the registry has already discarded the sessions, leaving the database/subscriptions open and cleanup non-retryable. Shutdown should settle all disposal attempts and ensure outer runtime cleanup runs even when one adapter fails.</violation>
</file>
<file name="packages/chat-runtime/src/journal/replay.test.ts">
<violation number="1" location="packages/chat-runtime/src/journal/replay.test.ts:130">
P2: The `older`/`oldest` readPage calls in the paging test have a bare `if (!ok) return;` with no preceding `expect(...ok).toBe(true)` (unlike the `newest` call). A regression that makes a follow-up page return ok:false would silently pass the test instead of failing, hiding the paging bug. Add `expect(older.ok).toBe(true);` / `expect(oldest.ok).toBe(true);` before each guard, or assert on the failure shape.</violation>
</file>
<file name="packages/chat-runtime/src/harness/fake/fakeHarness.ts">
<violation number="1" location="packages/chat-runtime/src/harness/fake/fakeHarness.ts:92">
P3: Concurrent consumers of the returned async iterator can leave an earlier `next()` promise pending forever because the queue has only one waiting slot. Pending resolvers should be queued in FIFO order, or concurrent pulls should be explicitly serialized or rejected.</violation>
<violation number="2" location="packages/chat-runtime/src/harness/fake/fakeHarness.ts:199">
P2: Canceling a delayed turn does not unblock the next FIFO prompt because this sleep cannot be interrupted and later runs are serialized behind it. A cancellable delay should release the canceled run immediately, with disposal using the same cancellation path.</violation>
</file>
<file name="packages/chat-runtime/src/journal/epoch.ts">
<violation number="1" location="packages/chat-runtime/src/journal/epoch.ts:29">
P2: For a session whose journal has zero rows, `open()`/`openEpoch()` always mints a fresh epoch, because the `latestSeq() === 0` branch can't distinguish a freshly-created-but-never-appended session from one whose journal was actually lost. As a result, reopening a just-opened session (process restart, reconnect, or a second registry) mutates its epoch and bumps `updated_at`, silently invalidating any cursor previously handed out and triggering spurious `epoch_changed` resets even though no data was lost. The keeps-epoch path is only covered by tests after events exist. Consider minting a new epoch only when there is evidence the journal previously held rows (or tracking whether the row was inserted in this very call), rather than whenever the epoch is empty.</violation>
<violation number="2" location="packages/chat-runtime/src/journal/epoch.ts:48">
P2: When the journal is lost, the recovery path mints a new epoch and only calls `setEpoch`, leaving the session's projection row (status / title / queued_count) holding pre-loss values that no longer match the empty journal. Since `ChatSessionStore.get()/list()` read that projection directly, a session recovered from a lost journal briefly reports stale queued_count/status/title (e.g. queued messages that no longer exist) until the next append rewrites the projection. Consider resetting the projection columns (status/title/queuedCount) as part of the epoch-mint recovery so the DB projection is consistent with the rebuilt, empty journal.</violation>
</file>
<file name="packages/chat-runtime/src/journal/replayParity.test.ts">
<violation number="1" location="packages/chat-runtime/src/journal/replayParity.test.ts:135">
P3: The hard-coded drain counts (1/6/5) are brittle: the parity assertions are internally self-consistent (reduceMany(live) and reduceMany(replay) both derive from whatever was appended, and cursor.seq is pinned to live.length), so if the script or harness changes, the test can silently consume a different number of events and still pass without exercising the intended flow. Suggest asserting the expected durable-event count independently (e.g., expect(live).toHaveLength(10)) and/or draining until result.done with a timeout so under-supplied scripts fail fast rather than hang.</violation>
</file>
<file name="packages/chat-runtime/src/stream/subscriptions.ts">
<violation number="1" location="packages/chat-runtime/src/stream/subscriptions.ts:142">
P3: The per-session `coalescers` map is never pruned when the last subscriber for a session unsubscribes (the `unsubscribe` handler removes the subscriber set but leaves the coalescer), and `publish` eagerly creates a coalescer for any session that emits a delta even when it has zero subscribers. In a long-lived runtime that has served many streaming sessions, this retains an entry (and, while deltas keep arriving, a rescheduling 33ms timer that flushes to no one) per session until `dispose()`, which is an unbounded lifecycle/memory growth tied only to runtime teardown. Consider removing the coalescer from `coalescers` when its subscriber set empties (e.g., in the `unsubscribe` path) so per-session state is reclaimed with the last client.</violation>
</file>
Architecture diagram
sequenceDiagram
participant Host as Host Service
participant Runtime as ChatRuntime
participant Journal as ChatJournal
participant Queries as ChatQueries
participant DB as chat.db (SQLite)
participant LiveReg as LiveSessionRegistry
participant LiveSess as LiveSession
participant Harness as HarnessAdapter (FakeHarness)
participant SubHub as SubscriptionHub
participant Sink as Subscriber Sink
participant Cmd as CommandDedupe
Note over Host,Sink: Session Creation Flow
Host->>Runtime: createChatRuntime({ dataDir, harnesses })
Runtime->>DB: openChatDb() — bootstrap chat.db (WAL)
Runtime-->>Host: runtime object with journal, commands, subscriptions
Host->>Cmd: commands.createSession({ commandId, workspaceId, harness, cwd })
Cmd->>Cmd: dedupe.run(commandId) — LRU cache
alt Duplicate commandId
Cmd-->>Host: cached result (no journal/session created)
else First commandId
Cmd->>Journal: open({ sessionId, workspaceId, harness })
Journal->>Queries: session(sessionId) — check existing row
Queries->>DB: SELECT from chat_sessions_local
DB-->>Queries: null (new session)
Journal->>Queries: insertSession(...) — mint epoch
Queries->>DB: INSERT into chat_sessions_local
DB-->>Queries: ok
Journal->>Journal: cache session (epoch, lastSeq=0)
Journal-->>Cmd: { sessionId, epoch }
Cmd->>LiveReg: create({ sessionId, harness, cwd })
LiveReg->>LiveSess: new LiveSession(journal, publish)
LiveSess->>Harness: start({ cwd })
Harness-->>LiveSess: AsyncIterable<AdapterEvent>
LiveSess->>LiveSess: emitSession({ status: "starting" })
LiveSess->>Journal: appendEnvelope(session, session event)
Journal->>Queries: transaction(appendJournalRow + setProjection)
Queries->>DB: INSERT chat_journal + UPDATE chat_sessions_local
DB-->>Queries: ok
Journal-->>LiveSess: cursor
LiveSess->>SubHub: publish(envelope)
SubHub->>Sink: send(envelope) — to all subscribers
Cmd-->>Host: { sessionId, epoch }
end
Note over Host,Sink: Prompt Flow
Host->>Cmd: commands.prompt({ commandId, sessionId, clientId, content })
Cmd->>Cmd: dedupe.run(commandId)
alt Duplicate commandId
Cmd-->>Host: cached result
else First
Cmd->>LiveReg: require(sessionId)
LiveReg-->>Cmd: LiveSession
Cmd->>LiveSess: prompt(content, clientId)
LiveSess->>LiveSess: mint user_message id
alt Session is busy (turn running or queue pending)
LiveSess->>Journal: appendEnvelope(item with queued: true)
Journal->>Queries: transaction(append + update queuedCount)
Queries->>DB: INSERT + UPDATE
LiveSess->>LiveSess: push to FIFO queue
LiveSess-->>Cmd: { itemId, queued: true }
else Session idle
LiveSess->>Journal: appendEnvelope(item without queued flag)
Journal->>Queries: transaction(append + update)
LiveSess->>Harness: prompt(content)
LiveSess-->>Cmd: { itemId, queued: false }
end
Journal->>SubHub: publish(envelope)
SubHub->>SubHub: coalesce deltas (33ms window)
SubHub->>Sink: send(envelope) — durable events flush deltas first
Cmd-->>Host: result
end
Note over Host,Sink: Adapter Event Loop
loop Per adapter event
Harness-->>LiveSess: AdapterEvent
alt event is "item"
LiveSess->>Journal: appendEnvelope({ type:"item", item, turnId })
Journal->>Queries: transaction(append + project)
Journal->>SubHub: publish(envelope)
else event is "turn"
LiveSess->>LiveSess: update currentTurn
LiveSess->>Journal: appendEnvelope({ type:"turn", turn })
Journal->>SubHub: publish(envelope)
alt turn status is "running"
LiveSess->>LiveSess: attributeAwaitingPrompt(turnId) — re-append user_message with turnId
LiveSess->>Journal: appendEnvelope(item without queued)
Journal->>SubHub: publish(envelope)
else turn is completed/interrupted
LiveSess->>LiveSess: deliverNextQueued() — dequeue and push to adapter
end
else event is "session"
LiveSess->>LiveSess: merge session state (auto-set to "running" if queue pending)
LiveSess->>Journal: appendEnvelope({ type:"session", session })
Journal->>SubHub: publish(envelope)
else event is "delta"
LiveSess->>SubHub: publish(delta envelope) — non-durable, goes to coalescer
SubHub->>SubHub: createCoalescer(sessionId, 33ms schedule)
Note over SubHub: Deltas merge by type+itemId
end
end
Note over Host,Sink: Subscription (Replay-Then-Live)
Host->>SubHub: subscribe(sessionId, { since, deltas })
SubHub->>SubHub: flushDeltas(sessionId)
alt since cursor provided
SubHub->>Queries: readSince(sessionId, cursor)
Queries->>DB: SELECT from chat_journal WHERE epoch=? AND seq>?
DB-->>Queries: rows
alt epoch_changed or session_not_found
Queries-->>SubHub: { ok:false, reset:"epoch_changed" }
SubHub->>Sink: send(reset envelope) — reset frame, subscription stays
else ok
Queries-->>SubHub: { ok:true, envelopes }
SubHub->>Sink: send(envelopes) — replay spine
end
else no cursor (new subscriber)
SubHub->>Queries: readPage(sessionId, { limit:200 })
Queries->>DB: SELECT newest events
DB-->>Queries: rows
SubHub->>Sink: send(envelopes) — bootstrap spine
end
SubHub->>SubHub: register subscriber with delta filter set
SubHub-->>Host: subscription (unsubscribe handle)
Note over Sink: Live delivery continues via publish()
Note over Host,Sink: Cancel Turn Flow
Host->>Cmd: commands.cancelTurn({ commandId, sessionId, turnId })
Cmd->>Cmd: dedupe.run(commandId)
Cmd->>LiveReg: require(sessionId)
LiveReg-->>Cmd: LiveSession
Cmd->>LiveSess: cancelTurn(turnId)
LiveSess->>Harness: cancelTurn()
alt Turn is running with pending approval
Harness->>Harness: stale all pending approvals
Harness->>Harness: cancel all running tool calls
Harness-->>LiveSess: item events (stale, canceled) + turn event (interrupted)
LiveSess->>Journal: appendEnvelope(canceled items + interrupted turn)
Journal->>SubHub: publish(envelopes)
else Turn already completed
Harness-->>LiveSess: no-op (currentTurn already done)
end
Note over Host,Sink: Error / Reset Paths
alt Unknown harness
LiveReg->>LiveReg: factory not in HarnessRegistry
LiveReg-->>Cmd: throw "unknown harness nope"
end
alt Session not found in registry
Cmd->>LiveReg: require("missing")
LiveReg-->>Cmd: throw "chat session missing is not running"
end
alt Schema validation failure
Cmd->>Cmd: zod parse(input) — promptInputSchema, etc.
alt Invalid
Cmd-->>Host: throw ZodError (caught by host)
end
end
alt Journal unreadable (journal_missing after open)
Journal->>Queries: readSince returns { ok:false, reset:"journal_missing" }
Journal->>Journal: cacheFor() throws — session fails to open
end
alt Primary key conflict on append (transaction rollback)
Journal->>Queries: transaction(appendJournalRow + setProjection)
Queries->>DB: INSERT ... seq already exists
DB-->>Queries: SQLITE_CONSTRAINT error
Queries->>Queries: transaction rollback — no changes committed
Journal-->>Host: throw (projection unchanged, no orphaned journal row)
end
Reply with feedback, questions, or to request a fix.
Re-trigger cubic
| } | ||
| if (cursor.seq > last) return { ok: false, reset: "invalid_cursor" }; | ||
|
|
||
| const rows = queries.rowsSince(sessionId, session.epoch, cursor.seq); |
Contributor
There was a problem hiding this comment.
P1: Reconnects can silently accept a partial transcript when an interior journal row is missing: readSince only reports journal_missing when the entire epoch is empty, then returns sequences such as 1 and 3 as a successful replay. Validate that the returned rows are contiguous through the fetched head, or return journal_missing before exposing the replay.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At packages/chat-runtime/src/journal/replay.ts, line 54:
<comment>Reconnects can silently accept a partial transcript when an interior journal row is missing: `readSince` only reports `journal_missing` when the entire epoch is empty, then returns sequences such as 1 and 3 as a successful replay. Validate that the returned rows are contiguous through the fetched head, or return `journal_missing` before exposing the replay.</comment>
<file context>
@@ -0,0 +1,91 @@
+ }
+ if (cursor.seq > last) return { ok: false, reset: "invalid_cursor" };
+
+ const rows = queries.rowsSince(sessionId, session.epoch, cursor.seq);
+ return {
+ ok: true,
</file context>
| workspaceId: parsed.workspaceId, | ||
| harness: parsed.harness, | ||
| }); | ||
| options.live.create({ |
Contributor
There was a problem hiding this comment.
P1: A harness stream failure can become an unhandled rejection after createSession has returned, leaving the session stuck in its last state and potentially terminating the host process. The live-session startup path should observe the pump rejection and publish a failure state (and make disposal consume the failure) instead of letting it escape unhandled.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At packages/chat-runtime/src/commands/commands.ts, line 94:
<comment>A harness stream failure can become an unhandled rejection after `createSession` has returned, leaving the session stuck in its last state and potentially terminating the host process. The live-session startup path should observe the pump rejection and publish a failure state (and make disposal consume the failure) instead of letting it escape unhandled.</comment>
<file context>
@@ -0,0 +1,163 @@
+ workspaceId: parsed.workspaceId,
+ harness: parsed.harness,
+ });
+ options.live.create({
+ sessionId,
+ workspaceId: parsed.workspaceId,
</file context>
| now: this.options.now, | ||
| }); | ||
| this.live.set(options.sessionId, session); | ||
| session.start({ |
Contributor
There was a problem hiding this comment.
P1: A synchronous startup failure leaves this session registered as running even though its pump was never started, and leaks any partially initialized adapter. Startup should remove the entry and dispose the session on failure before rethrowing.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At packages/chat-runtime/src/sessions/registry.ts, line 48:
<comment>A synchronous startup failure leaves this session registered as running even though its pump was never started, and leaks any partially initialized adapter. Startup should remove the entry and dispose the session on failure before rethrowing.</comment>
<file context>
@@ -0,0 +1,79 @@
+ now: this.options.now,
+ });
+ this.live.set(options.sessionId, session);
+ session.start({
+ cwd: options.cwd,
+ modeId: options.modeId,
</file context>
| Lint guards (CI): `protocol/` imports zod only; `core/` imports `protocol/` + stdlib only; `react/` is the only entry importing react; nothing imports react-dom/react-native/node builtins. | ||
|
|
||
| ## 2. Workstream B — host runtime (`packages/host-service/src/runtime/chat/`) | ||
| ## 2. Workstream B — host runtime (**AMENDED 2026-08-04:** now standalone `packages/chat-runtime`, owning its own `chat.db` — never host.db; exports `createChatRuntime({ dataDir })` plus, later, a mountable tRPC router + WS handler that host-service mounts in a few lines. Paths below map from `host-service/src/runtime/chat/` → `chat-runtime/src/`; the db/ section's host.db tables become CREATE-IF-NOT-EXISTS bootstrap in the package's own file.) |
Contributor
There was a problem hiding this comment.
P3: The amended §2 header says packages/chat-runtime already landed as its own standalone package, but line 129 still describes the extraction as a future git mv of the host-service subtree — a stale, mutually contradicting statement now that the package ships directly. Update line 129 to reflect that the runtime is a standalone package that does not need the git mv step.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At plans/chat-ship-plan.md, line 78:
<comment>The amended §2 header says `packages/chat-runtime` already landed as its own standalone package, but line 129 still describes the extraction as a future `git mv` of the host-service subtree — a stale, mutually contradicting statement now that the package ships directly. Update line 129 to reflect that the runtime is a standalone package that does not need the `git mv` step.</comment>
<file context>
@@ -75,7 +75,7 @@ type SessionClient = {
Lint guards (CI): `protocol/` imports zod only; `core/` imports `protocol/` + stdlib only; `react/` is the only entry importing react; nothing imports react-dom/react-native/node builtins.
-## 2. Workstream B — host runtime (`packages/host-service/src/runtime/chat/`)
+## 2. Workstream B — host runtime (**AMENDED 2026-08-04:** now standalone `packages/chat-runtime`, owning its own `chat.db` — never host.db; exports `createChatRuntime({ dataDir })` plus, later, a mountable tRPC router + WS handler that host-service mounts in a few lines. Paths below map from `host-service/src/runtime/chat/` → `chat-runtime/src/`; the db/ section's host.db tables become CREATE-IF-NOT-EXISTS bootstrap in the package's own file.)
</file context>
</details>
| if (buffered) return Promise.resolve({ value: buffered, done: false }); | ||
| if (this.closed) return Promise.resolve({ value: undefined, done: true }); | ||
| return new Promise((resolve) => { | ||
| this.waiting = resolve; |
Contributor
There was a problem hiding this comment.
P3: Concurrent consumers of the returned async iterator can leave an earlier next() promise pending forever because the queue has only one waiting slot. Pending resolvers should be queued in FIFO order, or concurrent pulls should be explicitly serialized or rejected.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At packages/chat-runtime/src/harness/fake/fakeHarness.ts, line 92:
<comment>Concurrent consumers of the returned async iterator can leave an earlier `next()` promise pending forever because the queue has only one waiting slot. Pending resolvers should be queued in FIFO order, or concurrent pulls should be explicitly serialized or rejected.</comment>
<file context>
@@ -0,0 +1,233 @@
+ if (buffered) return Promise.resolve({ value: buffered, done: false });
+ if (this.closed) return Promise.resolve({ value: undefined, done: true });
+ return new Promise((resolve) => {
+ this.waiting = resolve;
+ });
+ }
</file context>
| const iterator = harness.start({ cwd: "/tmp" })[Symbol.asyncIterator](); | ||
| await drain(iterator, 1, handle); | ||
| harness.prompt([{ type: "text", text: "run ls" }]); | ||
| await drain(iterator, 6, handle); |
Contributor
There was a problem hiding this comment.
P3: The hard-coded drain counts (1/6/5) are brittle: the parity assertions are internally self-consistent (reduceMany(live) and reduceMany(replay) both derive from whatever was appended, and cursor.seq is pinned to live.length), so if the script or harness changes, the test can silently consume a different number of events and still pass without exercising the intended flow. Suggest asserting the expected durable-event count independently (e.g., expect(live).toHaveLength(10)) and/or draining until result.done with a timeout so under-supplied scripts fail fast rather than hang.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At packages/chat-runtime/src/journal/replayParity.test.ts, line 135:
<comment>The hard-coded drain counts (1/6/5) are brittle: the parity assertions are internally self-consistent (reduceMany(live) and reduceMany(replay) both derive from whatever was appended, and cursor.seq is pinned to live.length), so if the script or harness changes, the test can silently consume a different number of events and still pass without exercising the intended flow. Suggest asserting the expected durable-event count independently (e.g., expect(live).toHaveLength(10)) and/or draining until result.done with a timeout so under-supplied scripts fail fast rather than hang.</comment>
<file context>
@@ -0,0 +1,166 @@
+ const iterator = harness.start({ cwd: "/tmp" })[Symbol.asyncIterator]();
+ await drain(iterator, 1, handle);
+ harness.prompt([{ type: "text", text: "run ls" }]);
+ await drain(iterator, 6, handle);
+ harness.respondToApproval("ap1", { type: "accept" });
+ await drain(iterator, 5, handle);
</file context>
| this.subscribers.clear(); | ||
| } | ||
|
|
||
| private coalescerFor(sessionId: string): Coalescer<DeltaEnvelope> { |
Contributor
There was a problem hiding this comment.
P3: The per-session coalescers map is never pruned when the last subscriber for a session unsubscribes (the unsubscribe handler removes the subscriber set but leaves the coalescer), and publish eagerly creates a coalescer for any session that emits a delta even when it has zero subscribers. In a long-lived runtime that has served many streaming sessions, this retains an entry (and, while deltas keep arriving, a rescheduling 33ms timer that flushes to no one) per session until dispose(), which is an unbounded lifecycle/memory growth tied only to runtime teardown. Consider removing the coalescer from coalescers when its subscriber set empties (e.g., in the unsubscribe path) so per-session state is reclaimed with the last client.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At packages/chat-runtime/src/stream/subscriptions.ts, line 142:
<comment>The per-session `coalescers` map is never pruned when the last subscriber for a session unsubscribes (the `unsubscribe` handler removes the subscriber set but leaves the coalescer), and `publish` eagerly creates a coalescer for any session that emits a delta even when it has zero subscribers. In a long-lived runtime that has served many streaming sessions, this retains an entry (and, while deltas keep arriving, a rescheduling 33ms timer that flushes to no one) per session until `dispose()`, which is an unbounded lifecycle/memory growth tied only to runtime teardown. Consider removing the coalescer from `coalescers` when its subscriber set empties (e.g., in the `unsubscribe` path) so per-session state is reclaimed with the last client.</comment>
<file context>
@@ -0,0 +1,167 @@
+ this.subscribers.clear();
+ }
+
+ private coalescerFor(sessionId: string): Coalescer<DeltaEnvelope> {
+ const existing = this.coalescers.get(sessionId);
+ if (existing) return existing;
</file context>
…y, fake harness New @superset/chat-runtime package owning its own chat.db (WAL, idempotent bootstrap, no host.db contact): append-only journal with same-transaction projection, epoch minting on journal creation/loss only, readSince/readPage returning reset signals on epoch mismatch, session read model, HarnessAdapter interface, and a scripted fake harness. 29 tests including replay===live parity through the shared reducer. Injected SqliteDatabase seam (better-sqlite3 in prod, bun:sqlite in tests — Electron-ABI build can't load under bun). Plan docs amended: runtime extracted from host-service into this package; host passes resolved facts (workspaceId + cwd), never workspace concepts. Claude-Session: https://claude.ai/code/session_01XRFCEPsfmJHu2SenDKYJ2h
LiveSession pumps adapter events through the journal with FIFO prompt queueing and runtime-minted user_message attribution; SubscriptionHub does replay-then-live with per-subscriber delta filtering, 33ms coalescing flushed before every durable event, and reset frames that keep the subscriber attached; commands parse protocol schemas with result-replaying commandId dedupe. 50 tests including whole-envelope reconnect equality (live===replay by construction via appendEnvelope). Claude-Session: https://claude.ai/code/session_01XRFCEPsfmJHu2SenDKYJ2h
saddlepaddle
force-pushed
the
chat-runtime-live
branch
from
August 5, 2026 16:54
76f67b6 to
5eb6c1b
Compare
![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: 9
🧹 Nitpick comments (5)
packages/chat-runtime/src/harness/fake/fakeHarness.ts (1)
87-94: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueGuard the single waiter slot in
next().
next()stores one resolver inthis.waiting. If two calls tonext()happen before an event arrives, the second call overwrites the first resolver. The first promise then never settles. Current consumers read serially, so this is latent only. A guard makes the misuse explicit.♻️ Proposed guard
next(): Promise<IteratorResult<AdapterEvent>> { const buffered = this.buffered.shift(); if (buffered) return Promise.resolve({ value: buffered, done: false }); if (this.closed) return Promise.resolve({ value: undefined, done: true }); + if (this.waiting) { + return Promise.reject(new Error("EventQueue supports a single reader")); + } return new Promise((resolve) => { this.waiting = resolve; }); }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/chat-runtime/src/harness/fake/fakeHarness.ts` around lines 87 - 94, Update next() to detect an existing this.waiting resolver before assigning a new one, and explicitly reject or otherwise surface the concurrent next() misuse instead of overwriting the prior waiter. Preserve the current buffered-event and closed-stream behavior.packages/chat-runtime/src/journal/journal.ts (1)
100-162: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winExtract the projection fold so replay and append cannot diverge.
projectionForand the replay loop incacheForimplement the same fold twice. Both handlesessionevents,user_messagequeue add/remove, and title fallback. If one changes, the cached projection and the rebuilt projection diverge silently, andqueued_countinchat_sessions_localno longer matches the journal.Extract a single reducer and call it from both paths.
♻️ Proposed shape
function applyEvent(state: NextProjection, event: DurableEvent): NextProjection { if (event.type === "session") { return { status: event.session.status, title: event.session.title ?? state.title, queuedItemIds: state.queuedItemIds, }; } if (event.type === "item" && event.item.kind === "user_message") { const queuedItemIds = new Set(state.queuedItemIds); if ("queued" in event.item && event.item.queued === true) { queuedItemIds.add(event.item.id); } else { queuedItemIds.delete(event.item.id); } return { ...state, queuedItemIds }; } return state; }Then
projectionForcallsapplyEventonce, andcacheForfoldsreplay.envelopesthrough it.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/chat-runtime/src/journal/journal.ts` around lines 100 - 162, Extract the shared event-folding logic from projectionFor and the replay loop in cacheFor into a single applyEvent reducer operating on NextProjection and DurableEvent. Update projectionFor to call applyEvent with the current cache projection, and update cacheFor to fold every replay envelope through the same reducer while preserving lastSeq tracking and the existing initial session state.packages/chat-runtime/src/journal/replayParity.test.ts (1)
94-104: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAssert the drained event count so the parity check cannot pass vacuously.
drainreturns early when the iterator reportsdone. The parity assertions at lines 147-152 comparereplayagainstlive, and both derive from whatever actually drained. If the counts 1, 6, and 5 stop matchingSCRIPT— for example after a script event is added — the test drains fewer events and still passes with reduced coverage.The counts also sum to 12 while
SCRIPTcontains 11 events, which makes the intent hard to verify.Assert the expected number of durable envelopes explicitly.
💚 Proposed fix
await harness.dispose(); + // 11 scripted events minus the 2 non-durable deltas. + expect(live).toHaveLength(9); + const replay = readSince(runtime.queries, SESSION, {Deriving the count from
SCRIPTinstead of a literal keeps it correct as the script grows.Also applies to: 133-137
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/chat-runtime/src/journal/replayParity.test.ts` around lines 94 - 104, Update the drain calls and the drain helper in replayParity.test.ts to derive the expected event count from SCRIPT rather than hard-coded counts, and assert that exactly that many durable envelopes are consumed. Ensure early iterator completion fails the test instead of returning silently, while preserving the existing replay/live parity assertions.packages/chat-runtime/src/journal/replay.test.ts (1)
126-140: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAssert
okbefore the narrowing return.Lines 130 and 138 use
if (!older.ok) return;andif (!oldest.ok) return;without a preceding assertion. IfreadPageregresses and returns a reset, the test returns early and still passes. The first read at Line 121 does assertnewest.ok. Apply the same pattern to the two later reads.💚 Proposed fix
const older = readPage(runtime.queries, SESSION, { before: newest.nextBefore ?? undefined, limit: 2, }); + expect(older.ok).toBe(true); if (!older.ok) return; expect(itemIds(older.envelopes)).toEqual(["item-2", "item-3"]); expect(older.nextBefore).toEqual({ epoch: opened.epoch, seq: 2 }); const oldest = readPage(runtime.queries, SESSION, { before: older.nextBefore ?? undefined, limit: 2, }); + expect(oldest.ok).toBe(true); if (!oldest.ok) return;🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/chat-runtime/src/journal/replay.test.ts` around lines 126 - 140, Replace the early returns for the `older` and `oldest` results in the pagination test with explicit assertions that each result is `ok`, then narrow the type for the existing envelope and cursor expectations. Keep the established `newest.ok` assertion pattern and preserve all current expectations.packages/chat-runtime/src/journal/journal.test.ts (1)
16-18: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winUndisposed test runtimes across the journal test files. Each
beforeEachcreates a runtime withcreateTestRuntime()and never releases its database handle.replayParity.test.ts(Line 164) callsawait runtime.dispose(), which shows the runtime owns a resource that needs release. The shared root cause is one missingafterEachteardown per describe block.
packages/chat-runtime/src/journal/journal.test.ts#L16-L18: add anafterEachthat awaitsruntime.dispose()in theChatJournaldescribe block.packages/chat-runtime/src/journal/replay.test.ts#L36-L38: add the sameafterEachin thereadSincedescribe block.packages/chat-runtime/src/journal/replay.test.ts#L112-L114: add the sameafterEachin thereadPagedescribe block.Import
afterEachfrombun:testin both files.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/chat-runtime/src/journal/journal.test.ts` around lines 16 - 18, Add afterEach teardown for the runtimes created in the ChatJournal describe block in packages/chat-runtime/src/journal/journal.test.ts#L16-L18 and the readSince and readPage describe blocks in packages/chat-runtime/src/journal/replay.test.ts#L36-L38 and `#L112-L114`, awaiting runtime.dispose() after each test; import afterEach from bun:test in both files.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@packages/chat-runtime/src/commands/commands.ts`:
- Around line 85-103: Update createSession in the command implementation to
validate the harness through options.live before calling options.journal.open,
and wrap live-session startup after opening so any failure rolls back the
journal session before rethrowing. Preserve the returned sessionId and epoch
behavior on success, and add a regression test asserting an unknown harness
leaves listSessions() empty.
In `@packages/chat-runtime/src/db/db.ts`:
- Around line 32-36: Update openChatDb so PRAGMA and BOOTSTRAP_STATEMENTS
initialization runs in error handling that closes db before rethrowing any
failure. Preserve successful initialization and return behavior, while ensuring
every bootstrap error releases the database handle first.
In `@packages/chat-runtime/src/index.ts`:
- Around line 136-140: Update the adapter’s dispose function to wrap
live.disposeAll(), subscriptions.dispose(), and db.close() in nested finally
blocks so subscription and database cleanup always execute, while preserving and
propagating any error from live.disposeAll().
In `@packages/chat-runtime/src/journal/replay.ts`:
- Around line 20-59: Update readSince to guard the rows.map parsing path,
including JSON.parse and envelopeSchema.parse failures from parseJournalRow, and
convert any unreadable journal row into an unsuccessful ReplayResult with the
appropriate journal-missing reset reason. Preserve existing cursor and session
validation behavior, and keep parseJournalRow’s parsing responsibilities
unchanged.
In `@packages/chat-runtime/src/journal/sessions.test.ts`:
- Around line 9-21: Update the sessions test suite to dispose the SQLite runtime
after every test: add an afterEach cleanup that calls runtime.dispose(), while
preserving the existing beforeEach setup and journal session initialization.
In `@packages/chat-runtime/src/sessions/liveSession.ts`:
- Around line 99-102: Update LiveSession.cancelTurn to forward cancellation only
when the requested turn matches an existing, currently running turn. Track or
validate the turn’s active state, reject cancellation when currentTurn is absent
or completed/interrupted, and preserve the existing mismatch guard; ensure
currentTurn is cleared or marked inactive when the turn ends.
- Around line 78-97: Update the prompt delivery flow in deliver so awaitingTurn
is cleared when adapter.prompt rejects or throws. Preserve the already-appended
durable item and ensure the rejection does not leave hasPendingWork permanently
true, allowing subsequent prompts to be processed normally.
- Around line 73-76: Update start in the live session class to attach a
rejection handler to the stored pump promise, recording any stream or processing
failure as a valid SessionState-shaped failure payload while keeping the promise
handled for callers that later await dispose. Reuse the existing session
failure-recording mechanism and preserve the current pump execution flow.
In `@packages/chat-runtime/src/sessions/registry.ts`:
- Around line 33-55: Update LiveSessionRegistry.create to handle an existing
entry in this.live for options.sessionId before constructing or storing the new
LiveSession: either reject creation with an error or dispose the existing
session first, ensuring the old adapter, pump, journal activity, and
subscriptions are stopped before replacement. Preserve normal creation behavior
when no session exists.
---
Nitpick comments:
In `@packages/chat-runtime/src/harness/fake/fakeHarness.ts`:
- Around line 87-94: Update next() to detect an existing this.waiting resolver
before assigning a new one, and explicitly reject or otherwise surface the
concurrent next() misuse instead of overwriting the prior waiter. Preserve the
current buffered-event and closed-stream behavior.
In `@packages/chat-runtime/src/journal/journal.test.ts`:
- Around line 16-18: Add afterEach teardown for the runtimes created in the
ChatJournal describe block in
packages/chat-runtime/src/journal/journal.test.ts#L16-L18 and the readSince and
readPage describe blocks in
packages/chat-runtime/src/journal/replay.test.ts#L36-L38 and `#L112-L114`,
awaiting runtime.dispose() after each test; import afterEach from bun:test in
both files.
In `@packages/chat-runtime/src/journal/journal.ts`:
- Around line 100-162: Extract the shared event-folding logic from projectionFor
and the replay loop in cacheFor into a single applyEvent reducer operating on
NextProjection and DurableEvent. Update projectionFor to call applyEvent with
the current cache projection, and update cacheFor to fold every replay envelope
through the same reducer while preserving lastSeq tracking and the existing
initial session state.
In `@packages/chat-runtime/src/journal/replay.test.ts`:
- Around line 126-140: Replace the early returns for the `older` and `oldest`
results in the pagination test with explicit assertions that each result is
`ok`, then narrow the type for the existing envelope and cursor expectations.
Keep the established `newest.ok` assertion pattern and preserve all current
expectations.
In `@packages/chat-runtime/src/journal/replayParity.test.ts`:
- Around line 94-104: Update the drain calls and the drain helper in
replayParity.test.ts to derive the expected event count from SCRIPT rather than
hard-coded counts, and assert that exactly that many durable envelopes are
consumed. Ensure early iterator completion fails the test instead of returning
silently, while preserving the existing replay/live parity assertions.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 8aef12d7-7fcd-4ca2-9eed-c4365810f80a
⛔ Files ignored due to path filters (1)
bun.lockis excluded by!**/*.lock
📒 Files selected for processing (32)
packages/chat-runtime/package.jsonpackages/chat-runtime/src/commands/commandDedupe.test.tspackages/chat-runtime/src/commands/commandDedupe.tspackages/chat-runtime/src/commands/commands.test.tspackages/chat-runtime/src/commands/commands.tspackages/chat-runtime/src/db/db.test.tspackages/chat-runtime/src/db/db.tspackages/chat-runtime/src/db/queries.tspackages/chat-runtime/src/db/schema.tspackages/chat-runtime/src/db/testDb.tspackages/chat-runtime/src/fixtures.tspackages/chat-runtime/src/harness/fake/fakeHarness.test.tspackages/chat-runtime/src/harness/fake/fakeHarness.tspackages/chat-runtime/src/harness/types.tspackages/chat-runtime/src/index.tspackages/chat-runtime/src/journal/epoch.tspackages/chat-runtime/src/journal/journal.test.tspackages/chat-runtime/src/journal/journal.tspackages/chat-runtime/src/journal/replay.test.tspackages/chat-runtime/src/journal/replay.tspackages/chat-runtime/src/journal/replayParity.test.tspackages/chat-runtime/src/journal/sessions.test.tspackages/chat-runtime/src/journal/sessions.tspackages/chat-runtime/src/sessions/liveSession.test.tspackages/chat-runtime/src/sessions/liveSession.tspackages/chat-runtime/src/sessions/registry.tspackages/chat-runtime/src/stream/subscriptions.test.tspackages/chat-runtime/src/stream/subscriptions.tspackages/chat-runtime/src/testUtils.tspackages/chat-runtime/tsconfig.jsonplans/chat-harness-adapters.mdplans/chat-ship-plan.md
Comment on lines
+32
to
+36
| const db = openDatabase(join(dataDir, CHAT_DB_FILENAME)); | ||
| db.exec("PRAGMA journal_mode = WAL"); | ||
| db.exec("PRAGMA busy_timeout = 5000"); | ||
| db.exec("PRAGMA foreign_keys = ON"); | ||
| for (const statement of BOOTSTRAP_STATEMENTS) db.exec(statement); |
Contributor
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Close the database when bootstrap fails.
If a PRAGMA or bootstrap statement throws, openChatDb exits with an open database handle. A failed initialization can then retain a lock on chat.db and block a retry. Close db before rethrowing the initialization error.
Proposed fix
const db = openDatabase(join(dataDir, CHAT_DB_FILENAME));
- db.exec("PRAGMA journal_mode = WAL");
- db.exec("PRAGMA busy_timeout = 5000");
- db.exec("PRAGMA foreign_keys = ON");
- for (const statement of BOOTSTRAP_STATEMENTS) db.exec(statement);
- return db;
+ try {
+ db.exec("PRAGMA journal_mode = WAL");
+ db.exec("PRAGMA busy_timeout = 5000");
+ db.exec("PRAGMA foreign_keys = ON");
+ for (const statement of BOOTSTRAP_STATEMENTS) db.exec(statement);
+ return db;
+ } catch (error) {
+ db.close();
+ throw error;
+ }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| const db = openDatabase(join(dataDir, CHAT_DB_FILENAME)); | |
| db.exec("PRAGMA journal_mode = WAL"); | |
| db.exec("PRAGMA busy_timeout = 5000"); | |
| db.exec("PRAGMA foreign_keys = ON"); | |
| for (const statement of BOOTSTRAP_STATEMENTS) db.exec(statement); | |
| const db = openDatabase(join(dataDir, CHAT_DB_FILENAME)); | |
| try { | |
| db.exec("PRAGMA journal_mode = WAL"); | |
| db.exec("PRAGMA busy_timeout = 5000"); | |
| db.exec("PRAGMA foreign_keys = ON"); | |
| for (const statement of BOOTSTRAP_STATEMENTS) db.exec(statement); | |
| return db; | |
| } catch (error) { | |
| db.close(); | |
| throw error; | |
| } |
🧰 Tools
🪛 OpenGrep (1.26.0)
[ERROR] 36-36: Dynamic command passed to child_process.exec/execSync. Use child_process.execFile or spawn with an argument array instead.
(coderabbit.command-injection.exec-js)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@packages/chat-runtime/src/db/db.ts` around lines 32 - 36, Update openChatDb
so PRAGMA and BOOTSTRAP_STATEMENTS initialization runs in error handling that
closes db before rethrowing any failure. Preserve successful initialization and
return behavior, while ensuring every bootstrap error releases the database
handle first.
Comment on lines
+20
to
+59
| export function parseJournalRow( | ||
| sessionId: string, | ||
| row: JournalRow, | ||
| ): DurableEnvelope { | ||
| const envelope = envelopeSchema.parse({ | ||
| v: 1, | ||
| sessionId, | ||
| cursor: { epoch: row.epoch, seq: row.seq }, | ||
| ts: row.ts, | ||
| event: JSON.parse(row.eventJson), | ||
| }); | ||
| if (!isDurableEnvelope(envelope)) { | ||
| throw new Error(`chat journal row ${sessionId}:${row.seq} is not durable`); | ||
| } | ||
| return envelope; | ||
| } | ||
|
|
||
| export function readSince( | ||
| queries: ChatQueries, | ||
| sessionId: string, | ||
| cursor: Cursor, | ||
| ): ReplayResult { | ||
| const session = queries.session(sessionId); | ||
| if (!session) return { ok: false, reset: "session_not_found" }; | ||
| if (session.epoch !== cursor.epoch) { | ||
| return { ok: false, reset: "epoch_changed" }; | ||
| } | ||
|
|
||
| const last = queries.latestSeq(sessionId, session.epoch); | ||
| if (last === 0 && cursor.seq > 0) { | ||
| return { ok: false, reset: "journal_missing" }; | ||
| } | ||
| if (cursor.seq > last) return { ok: false, reset: "invalid_cursor" }; | ||
|
|
||
| const rows = queries.rowsSince(sessionId, session.epoch, cursor.seq); | ||
| return { | ||
| ok: true, | ||
| envelopes: rows.map((row) => parseJournalRow(sessionId, row)), | ||
| }; | ||
| } |
Contributor
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
A corrupt journal row throws instead of producing a reset.
parseJournalRow throws on two paths: JSON.parse(row.eventJson) raises SyntaxError for malformed JSON, and envelopeSchema.parse raises ZodError for a row that no longer matches the protocol. readSince calls it inside the rows.map at Line 57, so the exception escapes.
SessionSubscriptions.subscribe in packages/chat-runtime/src/stream/subscriptions.ts (lines 91-99) calls readSince without a try/catch and converts only ok: false into a reset envelope. One unreadable row therefore breaks the subscribe call rather than resetting the client, even though journal_missing exists for exactly that condition.
journal.test.ts (lines 149-155) shows that rows failing envelope validation are reachable in practice.
Convert parse failures into a reset reason.
🛡️ Proposed fix
const last = queries.latestSeq(sessionId, session.epoch);
if (last === 0 && cursor.seq > 0) {
return { ok: false, reset: "journal_missing" };
}
if (cursor.seq > last) return { ok: false, reset: "invalid_cursor" };
const rows = queries.rowsSince(sessionId, session.epoch, cursor.seq);
- return {
- ok: true,
- envelopes: rows.map((row) => parseJournalRow(sessionId, row)),
- };
+ try {
+ return {
+ ok: true,
+ envelopes: rows.map((row) => parseJournalRow(sessionId, row)),
+ };
+ } catch {
+ return { ok: false, reset: "journal_missing" };
+ }
}Apply the same guard around the page.map in readPage.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@packages/chat-runtime/src/journal/replay.ts` around lines 20 - 59, Update
readSince to guard the rows.map parsing path, including JSON.parse and
envelopeSchema.parse failures from parseJournalRow, and convert any unreadable
journal row into an unsuccessful ReplayResult with the appropriate
journal-missing reset reason. Preserve existing cursor and session validation
behavior, and keep parseJournalRow’s parsing responsibilities unchanged.
Comment on lines
+9
to
+21
| beforeEach(() => { | ||
| runtime = createTestRuntime(); | ||
| runtime.journal.open({ | ||
| sessionId: "s1", | ||
| workspaceId: "w1", | ||
| harness: "claude-code", | ||
| }); | ||
| runtime.journal.open({ | ||
| sessionId: "s2", | ||
| workspaceId: "w2", | ||
| harness: "codex", | ||
| }); | ||
| }); |
Contributor
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Dispose each test runtime.
Each beforeEach creates a SQLite runtime. The test suite does not call runtime.dispose(). Repeated test runs retain database handles until process exit.
Proposed fix
-import { beforeEach, describe, expect, test } from "bun:test";
+import { afterEach, beforeEach, describe, expect, test } from "bun:test";
@@
beforeEach(() => {
runtime = createTestRuntime();
@@
});
});
+
+ afterEach(async () => {
+ await runtime.dispose();
+ });📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| beforeEach(() => { | |
| runtime = createTestRuntime(); | |
| runtime.journal.open({ | |
| sessionId: "s1", | |
| workspaceId: "w1", | |
| harness: "claude-code", | |
| }); | |
| runtime.journal.open({ | |
| sessionId: "s2", | |
| workspaceId: "w2", | |
| harness: "codex", | |
| }); | |
| }); | |
| beforeEach(() => { | |
| runtime = createTestRuntime(); | |
| runtime.journal.open({ | |
| sessionId: "s1", | |
| workspaceId: "w1", | |
| harness: "claude-code", | |
| }); | |
| runtime.journal.open({ | |
| sessionId: "s2", | |
| workspaceId: "w2", | |
| harness: "codex", | |
| }); | |
| }); | |
| afterEach(async () => { | |
| await runtime.dispose(); | |
| }); |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@packages/chat-runtime/src/journal/sessions.test.ts` around lines 9 - 21,
Update the sessions test suite to dispose the SQLite runtime after every test:
add an afterEach cleanup that calls runtime.dispose(), while preserving the
existing beforeEach setup and journal session initialization.
Comment on lines
+99
to
+102
| cancelTurn(turnId?: string): void { | ||
| if (turnId && this.currentTurn && this.currentTurn.id !== turnId) return; | ||
| this.options.adapter.cancelTurn(); | ||
| } |
Contributor
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Forward cancelTurn only when the named turn is running.
The guard skips the cancel only when currentTurn exists and its id differs. Two cases still reach the adapter:
currentTurnis null, for example before the first turn event.currentTurn.id === turnIdbut the turn already completed or was interrupted.currentTurnis never cleared at Line 138.
FakeHarness ignores both cases, so liveSession.test.ts:194-207 still passes. A real adapter can reject the call or interrupt the next turn.
🐛 Proposed fix
cancelTurn(turnId?: string): void {
- if (turnId && this.currentTurn && this.currentTurn.id !== turnId) return;
+ const active =
+ this.currentTurn?.status === "running" ? this.currentTurn : null;
+ if (!active) return;
+ if (turnId && active.id !== turnId) return;
this.options.adapter.cancelTurn();
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| cancelTurn(turnId?: string): void { | |
| if (turnId && this.currentTurn && this.currentTurn.id !== turnId) return; | |
| this.options.adapter.cancelTurn(); | |
| } | |
| cancelTurn(turnId?: string): void { | |
| const active = | |
| this.currentTurn?.status === "running" ? this.currentTurn : null; | |
| if (!active) return; | |
| if (turnId && active.id !== turnId) return; | |
| this.options.adapter.cancelTurn(); | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@packages/chat-runtime/src/sessions/liveSession.ts` around lines 99 - 102,
Update LiveSession.cancelTurn to forward cancellation only when the requested
turn matches an existing, currently running turn. Track or validate the turn’s
active state, reject cancellation when currentTurn is absent or
completed/interrupted, and preserve the existing mismatch guard; ensure
currentTurn is cleared or marked inactive when the turn ends.
Drizzle mirrors host-service exactly (committed drizzle-kit generate migrations under src/db/drizzle, fail-loud migrate() at open, identical dep versions); query layer dissolved into per-module ownership with the same-transaction rollback test preserved. Folder-per-unit layout derived from the import graph; test doubles co-located (harness/fake); journal/sessions renamed projection to unclash with the sessions module. 50 tests green. Adds plans/chat-runtime-structure.md proposing the deeper journal-facade nesting for review. Claude-Session: https://claude.ai/code/session_01XRFCEPsfmJHu2SenDKYJ2h
Contributor
There was a problem hiding this comment.
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 (1)
packages/chat-runtime/src/commands/commands/commands.ts (1)
83-134: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winScope deduplication keys by command type.
The shared cache uses only
parsed.commandId. If a caller uses the same ID forcreateSessionandprompt,promptreturns the cachedCreateSessionResultand does not enqueue the prompt. Prefix the cache key with the command type for every handler.Proposed fix
- return options.dedupe.run(parsed.commandId, () => { + return options.dedupe.run(`createSession:${parsed.commandId}`, () => { // ... }); - return options.dedupe.run(parsed.commandId, () => + return options.dedupe.run(`prompt:${parsed.commandId}`, () => options.live.require(parsed.sessionId).prompt(parsed.content, parsed.clientId), ); - options.dedupe.run(parsed.commandId, () => { + options.dedupe.run(`cancelTurn:${parsed.commandId}`, () => { // ... }); - options.dedupe.run(parsed.commandId, () => { + options.dedupe.run(`respondToApproval:${parsed.commandId}`, () => { // ... }); - options.dedupe.run(parsed.commandId, () => { + options.dedupe.run(`setMode:${parsed.commandId}`, () => { // ... });🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/chat-runtime/src/commands/commands/commands.ts` around lines 83 - 134, Update every handler in the command object—createSession, prompt, cancelTurn, respondToApproval, and setMode—to prefix parsed.commandId with its command type before passing it to options.dedupe.run. Use a distinct, consistent prefix per handler so identical IDs used across different commands cannot share cached results, while preserving deduplication for repeated calls of the same command.
🧹 Nitpick comments (1)
packages/chat-runtime/src/testRuntime/testRuntime.ts (1)
27-29: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winUse a typed Bun database implementation.
createBunChatDbbypasses the sharedChatDbcontract withas unknown as ChatDbandas never, while runtime modules access Drizzle members such asdb.$client,db.insert, anddb.transaction. Define a Drizzle runtime type forChatDb, or parameterizeChatRuntimeOptions, socreateBunChatDbsatisfies the contract without assertions.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/chat-runtime/src/testRuntime/testRuntime.ts` around lines 27 - 29, Update createBunChatDb and the ChatDb typing used by ChatRuntimeOptions to use a typed Bun/Drizzle database implementation that exposes $client, insert, and transaction. Remove the as unknown as ChatDb and as never assertions, and ensure the resulting database instance satisfies the shared ChatDb contract while preserving the existing migration flow.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@packages/chat-runtime/src/db/createChatDb/createChatDb.ts`:
- Around line 35-37: The migration calls in createChatDb and testRuntime must
close their local SQLite clients when migrate() fails. In
packages/chat-runtime/src/db/createChatDb/createChatDb.ts#L35-L37 and
packages/chat-runtime/src/testRuntime/testRuntime.ts#L28-L30, wrap each
migrate() call in catch handling that closes sqlite and rethrows the original
migration error.
In `@packages/chat-runtime/src/stream/subscriptions/subscriptions.ts`:
- Around line 91-96: Update the replay loop in the subscription setup to apply
the same delta-filtering predicate used by deliver for live envelopes before
calling sink.send. Ensure subscribers with an empty or selective options.deltas
list receive only permitted replay envelopes, while preserving delivery of
non-delta envelopes.
In `@plans/chat-runtime-structure.md`:
- Line 12: Update the fenced directory-tree block in the documentation to
declare the text language identifier, using a text fence instead of an untyped
fence. Run the project’s lint fix command and verify the lint command completes
without output.
---
Outside diff comments:
In `@packages/chat-runtime/src/commands/commands/commands.ts`:
- Around line 83-134: Update every handler in the command object—createSession,
prompt, cancelTurn, respondToApproval, and setMode—to prefix parsed.commandId
with its command type before passing it to options.dedupe.run. Use a distinct,
consistent prefix per handler so identical IDs used across different commands
cannot share cached results, while preserving deduplication for repeated calls
of the same command.
---
Nitpick comments:
In `@packages/chat-runtime/src/testRuntime/testRuntime.ts`:
- Around line 27-29: Update createBunChatDb and the ChatDb typing used by
ChatRuntimeOptions to use a typed Bun/Drizzle database implementation that
exposes $client, insert, and transaction. Remove the as unknown as ChatDb and as
never assertions, and ensure the resulting database instance satisfies the
shared ChatDb contract while preserving the existing migration flow.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: a8ad247d-dff7-43c2-bfad-4f72ab54de5c
⛔ Files ignored due to path filters (1)
bun.lockis excluded by!**/*.lock
📒 Files selected for processing (51)
packages/chat-runtime/README.mdpackages/chat-runtime/drizzle.config.tspackages/chat-runtime/package.jsonpackages/chat-runtime/src/commands/commandDedupe/commandDedupe.test.tspackages/chat-runtime/src/commands/commandDedupe/commandDedupe.tspackages/chat-runtime/src/commands/commandDedupe/index.tspackages/chat-runtime/src/commands/commands/commands.test.tspackages/chat-runtime/src/commands/commands/commands.tspackages/chat-runtime/src/commands/commands/index.tspackages/chat-runtime/src/commands/index.tspackages/chat-runtime/src/db/createChatDb/createChatDb.tspackages/chat-runtime/src/db/createChatDb/index.tspackages/chat-runtime/src/db/drizzle/0000_nice_stellaris.sqlpackages/chat-runtime/src/db/drizzle/meta/0000_snapshot.jsonpackages/chat-runtime/src/db/drizzle/meta/_journal.jsonpackages/chat-runtime/src/db/index.tspackages/chat-runtime/src/db/schema/index.tspackages/chat-runtime/src/db/schema/schema.tspackages/chat-runtime/src/fixtures/fixtures.tspackages/chat-runtime/src/fixtures/index.tspackages/chat-runtime/src/harness/fake/index.tspackages/chat-runtime/src/harness/index.tspackages/chat-runtime/src/index.tspackages/chat-runtime/src/journal/epoch/epoch.tspackages/chat-runtime/src/journal/epoch/index.tspackages/chat-runtime/src/journal/index.tspackages/chat-runtime/src/journal/journal/index.tspackages/chat-runtime/src/journal/journal/journal.test.tspackages/chat-runtime/src/journal/journal/journal.tspackages/chat-runtime/src/projection/index.tspackages/chat-runtime/src/projection/projection.test.tspackages/chat-runtime/src/projection/projection.tspackages/chat-runtime/src/replay/index.tspackages/chat-runtime/src/replay/replay.test.tspackages/chat-runtime/src/replay/replay.tspackages/chat-runtime/src/replay/replayParity.test.tspackages/chat-runtime/src/sessions/index.tspackages/chat-runtime/src/sessions/liveSession/index.tspackages/chat-runtime/src/sessions/liveSession/liveSession.test.tspackages/chat-runtime/src/sessions/liveSession/liveSession.tspackages/chat-runtime/src/sessions/registry/index.tspackages/chat-runtime/src/sessions/registry/registry.tspackages/chat-runtime/src/stream/index.tspackages/chat-runtime/src/stream/subscriptions/index.tspackages/chat-runtime/src/stream/subscriptions/subscriptions.test.tspackages/chat-runtime/src/stream/subscriptions/subscriptions.tspackages/chat-runtime/src/testRuntime/index.tspackages/chat-runtime/src/testRuntime/testRuntime.tspackages/chat-runtime/src/testUtils/index.tspackages/chat-runtime/src/testUtils/testUtils.tsplans/chat-runtime-structure.md
🚧 Files skipped from review as they are similar to previous changes (2)
- packages/chat-runtime/package.json
- packages/chat-runtime/src/index.ts
Comment on lines
+35
to
+37
| migrate(db, { | ||
| migrationsFolder: options.migrationsFolder ?? DEFAULT_MIGRATIONS_FOLDER, | ||
| }); |
Contributor
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
rg -n -C 5 'migrate\(|\.close\(' \
packages/chat-runtime/src/db/createChatDb/createChatDb.ts \
packages/chat-runtime/src/testRuntime/testRuntime.tsRepository: superset-sh/superset
Length of output: 1183
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '--- repository files containing createChatDb/testRuntime ---\n'
git ls-files | rg '(^|/)createChatDb/\.(ts|tsx|js|jsx)$|(^|/)testRuntime\.(ts|tsx|js|jsx)$|packages/chat-runtime/src/(db/createChatDb|testRuntime)' || true
printf '\n--- createChatDb.ts ---\n'
cat -n packages/chat-runtime/src/db/createChatDb/createChatDb.ts | sed -n '1,120p'
printf '\n--- testRuntime candidates snippets ---\n'
for f in $(git ls-files | rg 'packages/chat-runtime/src/testRuntime/.+\.(ts|tsx|js|jsx)$|testRuntime\.(ts|tsx|js|jsx)$'); do
echo "FILE: $f"
rg -n -C 4 'createChatDb|migrate\(|sqlite|DEFAULT_MIGRATIONS_FOLDER|\.close\(' "$f" || true
doneRepository: superset-sh/superset
Length of output: 249
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '%s\n' '--- repository files containing createChatDb/testRuntime ---" packages/chat-runtime/src createChatDb testRuntime'
git ls-files | rg '(^|/)createChatDb/\.ts$|(^|/)createChatDb/\.tsx$|testRuntime\.(ts|tsx)$|packages/chat-runtime/src/(db/createChatDb|testRuntime)' || true
printf '\n%s\n' '--- createChatDb.ts ---'
cat -n packages/chat-runtime/src/db/createChatDb/createChatDb.ts | sed -n '1,120p'
printf '%s\n' '--- testRuntime candidates snippets ---'
git ls-files | rg 'packages/chat-runtime/src/testRuntime/.+\.(ts|tsx|js|jsx)$|testRuntime\.(ts|tsx|js|jsx)$' | while IFS= read -r f; do
echo "FILE: $f"
rg -n -C 4 'createChatDb|migrate\(|sqlite|DEFAULT_MIGRATIONS_FOLDER|\.close\(' "$f" || true
doneRepository: superset-sh/superset
Length of output: 3175
Close the SQLite client when migration fails.
If migrate() throws, each factory exits before returning the database handle. That leaves the opened SQLite client open until garbage collection, which can retain resources locks. Wrap migrate(), close the local sqlite client in catch, then rethrow the migration error.
packages/chat-runtime/src/db/createChatDb/createChatDb.ts#L35-L37packages/chat-runtime/src/testing/testRuntime/testRuntime.ts#L28-L30
📍 Affects 2 files
packages/chat-runtime/src/db/createChatDb/createChatDb.ts#L35-L37(this comment)packages/chat-runtime/src/testRuntime/testRuntime.ts#L28-L30
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@packages/chat-runtime/src/db/createChatDb/createChatDb.ts` around lines 35 -
37, The migration calls in createChatDb and testRuntime must close their local
SQLite clients when migrate() fails. In
packages/chat-runtime/src/db/createChatDb/createChatDb.ts#L35-L37 and
packages/chat-runtime/src/testRuntime/testRuntime.ts#L28-L30, wrap each
migrate() call in catch handling that closes sqlite and rethrows the original
migration error.
Comment on lines
+91
to
+96
| const replay = options.since | ||
| ? readSince(this.db, sessionId, options.since) | ||
| : readPage(this.db, sessionId, { limit: this.bootstrapLimit }); | ||
|
|
||
| if (replay.ok) { | ||
| for (const envelope of replay.envelopes) sink.send(envelope); |
Contributor
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Apply the delta filter to replay.
deliver filters live delta envelopes at Lines 157-160. The replay branch sends every envelope directly at Lines 95-96. A subscriber with deltas: [] or a selective delta list therefore receives delta envelopes during bootstrap, then rejects those channels after registration.
Use the same delivery predicate for replay and live envelopes.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@packages/chat-runtime/src/stream/subscriptions/subscriptions.ts` around lines
91 - 96, Update the replay loop in the subscription setup to apply the same
delta-filtering predicate used by deliver for live envelopes before calling
sink.send. Ensure subscribers with an empty or selective options.deltas list
receive only permitted replay envelopes, while preserving delivery of non-delta
envelopes.
|
|
||
| ## Ideal tree | ||
|
|
||
| ``` |
Contributor
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Add a language identifier to the fenced block.
markdownlint reports MD040 for this fence. Use text for the directory tree. This removes lint output.
As per coding guidelines, “Use bun run lint:fix after edits and verify bun run lint exits successfully before pushing; never push code that produces lint output.”
Proposed fix
-```
+```text📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| ``` |
🧰 Tools
🪛 markdownlint-cli2 (0.23.2)
[warning] 12-12: Fenced code blocks should have a language specified
(MD040, fenced-code-language)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@plans/chat-runtime-structure.md` at line 12, Update the fenced directory-tree
block in the documentation to declare the text language identifier, using a text
fence instead of an untyped fence. Run the project’s lint fix command and verify
the lint command completes without output.
Sources: Coding guidelines, Linters/SAST tools
Contributor
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
packages/chat-runtime/src/testing/testRuntime/testRuntime.ts (1)
1-33: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick winKeep the temporary test runtime path self-contained.
createTestRuntime()returns aChatRuntimetied to a temporarydataDirand Bun SQLite handle. Current callers close the runtime, but the directory/dataDir remains on disk. Add a cleanup function or test runtime disposer if repeated test runs should remove those temp directories.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/chat-runtime/src/testing/testRuntime/testRuntime.ts` around lines 1 - 33, Update the test runtime creation flow around createBunChatDb and createTestRuntime so the temporary dataDir is removed when the runtime is disposed. Expose or attach a cleanup/disposer that closes the Bun SQLite handle and recursively deletes the temporary directory, while preserving the existing ChatRuntime behavior and allowing callers to invoke cleanup reliably.packages/chat-runtime/src/testing/testUtils/testUtils.ts (1)
73-84: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winDo not convert replay failures into an empty journal.
journalEnvelopesshould return[]only whenruntime.sessions.get(sessionId)is absent. WhenreadSincereturnsok: false, this also hides database, migration, or replay errors such asepoch_changed,invalid_cursor, andjournal_missing, and tests can then compare a true failure against false empty journal state. Propagate the replay failure instead of mapping it to[].🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/chat-runtime/src/testing/testUtils/testUtils.ts` around lines 73 - 84, Update journalEnvelopes so it returns [] only when runtime.sessions.get(sessionId) is absent; when readSince returns ok: false, propagate that replay failure instead of converting it to an empty envelope list. Adjust the function’s return type and callers as needed to represent the readSince failure while preserving successful replay envelopes.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@packages/chat-runtime/src/testing/testRuntime/testRuntime.ts`:
- Around line 1-33: Update the test runtime creation flow around createBunChatDb
and createTestRuntime so the temporary dataDir is removed when the runtime is
disposed. Expose or attach a cleanup/disposer that closes the Bun SQLite handle
and recursively deletes the temporary directory, while preserving the existing
ChatRuntime behavior and allowing callers to invoke cleanup reliably.
In `@packages/chat-runtime/src/testing/testUtils/testUtils.ts`:
- Around line 73-84: Update journalEnvelopes so it returns [] only when
runtime.sessions.get(sessionId) is absent; when readSince returns ok: false,
propagate that replay failure instead of converting it to an empty envelope
list. Adjust the function’s return type and callers as needed to represent the
readSince failure while preserving successful replay envelopes.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 9e710e39-6bc6-417a-b00f-7a1b448b40c2
📒 Files selected for processing (15)
packages/chat-runtime/README.mdpackages/chat-runtime/src/commands/commands/commands.test.tspackages/chat-runtime/src/harness/fake/fakeHarness.test.tspackages/chat-runtime/src/journal/journal/journal.test.tspackages/chat-runtime/src/projection/projection.test.tspackages/chat-runtime/src/replay/replay.test.tspackages/chat-runtime/src/replay/replayParity.test.tspackages/chat-runtime/src/sessions/liveSession/liveSession.test.tspackages/chat-runtime/src/stream/subscriptions/subscriptions.test.tspackages/chat-runtime/src/testing/fixtures/fixtures.tspackages/chat-runtime/src/testing/fixtures/index.tspackages/chat-runtime/src/testing/testRuntime/index.tspackages/chat-runtime/src/testing/testRuntime/testRuntime.tspackages/chat-runtime/src/testing/testUtils/index.tspackages/chat-runtime/src/testing/testUtils/testUtils.ts
🚧 Files skipped from review as they are similar to previous changes (7)
- packages/chat-runtime/src/projection/projection.test.ts
- packages/chat-runtime/src/replay/replayParity.test.ts
- packages/chat-runtime/README.md
- packages/chat-runtime/src/harness/fake/fakeHarness.test.ts
- packages/chat-runtime/src/journal/journal/journal.test.ts
- packages/chat-runtime/src/commands/commands/commands.test.ts
- packages/chat-runtime/src/sessions/liveSession/liveSession.test.ts
This was referenced Aug 5, 2026
Merged
…g/ layout The doc still described test helpers living at src root and asserted there was no testing/ folder, which the shipped package contradicts. Restate the rule the code actually follows: helpers split on ownership, not on being test code. A helper with an owner co-locates with it (fake/ inside harness/); helpers no single module owns — fixtures, testUtils, testRuntime — live in src/testing/, internal, relative-import, with no package export. Fixes all three places that disagreed: the src-root fixtures.ts/testUtils.ts entries, the stray testRuntime note under sessions/, and the "No testing/ folder" bullet. Claude-Session: https://claude.ai/code/session_01XRFCEPsfmJHu2SenDKYJ2h
Review of PR #6167 surfaced one crash-class chain and several ways a failure left the runtime in a state it could not recover from. The adapter pump had no rejection handler, so a throwing harness stream became an unhandled rejection and then cascaded: the rejected pump made dispose() reject, which made the registry's Promise.all reject, which skipped closing subscriptions and the database. The pump now journals the truth instead — running turn to interrupted, an error notice, session dead — and disposal settles every adapter before surfacing the first failure, with the runtime's own teardown in a finally chain so the socket and SQLite handles always close. deliver() left awaitingTurn set when adapter.prompt threw, which made hasPendingWork() permanently true and queued every later prompt forever. An unreadable journal row threw out of subscribe() instead of producing the journal_missing reset that already existed for exactly that case, so one corrupt row took down the subscription rather than resyncing the client. Pagination now rejects a before cursor past the head for the same reason readSince already did. SubscriptionHub trusted sinks: one throwing send propagated into publish, into the adapter pump, and back into the unhandled-rejection path above. A throwing sink is now dropped and closed, and its session's coalescer is released with it — deltas published to a session with no subscribers are dropped rather than buffered, which is what the protocol says they are. createSession committed the journal row before the harness was validated, leaving an unusable session visible to listSessions on failure. The harness is checked first and the row is compensated if start still fails. Command dedupe is keyed by command kind, so one client-minted id reused across two verbs no longer returns the first verb's cached result. Epoch recovery reset the epoch but left the projection's status, title and queued count describing a journal that no longer exists. Claude-Session: https://claude.ai/code/session_01XRFCEPsfmJHu2SenDKYJ2h
Contributor
There was a problem hiding this comment.
5 issues found across 16 files (changes from recent commits).
Prompt for AI agents (unresolved issues)
Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.
<file name="packages/chat-runtime/src/sessions/registry/registry.ts">
<violation number="1" location="packages/chat-runtime/src/sessions/registry/registry.ts:64">
P2: A teardown failure during startup recovery is silently discarded, so a harness can remain running while `create()` reports only the original startup error. Logging the cleanup failure with the session id would preserve observability instead of swallowing it.
(Based on your team's feedback about handling rejected async cleanup.)</violation>
</file>
<file name="packages/chat-runtime/src/replay/replay.ts">
<violation number="1" location="packages/chat-runtime/src/replay/replay.ts:43">
P2: Corrupted journal rows now produce a client reset but silently lose the parse error and exact `sessionId:seq`, leaving durable-data corruption difficult to diagnose or repair. Logging the row identity and original error before returning `null` would preserve the intended failure containment while making the failure observable.</violation>
</file>
<file name="packages/chat-runtime/src/index.ts">
<violation number="1" location="packages/chat-runtime/src/index.ts:84">
P2: A throwing sink close can mask the original live-session teardown failure during runtime shutdown and leave later subscribers unclosed. Cleanup errors should be contained or aggregated while continuing every sink close, preserving the first failure for `dispose()`.</violation>
</file>
<file name="packages/chat-runtime/src/sessions/liveSession/liveSession.ts">
<violation number="1" location="packages/chat-runtime/src/sessions/liveSession/liveSession.ts:79">
P2: Failure recording errors are silently swallowed here: if `fail()` throws while journaling or publishing the dead state, the original pump error and the recording error disappear and the session can remain persisted as starting/running. Log both errors with the session ID (or expose a failure reporter) before stopping so journal and sink failures remain diagnosable.
(Based on your team's feedback about surfacing asynchronous and failure-handling errors.) .</violation>
<violation number="2" location="packages/chat-runtime/src/sessions/liveSession/liveSession.ts:202">
P2: A failed pump leaves persisted queued prompts marked as pending after the in-memory queue is discarded, so dead sessions report work that will never run. Append resolving snapshots for the dropped prompts (or another explicit terminal representation) before clearing the queue so the journal projection reaches zero.</violation>
</file>
Tip: Review your code locally with the cubic CLI to iterate faster.
Re-trigger cubic
| }); | ||
| } catch (error) { | ||
| this.live.delete(options.sessionId); | ||
| void session.dispose().catch(() => undefined); |
Contributor
There was a problem hiding this comment.
P2: A teardown failure during startup recovery is silently discarded, so a harness can remain running while create() reports only the original startup error. Logging the cleanup failure with the session id would preserve observability instead of swallowing it.
(Based on your team's feedback about handling rejected async cleanup.)
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At packages/chat-runtime/src/sessions/registry/registry.ts, line 64:
<comment>A teardown failure during startup recovery is silently discarded, so a harness can remain running while `create()` reports only the original startup error. Logging the cleanup failure with the session id would preserve observability instead of swallowing it.
(Based on your team's feedback about handling rejected async cleanup.) </comment>
<file context>
@@ -45,12 +52,18 @@ export class LiveSessionRegistry {
+ });
+ } catch (error) {
+ this.live.delete(options.sessionId);
+ void session.dispose().catch(() => undefined);
+ throw error;
+ }
</file context>
Suggested change
| void session.dispose().catch(() => undefined); | |
| void session.dispose().catch((disposeError: unknown) => { | |
| console.warn( | |
| `failed to dispose chat session ${options.sessionId} after startup failure`, | |
| disposeError, | |
| ); | |
| }); |
| for (const row of rows) { | ||
| try { | ||
| envelopes.push(parseJournalRow(row)); | ||
| } catch { |
Contributor
There was a problem hiding this comment.
P2: Corrupted journal rows now produce a client reset but silently lose the parse error and exact sessionId:seq, leaving durable-data corruption difficult to diagnose or repair. Logging the row identity and original error before returning null would preserve the intended failure containment while making the failure observable.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At packages/chat-runtime/src/replay/replay.ts, line 43:
<comment>Corrupted journal rows now produce a client reset but silently lose the parse error and exact `sessionId:seq`, leaving durable-data corruption difficult to diagnose or repair. Logging the row identity and original error before returning `null` would preserve the intended failure containment while making the failure observable.</comment>
<file context>
@@ -35,6 +35,18 @@ export function parseJournalRow(row: JournalRow): DurableEnvelope {
+ for (const row of rows) {
+ try {
+ envelopes.push(parseJournalRow(row));
+ } catch {
+ return null;
+ }
</file context>
| await live.disposeAll(); | ||
| } finally { | ||
| try { | ||
| subscriptions.dispose(); |
Contributor
There was a problem hiding this comment.
P2: A throwing sink close can mask the original live-session teardown failure during runtime shutdown and leave later subscribers unclosed. Cleanup errors should be contained or aggregated while continuing every sink close, preserving the first failure for dispose().
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At packages/chat-runtime/src/index.ts, line 84:
<comment>A throwing sink close can mask the original live-session teardown failure during runtime shutdown and leave later subscribers unclosed. Cleanup errors should be contained or aggregated while continuing every sink close, preserving the first failure for `dispose()`.</comment>
<file context>
@@ -77,9 +77,15 @@ export function createChatRuntime(options: ChatRuntimeOptions): ChatRuntime {
+ await live.disposeAll();
+ } finally {
+ try {
+ subscriptions.dispose();
+ } finally {
+ db.$client.close();
</file context>
Comment on lines
+79
to
+81
| } catch { | ||
| this.stopped = true; | ||
| } |
Contributor
There was a problem hiding this comment.
P2: Failure recording errors are silently swallowed here: if fail() throws while journaling or publishing the dead state, the original pump error and the recording error disappear and the session can remain persisted as starting/running. Log both errors with the session ID (or expose a failure reporter) before stopping so journal and sink failures remain diagnosable.
(Based on your team's feedback about surfacing asynchronous and failure-handling errors.) .
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At packages/chat-runtime/src/sessions/liveSession/liveSession.ts, line 79:
<comment>Failure recording errors are silently swallowed here: if `fail()` throws while journaling or publishing the dead state, the original pump error and the recording error disappear and the session can remain persisted as starting/running. Log both errors with the session ID (or expose a failure reporter) before stopping so journal and sink failures remain diagnosable.
(Based on your team's feedback about surfacing asynchronous and failure-handling errors.) .</comment>
<file context>
@@ -72,7 +72,15 @@ export class LiveSession {
+ (error: unknown) => {
+ try {
+ this.fail(error);
+ } catch {
+ this.stopped = true;
+ }
</file context>
Suggested change
| } catch { | |
| this.stopped = true; | |
| } | |
| } catch (failureError) { | |
| console.warn("chat session failure could not be recorded", { | |
| sessionId: this.options.sessionId, | |
| error, | |
| failureError, | |
| }); | |
| this.stopped = true; | |
| } |
| }; | ||
| this.appendDurable({ type: "turn", turn: this.currentTurn }); | ||
| } | ||
| this.queue.length = 0; |
Contributor
There was a problem hiding this comment.
P2: A failed pump leaves persisted queued prompts marked as pending after the in-memory queue is discarded, so dead sessions report work that will never run. Append resolving snapshots for the dropped prompts (or another explicit terminal representation) before clearing the queue so the journal projection reaches zero.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At packages/chat-runtime/src/sessions/liveSession/liveSession.ts, line 202:
<comment>A failed pump leaves persisted queued prompts marked as pending after the in-memory queue is discarded, so dead sessions report work that will never run. Append resolving snapshots for the dropped prompts (or another explicit terminal representation) before clearing the queue so the journal projection reaches zero.</comment>
<file context>
@@ -171,7 +179,41 @@ export class LiveSession {
+ };
+ this.appendDurable({ type: "turn", turn: this.currentTurn });
+ }
+ this.queue.length = 0;
+ this.awaitingTurn = null;
+ this.appendDurable({
</file context>
Suggested change
| this.queue.length = 0; | |
| for (const pending of this.queue.splice(0)) { | |
| this.appendDurable({ | |
| type: "item", | |
| item: withoutQueued(pending.item), | |
| turnId: turn?.id ?? this.mintId(), | |
| }); | |
| } |
This was referenced Aug 5, 2026
Merged
Merged
This branch was previously deployed
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
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.
Stacked on #6166. The mountable chat runtime: journal with epoch-qualified replay, LiveSession adapter pump with FIFO prompt queue, SubscriptionHub (replay-then-live, delta filtering, coalescing), commands with commandId dedupe, fake harness. 50 tests incl. whole-envelope reconnect equality. Also folds the Codex desktop teardown findings into the ship plan.
https://claude.ai/code/session_01XRFCEPsfmJHu2SenDKYJ2h
Summary by cubic
Adds a standalone chat session runtime as
@superset/chat-runtimewith an append-only journal, epoch-qualified replay, live session pumping, and replay-then-live subscriptions. Now hardened against adapter/journal/sink failures to keep sessions consistent and recoverable; uses Drizzle-backed SQLite with committed migrations, and 50 tests verify live===replay parity.New Features
@superset/chat-runtimeexportingcreateChatRuntime({ dataDir, harnesses, openDatabase? }); SQLitechat.dbvia Drizzle (migrations on open, WAL) with an injectable DB seam (better-sqlite3in prod,bun:sqlitein tests).readSince/readPagereturn reset signals on epoch mismatch.user_messageIDs, and pass-through for cancel/approval/mode.commandIdLRU dedupe; harness registry and a scripted fake harness for testing.Bug Fixes
adapter.promptthrows.createSessionvalidates the harness before committing; compensates the row if start fails.Written for commit d9fe416. Summary will update on new commits.
Summary by CodeRabbit
New Features
Bug Fixes
Documentation