From f395930018addfd550d99bb4d2b7b6bfeabc9632 Mon Sep 17 00:00:00 2001 From: me2seeks Date: Tue, 18 Aug 2026 10:32:28 +0800 Subject: [PATCH 01/17] fix(cli): resubscribe instead of failing when a turn consumer falls behind MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A TUI turn event consumer that fell more than MAX_PENDING_EVENTS_PER_TURN behind had its stream failed permanently while the Host turn kept running; the channel kept draining frames into the dead queue and nothing recovered until the terminal transcript refresh. Shed offset-bearing deltas (healed by the next canonical resync or text completion) and evict the oldest sheddable delta to make room for other events so terminal records always land, and notify the channel once per lag episode. The channel retires the healthy-but-lagged subscription through the existing recovery path — the same resubscribe a Host slow-consumer eviction triggers, and the TUI equivalent of the Desktop subscription owner (#2630). Fixes #3180 Generated-by: Maka --- .../cli/src/runtime-host-session-channel.ts | 58 +++++++++++++++++-- 1 file changed, 54 insertions(+), 4 deletions(-) diff --git a/packages/cli/src/runtime-host-session-channel.ts b/packages/cli/src/runtime-host-session-channel.ts index 6ce3fe35b5..26b9fe096b 100644 --- a/packages/cli/src/runtime-host-session-channel.ts +++ b/packages/cli/src/runtime-host-session-channel.ts @@ -66,6 +66,7 @@ export class RuntimeHostSessionChannel { readonly #pendingResolvedInteractions: InteractionPendingSnapshot[] = []; readonly #pendingTerminalTurns: TerminalTurnSnapshot[] = []; readonly #failedSubscriptions = new WeakSet(); + readonly #retiringSubscriptions = new WeakSet(); #projector: RuntimeHostSessionProjector | undefined; #ready = false; #activated = false; @@ -266,6 +267,10 @@ export class RuntimeHostSessionChannel { if (!this.#closing) throw new Error('Runtime Host Session subscription ended unexpectedly'); } catch (error) { if (this.#closing || this.#subscription !== subscription) return; + // A subscription retired because a turn consumer fell behind is closed + // deliberately; its pump must not turn that expected close into a + // channel failure. + if (this.#retiringSubscriptions.has(subscription)) return; if (this.#canRecover(error)) { this.#failedSubscriptions.add(subscription); if (!this.#ready) return; @@ -506,13 +511,25 @@ export class RuntimeHostSessionChannel { #queue(turnId: string): SessionEventQueue { let queue = this.#turns.get(turnId); if (!queue) { - queue = new SessionEventQueue(); + queue = new SessionEventQueue(() => this.#noteTurnConsumerLagging()); this.#turns.set(turnId, queue); if (this.#failure) queue.fail(this.#failure); } return queue; } + /** + * A turn consumer that cannot keep up is a slow client: retire the healthy + * subscription through the same recovery path a Host eviction would take so + * the session re-syncs from canonical state instead of dying mid-turn. + */ + #noteTurnConsumerLagging(): void { + if (this.#closing || this.#failure || !this.#ready) return; + const subscription = this.#subscription; + this.#retiringSubscriptions.add(subscription); + this.#scheduleRecovery(subscription); + } + #fail(error: unknown): void { if (this.#failure) return; this.#failure = error instanceof Error ? error : new Error(String(error)); @@ -522,6 +539,7 @@ export class RuntimeHostSessionChannel { class SessionEventQueue implements AsyncIterable, AsyncIterator { readonly #items: SessionEvent[] = []; + readonly #onLag: () => void; #waiting: | { resolve(value: IteratorResult): void; @@ -531,6 +549,11 @@ class SessionEventQueue implements AsyncIterable, AsyncIterator void) { + this.#onLag = onLag; + } [Symbol.asyncIterator](): AsyncIterator { return this; @@ -538,7 +561,10 @@ class SessionEventQueue implements AsyncIterable, AsyncIterator> { const item = this.#items.shift(); - if (item) return Promise.resolve({ done: false, value: item }); + if (item) { + if (this.#items.length === 0) this.#lagging = false; + return Promise.resolve({ done: false, value: item }); + } if (this.#error !== undefined) return Promise.reject(this.#error); if (this.#done || this.#finishAfterItems) { this.#done = true; @@ -560,12 +586,32 @@ class SessionEventQueue implements AsyncIterable, AsyncIterator= MAX_PENDING_EVENTS_PER_TURN) { - this.fail(new Error('Runtime Host Session event consumer is too slow')); - return; + // A consumer that falls behind must not kill the stream. Shed + // offset-bearing deltas (the next canonical resync or completion heals + // them) and make room for every other event so terminal records always + // land; the channel resubscribes to re-sync state, like the Desktop + // subscription owner does (#2630). + if (isSheddableDelta(event)) { + this.#noteLag(); + return; + } + const shedIndex = this.#items.findIndex(isSheddableDelta); + if (shedIndex === -1) { + this.#noteLag(); + return; + } + this.#items.splice(shedIndex, 1); + this.#noteLag(); } this.#items.push(event); } + #noteLag(): void { + if (this.#lagging) return; + this.#lagging = true; + this.#onLag(); + } + finish(): void { if (this.#done || this.#error !== undefined) return; this.#finishAfterItems = true; @@ -585,6 +631,10 @@ class SessionEventQueue implements AsyncIterable, AsyncIterator Date: Tue, 18 Aug 2026 10:32:28 +0800 Subject: [PATCH 02/17] test(cli): cover turn consumer lag recovery Flood an unconsumed turn stream past its bound: the channel resubscribes, the stream never rejects, live deltas continue after recovery, and terminal events still land while deltas are shed. Generated-by: Maka --- .../runtime-host-session-driver.test.ts | 107 ++++++++++++++++++ 1 file changed, 107 insertions(+) diff --git a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts index c35f0267b0..7a1f4010e7 100644 --- a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts +++ b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts @@ -1663,3 +1663,110 @@ async function waitFor(predicate: () => boolean): Promise { } assert.fail('Timed out waiting for fake Host state'); } + +describe('turn consumer lag recovery (#3180)', () => { + async function floodTurnStream( + subscription: InstanceType, + count: number, + startOffset: number, + ): Promise { + let offset = startOffset; + for (let index = 0; index < count; index += 1) { + const text = `x${String(index).padStart(4, '0')}`; + subscription.push(deltaFrame(index + 1, 'turn-1', offset, text)); + offset += text.length; + if (index % 64 === 63) await delay(0); + } + await delay(0); + } + + async function waitForSubscriptions(connection: FakeConnection, count: number): Promise { + const deadline = Date.now() + WAIT_BUDGET_MS; + while (connection.openedSubscriptions !== count && Date.now() < deadline) await delay(5); + assert.equal(connection.openedSubscriptions, count); + } + + test('resubscribes instead of failing when a turn event consumer falls behind', async () => { + const initial = new FakeSubscription( + continuitySnapshot(), + Promise.resolve([assistantMessage('turn-1', 'Hello')]), + ); + const replacement = new FakeSubscription( + continuitySnapshot({ projectionRevision: 2 }), + Promise.resolve([assistantMessage('turn-1', 'Hello')]), + 'subscription-2', + ); + const connection = new FakeConnection([initial, replacement], true); + const driver = createRuntimeHostMakaSessionDriver({ + connection: connection.value, + cwd: '/tmp', + llmConnectionSlug: 'openai-main', + model: 'gpt-5', + now: () => 50, + }); + const switched = await driver.switchSession('session-1'); + assert.ok(switched.activeTurn); + + // Flood the unconsumed turn stream past its 1024-event bound. + await floodTurnStream(initial, 1_100, 5); + + // The channel retires the lagged subscription and resubscribes instead of + // failing the stream. + await waitForSubscriptions(connection, 2); + + // Drain part of the backlog, then confirm live events keep flowing. + const iterator = switched.activeTurn.events[Symbol.asyncIterator](); + for (let index = 0; index < 100; index += 1) { + const result = await iterator.next(); + assert.equal(result.done, false); + } + replacement.push(deltaFrame(1, 'turn-1', 5, ' world', 'subscription-2')); + let seen = ''; + for (let index = 0; index < 1_100 && seen !== ' world'; index += 1) { + const result = await iterator.next(); + assert.equal(result.done, false); + seen = (result.value as { text?: string }).text ?? ''; + } + assert.equal(seen, ' world'); + }); + + test('lands terminal events while shedding deltas from a lagging consumer', async () => { + const initial = new FakeSubscription( + continuitySnapshot(), + Promise.resolve([assistantMessage('turn-1', 'Hello')]), + ); + const replacement = new FakeSubscription( + continuitySnapshot({ projectionRevision: 2 }), + Promise.resolve([assistantMessage('turn-1', 'Hello')]), + 'subscription-2', + ); + const connection = new FakeConnection([initial, replacement], true); + const driver = createRuntimeHostMakaSessionDriver({ + connection: connection.value, + cwd: '/tmp', + llmConnectionSlug: 'openai-main', + model: 'gpt-5', + now: () => 50, + }); + const switched = await driver.switchSession('session-1'); + assert.ok(switched.activeTurn); + + await floodTurnStream(initial, 1_100, 5); + await waitForSubscriptions(connection, 2); + + replacement.push(projectionFrame(1, completedTurn('turn-1', 'run-1'), 2, 'subscription-2')); + const iterator = switched.activeTurn.events[Symbol.asyncIterator](); + let completed = false; + let done = false; + for (let index = 0; index < 1_200 && !done; index += 1) { + const result = await iterator.next(); + if (result.done) { + done = true; + break; + } + if ((result.value as { type?: string }).type === 'complete') completed = true; + } + assert.ok(completed, 'terminal complete event survived the lagged backlog'); + assert.ok(done, 'turn stream finished cleanly'); + }); +}); From e9bfe39995ab92c4f71a916b61fb1573aeb818ad Mon Sep 17 00:00:00 2001 From: me2seeks Date: Tue, 18 Aug 2026 11:07:04 +0800 Subject: [PATCH 03/17] fix(cli): guarantee terminal admission and compact lagged delta backlog on resync Address qodo-code-review findings on #3181: - A full queue with no sheddable delta silently dropped an incoming complete/error/abort, letting the consumer reach end-of-stream without a terminal outcome. Terminal outcomes now evict the oldest event in that corner, so they always land. - After lag recovery, a still-full queue kept shedding the fresh post-resync stream behind stale deltas the canonical replacement had already superseded. Lagging queues now drop their unseen sheddable backlog when the canonical replacement lands. Generated-by: Maka --- .../cli/src/runtime-host-session-channel.ts | 34 ++++++++++++++++--- 1 file changed, 29 insertions(+), 5 deletions(-) diff --git a/packages/cli/src/runtime-host-session-channel.ts b/packages/cli/src/runtime-host-session-channel.ts index 26b9fe096b..3b7ee3d2a0 100644 --- a/packages/cli/src/runtime-host-session-channel.ts +++ b/packages/cli/src/runtime-host-session-channel.ts @@ -366,6 +366,10 @@ export class RuntimeHostSessionChannel { this.#now, this.#subscription.activeAssistantStreams, ); + // Deltas a lagging consumer has not seen are superseded by this canonical + // resync; keeping them would shed the fresh post-recovery stream behind + // them. + for (const queue of this.#turns.values()) queue.shedLaggedDeltas(); if (!replacedLiveState) { for (const event of this.#projector.seedActive(false)) this.#emit(event); return false; @@ -588,24 +592,40 @@ class SessionEventQueue implements AsyncIterable, AsyncIterator= MAX_PENDING_EVENTS_PER_TURN) { // A consumer that falls behind must not kill the stream. Shed // offset-bearing deltas (the next canonical resync or completion heals - // them) and make room for every other event so terminal records always - // land; the channel resubscribes to re-sync state, like the Desktop - // subscription owner does (#2630). + // them) and make room for every other event; the channel resubscribes + // to re-sync state, like the Desktop subscription owner does (#2630). if (isSheddableDelta(event)) { this.#noteLag(); return; } const shedIndex = this.#items.findIndex(isSheddableDelta); - if (shedIndex === -1) { + if (shedIndex !== -1) { + this.#items.splice(shedIndex, 1); + } else if (isTerminalOutcome(event)) { + // Terminal outcomes always land, even when the backlog holds no + // delta to evict: without one the consumer reaches end-of-stream + // without a result. + this.#items.shift(); + } else { this.#noteLag(); return; } - this.#items.splice(shedIndex, 1); this.#noteLag(); } this.#items.push(event); } + /** + * Drop sheddable deltas a lagging consumer has not seen. Called when the + * channel re-syncs from canonical state, which heals the omitted ranges. + */ + shedLaggedDeltas(): void { + if (!this.#lagging) return; + for (let index = this.#items.length - 1; index >= 0; index -= 1) { + if (isSheddableDelta(this.#items[index]!)) this.#items.splice(index, 1); + } + } + #noteLag(): void { if (this.#lagging) return; this.#lagging = true; @@ -635,6 +655,10 @@ function isSheddableDelta(event: SessionEvent): boolean { return event.type === 'text_delta' || event.type === 'thinking_delta'; } +function isTerminalOutcome(event: SessionEvent): boolean { + return event.type === 'complete' || event.type === 'abort' || event.type === 'error'; +} + function sameTerminalTurn( previous: SessionContinuitySnapshot['rootTurn'], next: TerminalTurnSnapshot, From 0304d195b488c7959745ddc149a7ab839a9853d4 Mon Sep 17 00:00:00 2001 From: me2seeks Date: Tue, 18 Aug 2026 11:07:04 +0800 Subject: [PATCH 04/17] test(cli): pin terminal admission and resync backlog compaction Both new tests fail against the pre-review channel: the resync marker delta is shed behind the uncompacted backlog, and the terminal outcome is dropped from an all-tool-event backlog. Deterministic ordering: wait for the canonical resync before producing the terminal frame, so the backlog is still full when it is emitted. Generated-by: Maka --- .../runtime-host-session-driver.test.ts | 133 ++++++++++++------ 1 file changed, 87 insertions(+), 46 deletions(-) diff --git a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts index 7a1f4010e7..a4bc46aa38 100644 --- a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts +++ b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts @@ -1680,13 +1680,24 @@ describe('turn consumer lag recovery (#3180)', () => { await delay(0); } + async function floodToolStream( + subscription: InstanceType, + count: number, + ): Promise { + for (let index = 0; index < count; index += 1) { + subscription.push(toolStartFrame(index + 1, index)); + if (index % 64 === 63) await delay(0); + } + await delay(0); + } + async function waitForSubscriptions(connection: FakeConnection, count: number): Promise { const deadline = Date.now() + WAIT_BUDGET_MS; while (connection.openedSubscriptions !== count && Date.now() < deadline) await delay(5); assert.equal(connection.openedSubscriptions, count); } - test('resubscribes instead of failing when a turn event consumer falls behind', async () => { + function lagRecoveryFixture() { const initial = new FakeSubscription( continuitySnapshot(), Promise.resolve([assistantMessage('turn-1', 'Hello')]), @@ -1697,6 +1708,7 @@ describe('turn consumer lag recovery (#3180)', () => { 'subscription-2', ); const connection = new FakeConnection([initial, replacement], true); + const resynced = deferred(); const driver = createRuntimeHostMakaSessionDriver({ connection: connection.value, cwd: '/tmp', @@ -1704,69 +1716,98 @@ describe('turn consumer lag recovery (#3180)', () => { model: 'gpt-5', now: () => 50, }); + driver.subscribeTranscriptReplacements!((_sessionId, _turnId, _messages, reason) => { + if (reason === 'reconnect') resynced.resolve(); + }); + return { initial, replacement, connection, driver, resynced }; + } + + async function drainUntilDone(events: AsyncIterable): Promise { + const iterator = events[Symbol.asyncIterator](); + let completed = false; + for (let index = 0; index < 1_200; index += 1) { + const result = await iterator.next(); + if (result.done) return completed; + if ((result.value as { type?: string }).type === 'complete') completed = true; + } + return false; + } + + test('resubscribes instead of failing when a turn event consumer falls behind', async () => { + const { initial, replacement, connection, driver, resynced } = lagRecoveryFixture(); const switched = await driver.switchSession('session-1'); assert.ok(switched.activeTurn); // Flood the unconsumed turn stream past its 1024-event bound. await floodTurnStream(initial, 1_100, 5); - // The channel retires the lagged subscription and resubscribes instead of - // failing the stream. + // The channel retires the lagged subscription, resubscribes, and compacts + // the sheddable backlog the canonical resync supersedes. await waitForSubscriptions(connection, 2); + await resynced.promise; - // Drain part of the backlog, then confirm live events keep flowing. - const iterator = switched.activeTurn.events[Symbol.asyncIterator](); - for (let index = 0; index < 100; index += 1) { - const result = await iterator.next(); - assert.equal(result.done, false); - } + // The stream never rejected, and live events land right away. replacement.push(deltaFrame(1, 'turn-1', 5, ' world', 'subscription-2')); - let seen = ''; - for (let index = 0; index < 1_100 && seen !== ' world'; index += 1) { - const result = await iterator.next(); - assert.equal(result.done, false); - seen = (result.value as { text?: string }).text ?? ''; - } - assert.equal(seen, ' world'); + assert.equal((await nextEvent(switched.activeTurn.events)).text, ' world'); }); test('lands terminal events while shedding deltas from a lagging consumer', async () => { - const initial = new FakeSubscription( - continuitySnapshot(), - Promise.resolve([assistantMessage('turn-1', 'Hello')]), - ); - const replacement = new FakeSubscription( - continuitySnapshot({ projectionRevision: 2 }), - Promise.resolve([assistantMessage('turn-1', 'Hello')]), - 'subscription-2', - ); - const connection = new FakeConnection([initial, replacement], true); - const driver = createRuntimeHostMakaSessionDriver({ - connection: connection.value, - cwd: '/tmp', - llmConnectionSlug: 'openai-main', - model: 'gpt-5', - now: () => 50, - }); + const { initial, replacement, connection, driver, resynced } = lagRecoveryFixture(); const switched = await driver.switchSession('session-1'); assert.ok(switched.activeTurn); await floodTurnStream(initial, 1_100, 5); await waitForSubscriptions(connection, 2); + await resynced.promise; replacement.push(projectionFrame(1, completedTurn('turn-1', 'run-1'), 2, 'subscription-2')); - const iterator = switched.activeTurn.events[Symbol.asyncIterator](); - let completed = false; - let done = false; - for (let index = 0; index < 1_200 && !done; index += 1) { - const result = await iterator.next(); - if (result.done) { - done = true; - break; - } - if ((result.value as { type?: string }).type === 'complete') completed = true; - } - assert.ok(completed, 'terminal complete event survived the lagged backlog'); - assert.ok(done, 'turn stream finished cleanly'); + await delay(0); + assert.ok( + await drainUntilDone(switched.activeTurn.events), + 'terminal complete event survived the lagged delta backlog', + ); + }); + + test('admits a terminal outcome when the lagged backlog holds no deltas', async () => { + const { initial, replacement, connection, driver, resynced } = lagRecoveryFixture(); + const switched = await driver.switchSession('session-1'); + assert.ok(switched.activeTurn); + + // Fill the bound with non-delta events: nothing sheddable to evict. + await floodToolStream(initial, 1_100); + await waitForSubscriptions(connection, 2); + await resynced.promise; + + // The terminal outcome must land even though no delta can be evicted; + // process the frame before draining so the backlog is still full. + replacement.push(projectionFrame(1, completedTurn('turn-1', 'run-1'), 2, 'subscription-2')); + await delay(0); + assert.ok( + await drainUntilDone(switched.activeTurn.events), + 'terminal complete event was admitted over a non-delta backlog', + ); }); }); + +function toolStartFrame( + sequence: number, + index: number, + subscriptionId = 'subscription-1', +): SubscriptionFrame { + return { + kind: 'subscription.session_event', + hostEpoch: 'host-1', + subscriptionId, + sequence, + sessionId: 'session-1', + runId: 'run-1', + event: { + type: 'tool_start', + id: `tool-${index}`, + turnId: 'turn-1', + ts: 10, + toolUseId: `tool-${index}`, + toolName: 'Bash', + }, + }; +} From 87f46fca0a80230441883b9e5e41539770d30698 Mon Sep 17 00:00:00 2001 From: me2seeks Date: Tue, 18 Aug 2026 11:29:13 +0800 Subject: [PATCH 05/17] fix(cli): re-arm lag detection with hysteresis after the consumer drains CodeRabbit review on #3181: after a lag recovery over a non-delta backlog, the latch stayed on until the queue fully emptied, so fresh output shed while the consumer was still behind could never schedule another canonical recovery. Re-arm once the backlog drains to half the bound: a consumer making progress gets later episodes recovered, while a wedged consumer never drains and cannot loop resubscribes. Generated-by: Maka --- packages/cli/src/runtime-host-session-channel.ts | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/packages/cli/src/runtime-host-session-channel.ts b/packages/cli/src/runtime-host-session-channel.ts index 3b7ee3d2a0..c6359cabb1 100644 --- a/packages/cli/src/runtime-host-session-channel.ts +++ b/packages/cli/src/runtime-host-session-channel.ts @@ -24,6 +24,7 @@ import type { MakaPreparedSessionTurn } from './session-driver.js'; const MAX_PENDING_FRAMES = 512; const MAX_PENDING_EVENTS_PER_TURN = 1_024; +const LAG_REARM_PENDING_EVENTS = MAX_PENDING_EVENTS_PER_TURN / 2; export interface RuntimeHostSessionChannelOpenResult { channel: RuntimeHostSessionChannel; @@ -566,7 +567,10 @@ class SessionEventQueue implements AsyncIterable, AsyncIterator> { const item = this.#items.shift(); if (item) { - if (this.#items.length === 0) this.#lagging = false; + // Re-arm with hysteresis: a consumer that has drained half the backlog + // is making progress, so a later lag episode may trigger another + // recovery; a wedged consumer never drains and cannot loop resubscribes. + if (this.#items.length <= LAG_REARM_PENDING_EVENTS) this.#lagging = false; return Promise.resolve({ done: false, value: item }); } if (this.#error !== undefined) return Promise.reject(this.#error); From d8f4329b7b7efa5861a24cde7e2150e68a2d78bb Mon Sep 17 00:00:00 2001 From: me2seeks Date: Tue, 18 Aug 2026 11:29:13 +0800 Subject: [PATCH 06/17] test(cli): cover repeated lag recovery after consumer progress Sends fresh output after a non-delta-backlog recovery, then lags the consumer a second time and expects a third subscription. Fails without the hysteresis re-arm: the latch never clears while events remain queued. Generated-by: Maka --- .../runtime-host-session-driver.test.ts | 71 +++++++++++++++++++ 1 file changed, 71 insertions(+) diff --git a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts index a4bc46aa38..6416c2e0ac 100644 --- a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts +++ b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts @@ -1787,6 +1787,77 @@ describe('turn consumer lag recovery (#3180)', () => { 'terminal complete event was admitted over a non-delta backlog', ); }); + + test('recovers again when the consumer lags again after making progress', async () => { + const initial = new FakeSubscription( + continuitySnapshot(), + Promise.resolve([assistantMessage('turn-1', 'Hello')]), + ); + const second = new FakeSubscription( + continuitySnapshot({ projectionRevision: 2 }), + Promise.resolve([assistantMessage('turn-1', 'Hello')]), + 'subscription-2', + ); + const third = new FakeSubscription( + continuitySnapshot({ projectionRevision: 3 }), + Promise.resolve([assistantMessage('turn-1', 'Hello')]), + 'subscription-3', + ); + const connection = new FakeConnection([initial, second, third], true); + let resyncs = 0; + const driver = createRuntimeHostMakaSessionDriver({ + connection: connection.value, + cwd: '/tmp', + llmConnectionSlug: 'openai-main', + model: 'gpt-5', + now: () => 50, + }); + driver.subscribeTranscriptReplacements!((_sessionId, _turnId, _messages, reason) => { + if (reason === 'reconnect') resyncs += 1; + }); + const switched = await driver.switchSession('session-1'); + assert.ok(switched.activeTurn); + + // First lag episode over a non-delta backlog: nothing to compact, and the + // latch stays on while the consumer remains behind. + await floodToolStream(initial, 1_100); + await waitForSubscriptions(connection, 2); + await waitFor(() => resyncs === 1); + + // The consumer drains past the hysteresis watermark, re-arming lag + // detection, and fresh output flows again. One hundred events stay queued + // behind the delta, so the backlog never empties. + const iterator = switched.activeTurn.events[Symbol.asyncIterator](); + for (let index = 0; index < 600; index += 1) { + const result = await iterator.next(); + assert.equal(result.done, false); + } + second.push(deltaFrame(1, 'turn-1', 5, ' world', 'subscription-2')); + for (let index = 0; index < 100; index += 1) { + second.push(toolStartFrame(100 + index, 2_000 + index, 'subscription-2')); + } + await delay(0); + let fresh = ''; + for (let index = 0; index < 425; index += 1) { + const result = await iterator.next(); + assert.equal(result.done, false); + fresh = (result.value as { text?: string }).text ?? ''; + } + assert.equal(fresh, ' world'); + + // A second lag episode is a new episode, not a dead latch: it triggers a + // fresh canonical resync. + await floodToolStream(second, 1_100); + await waitForSubscriptions(connection, 3); + await waitFor(() => resyncs === 2); + + third.push(projectionFrame(1, completedTurn('turn-1', 'run-1'), 3, 'subscription-3')); + await delay(0); + assert.ok( + await drainUntilDone(switched.activeTurn.events), + 'stream still completes after repeated lag recoveries', + ); + }); }); function toolStartFrame( From afb2aea55ce8ea0ec442200795e7490c547d3e8e Mon Sep 17 00:00:00 2001 From: me2seeks Date: Tue, 18 Aug 2026 11:41:41 +0800 Subject: [PATCH 07/17] test(cli): keep the second subscription's frame stream contiguous MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CodeRabbit review on #3181: the repeated-recovery test sent sequences 1, then 100..199, then a flood restarting at 1 with the first subscription's id — a stream the real ClientSessionSubscription would reject as a sequence gap, masked by the fake. Thread the subscription id and a starting sequence through floodToolStream so every fake stream stays valid. Generated-by: Maka --- .../__tests__/runtime-host-session-driver.test.ts | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts index 6416c2e0ac..398db7623b 100644 --- a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts +++ b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts @@ -1683,9 +1683,13 @@ describe('turn consumer lag recovery (#3180)', () => { async function floodToolStream( subscription: InstanceType, count: number, + subscriptionId = 'subscription-1', + startSequence = 1, ): Promise { for (let index = 0; index < count; index += 1) { - subscription.push(toolStartFrame(index + 1, index)); + subscription.push( + toolStartFrame(startSequence + index, startSequence + index, subscriptionId), + ); if (index % 64 === 63) await delay(0); } await delay(0); @@ -1834,7 +1838,7 @@ describe('turn consumer lag recovery (#3180)', () => { } second.push(deltaFrame(1, 'turn-1', 5, ' world', 'subscription-2')); for (let index = 0; index < 100; index += 1) { - second.push(toolStartFrame(100 + index, 2_000 + index, 'subscription-2')); + second.push(toolStartFrame(2 + index, 2_000 + index, 'subscription-2')); } await delay(0); let fresh = ''; @@ -1846,8 +1850,8 @@ describe('turn consumer lag recovery (#3180)', () => { assert.equal(fresh, ' world'); // A second lag episode is a new episode, not a dead latch: it triggers a - // fresh canonical resync. - await floodToolStream(second, 1_100); + // fresh canonical resync. The stream stays contiguous on `second`. + await floodToolStream(second, 1_100, 'subscription-2', 102); await waitForSubscriptions(connection, 3); await waitFor(() => resyncs === 2); From 051ebfb2026357895566affcf3a1ad5e2b09f83b Mon Sep 17 00:00:00 2001 From: me2seeks Date: Tue, 18 Aug 2026 15:12:47 +0800 Subject: [PATCH 08/17] fix(cli): shed lagged tool output deltas so tool results land Astro-Han review on #3181 (P2): during a tool-output flood the queue filled with non-sheddable tool_output_delta events, so an incoming tool_result had nothing to evict and was silently dropped, leaving the live tool card stuck at "running" until the durable transcript healed. tool_output_delta is sheddable by design: the protocol documents its chunks as transient UI updates with a monotonic per-tool seq that renderers de-dupe and order by, and the terminal tool_result plus the durable transcript remain the authoritative output. Shedding them under lag matches the text_delta story; a shed range leaves a display gap, never corruption. The remaining boundary is documented at the drop branch: a non-delta, non-terminal event behind a backlog with nothing sheddable (e.g. an all-control backlog) is still dropped, and the durable transcript heals the terminal state. Generated-by: Maka --- .../runtime-host-session-driver.test.ts | 94 +++++++++++++++++++ .../cli/src/runtime-host-session-channel.ts | 30 ++++-- 2 files changed, 118 insertions(+), 6 deletions(-) diff --git a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts index 398db7623b..f0df3e6235 100644 --- a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts +++ b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts @@ -1695,6 +1695,21 @@ describe('turn consumer lag recovery (#3180)', () => { await delay(0); } + async function floodToolOutput( + subscription: InstanceType, + count: number, + subscriptionId = 'subscription-1', + startSequence = 1, + ): Promise { + for (let index = 0; index < count; index += 1) { + subscription.push( + toolOutputDeltaFrame(startSequence + index, startSequence + index, subscriptionId), + ); + if (index % 64 === 63) await delay(0); + } + await delay(0); + } + async function waitForSubscriptions(connection: FakeConnection, count: number): Promise { const deadline = Date.now() + WAIT_BUDGET_MS; while (connection.openedSubscriptions !== count && Date.now() < deadline) await delay(5); @@ -1792,6 +1807,39 @@ describe('turn consumer lag recovery (#3180)', () => { ); }); + test('sheds lagged tool output deltas so the tool result and terminal outcome land', async () => { + const { initial, replacement, connection, driver, resynced } = lagRecoveryFixture(); + const switched = await driver.switchSession('session-1'); + assert.ok(switched.activeTurn); + + // A noisy tool floods the unconsumed stream with seq-ordered output + // deltas, the realistic way a consumer falls behind. + await floodToolOutput(initial, 1_100); + await waitForSubscriptions(connection, 2); + await resynced.promise; + + // The canonical resync compacts the unseen tool deltas, so the tool + // result lands instead of being dropped behind a full non-delta backlog + // (which would leave the live card stuck at "running" until the durable + // transcript heals it). + replacement.push(toolResultFrame(1, 'subscription-2')); + replacement.push(projectionFrame(2, completedTurn('turn-1', 'run-1'), 2, 'subscription-2')); + await delay(0); + + let sawToolResult = false; + const iterator = switched.activeTurn.events[Symbol.asyncIterator](); + for (let index = 0; index < 1_200; index += 1) { + const result = await iterator.next(); + if (result.done) break; + if ((result.value as { type?: string }).type === 'tool_result') sawToolResult = true; + if ((result.value as { type?: string }).type === 'complete') { + assert.ok(sawToolResult, 'tool_result landed ahead of the terminal outcome'); + return; + } + } + assert.fail('stream ended without the terminal complete event'); + }); + test('recovers again when the consumer lags again after making progress', async () => { const initial = new FakeSubscription( continuitySnapshot(), @@ -1886,3 +1934,49 @@ function toolStartFrame( }, }; } + +function toolOutputDeltaFrame( + sequence: number, + seq: number, + subscriptionId = 'subscription-1', +): SubscriptionFrame { + return { + kind: 'subscription.session_event', + hostEpoch: 'host-1', + subscriptionId, + sequence, + sessionId: 'session-1', + runId: 'run-1', + event: { + type: 'tool_output_delta', + id: `output-${seq}`, + turnId: 'turn-1', + ts: 10, + toolUseId: 'tool-1', + seq, + stream: 'stdout', + chunk: `chunk-${seq}`, + redacted: false, + createdAt: 10, + }, + }; +} + +function toolResultFrame(sequence: number, subscriptionId = 'subscription-1'): SubscriptionFrame { + return { + kind: 'subscription.session_event', + hostEpoch: 'host-1', + subscriptionId, + sequence, + sessionId: 'session-1', + runId: 'run-1', + event: { + type: 'tool_result', + id: 'result-tool-1', + turnId: 'turn-1', + ts: 11, + toolUseId: 'tool-1', + status: 'completed', + }, + }; +} diff --git a/packages/cli/src/runtime-host-session-channel.ts b/packages/cli/src/runtime-host-session-channel.ts index c6359cabb1..e84eaf3011 100644 --- a/packages/cli/src/runtime-host-session-channel.ts +++ b/packages/cli/src/runtime-host-session-channel.ts @@ -594,10 +594,13 @@ class SessionEventQueue implements AsyncIterable, AsyncIterator= MAX_PENDING_EVENTS_PER_TURN) { - // A consumer that falls behind must not kill the stream. Shed - // offset-bearing deltas (the next canonical resync or completion heals - // them) and make room for every other event; the channel resubscribes - // to re-sync state, like the Desktop subscription owner does (#2630). + // A consumer that falls behind must not kill the stream. Shed deltas + // (text/thinking ranges are healed by the next canonical resync or + // completion; tool_output_delta chunks are seq-deduped transient UI + // updates healed by the terminal tool_result and the durable + // transcript) and make room for every other event; the channel + // resubscribes to re-sync state, like the Desktop subscription owner + // does (#2630). if (isSheddableDelta(event)) { this.#noteLag(); return; @@ -611,6 +614,10 @@ class SessionEventQueue implements AsyncIterable, AsyncIterator, AsyncIterator, AsyncIterator Date: Tue, 18 Aug 2026 15:13:19 +0800 Subject: [PATCH 09/17] test(cli): pin the lag hysteresis watermark boundary Astro-Han review on #3181 (P3s): the re-arm boundary was only indirectly covered. The new test drains a full non-delta backlog to one event above the watermark (513 pending) and asserts a fresh overflow does not resubscribe, then drains to the watermark (512 pending) and asserts the next overflow is treated as a new lag episode. Also pin two reviewed behaviors in comments: the retiring-subscription guard swallowing a genuine error racing the deliberate close (the replacement pump re-surfaces real failures via #fail), and the per-queue lag escalating to a session-wide recovery (benign superset: the resync heals every turn, and the latch plus hysteresis prevent resubscribe loops). Generated-by: Maka --- .../runtime-host-session-driver.test.ts | 58 +++++++++++++++++++ .../cli/src/runtime-host-session-channel.ts | 11 +++- 2 files changed, 68 insertions(+), 1 deletion(-) diff --git a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts index f0df3e6235..3e6a45bc2b 100644 --- a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts +++ b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts @@ -1840,6 +1840,64 @@ describe('turn consumer lag recovery (#3180)', () => { assert.fail('stream ended without the terminal complete event'); }); + test('re-arms lag detection exactly at the hysteresis watermark', async () => { + const initial = new FakeSubscription( + continuitySnapshot(), + Promise.resolve([assistantMessage('turn-1', 'Hello')]), + ); + const second = new FakeSubscription( + continuitySnapshot({ projectionRevision: 2 }), + Promise.resolve([assistantMessage('turn-1', 'Hello')]), + 'subscription-2', + ); + const third = new FakeSubscription( + continuitySnapshot({ projectionRevision: 3 }), + Promise.resolve([assistantMessage('turn-1', 'Hello')]), + 'subscription-3', + ); + const connection = new FakeConnection([initial, second, third], true); + let resyncs = 0; + const driver = createRuntimeHostMakaSessionDriver({ + connection: connection.value, + cwd: '/tmp', + llmConnectionSlug: 'openai-main', + model: 'gpt-5', + now: () => 50, + }); + driver.subscribeTranscriptReplacements!((_sessionId, _turnId, _messages, reason) => { + if (reason === 'reconnect') resyncs += 1; + }); + const switched = await driver.switchSession('session-1'); + assert.ok(switched.activeTurn); + + // Latch the lag flag with a full non-delta backlog: all 1_024 queued + // events stay because nothing is sheddable. + await floodToolStream(initial, 1_100); + await waitForSubscriptions(connection, 2); + await waitFor(() => resyncs === 1); + + // Draining to one event above the watermark (513 pending) must NOT + // re-arm: a fresh overflow on the still-latched queue is the same lag + // episode and triggers no new recovery. The flood refills the backlog + // to the bound. + const iterator = switched.activeTurn.events[Symbol.asyncIterator](); + for (let index = 0; index < 511; index += 1) { + assert.equal((await iterator.next()).done, false); + } + await floodToolStream(second, 600, 'subscription-2', 2_000); + await delay(20); + assert.equal(connection.openedSubscriptions, 2, 'lag latch held above the watermark'); + + // Draining the refilled backlog down to the watermark (512 pending) + // re-arms: the next overflow is a new lag episode and resubscribes again. + for (let index = 0; index < 512; index += 1) { + assert.equal((await iterator.next()).done, false); + } + await floodToolStream(second, 600, 'subscription-2', 3_000); + await waitForSubscriptions(connection, 3); + await waitFor(() => resyncs === 2); + }); + test('recovers again when the consumer lags again after making progress', async () => { const initial = new FakeSubscription( continuitySnapshot(), diff --git a/packages/cli/src/runtime-host-session-channel.ts b/packages/cli/src/runtime-host-session-channel.ts index e84eaf3011..96cd21b746 100644 --- a/packages/cli/src/runtime-host-session-channel.ts +++ b/packages/cli/src/runtime-host-session-channel.ts @@ -270,7 +270,10 @@ export class RuntimeHostSessionChannel { if (this.#closing || this.#subscription !== subscription) return; // A subscription retired because a turn consumer fell behind is closed // deliberately; its pump must not turn that expected close into a - // channel failure. + // channel failure. The guard also drops a genuine error racing the + // deliberate close on this pump; that is safe because the replacement + // subscription's own pump and recovery path re-surface any real + // failure through #fail. if (this.#retiringSubscriptions.has(subscription)) return; if (this.#canRecover(error)) { this.#failedSubscriptions.add(subscription); @@ -527,6 +530,12 @@ export class RuntimeHostSessionChannel { * A turn consumer that cannot keep up is a slow client: retire the healthy * subscription through the same recovery path a Host eviction would take so * the session re-syncs from canonical state instead of dying mid-turn. + * + * Note this escalates a single lagging queue to a session-wide recovery. + * That is a benign superset even when the lagging queue belongs to an + * abandoned old turn: the resync heals every turn's state, and the + * per-queue lag latch plus hysteresis keep a wedged consumer from looping + * resubscribes. */ #noteTurnConsumerLagging(): void { if (this.#closing || this.#failure || !this.#ready) return; From 1d5097c7220d170c309935c9b8529e1c04b3bb8c Mon Sep 17 00:00:00 2001 From: me2seeks Date: Tue, 18 Aug 2026 17:45:07 +0800 Subject: [PATCH 10/17] fix(runtime-host): coalesce queued assistant deltas instead of evicting MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Dogfooding this branch surfaced the loop the resubscribe fix could spin: an active turn's thinking/text delta flood outpaces the one-awaited-send flush, the 32-frame subscriber budget overflowed, and the Host evicted the subscription as slow_consumer within seconds of every resubscribe — while the recovering channel had a permanently fatal path (clean iterator end mid-catch-up) and a silently freezing one. The coordinator now folds a queued assistant delta into its queued tail when it continues the same stream contiguously: projectors apply deltas by absolute startOffset, so a merged frame carries byte-identical content, and the absorbed frame never spends a sequence. Eviction stays the backstop for genuinely undrainable backlogs (covered by alternating-stream tests). The channel treats a live stream that ends without subscription.closed as connection_closed, routing it through resync recovery instead of failing the session. Co-Authored-By: Maka Generated-by: Maka --- .../runtime-host-session-driver.test.ts | 37 ++++++ .../cli/src/runtime-host-session-channel.ts | 11 +- .../src/__tests__/connection-session.test.ts | 8 +- .../session-continuity-coordinator.test.ts | 124 +++++++++++++++++- .../server/session-continuity-coordinator.ts | 54 ++++++++ 5 files changed, 227 insertions(+), 7 deletions(-) diff --git a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts index 3e6a45bc2b..a9f6565176 100644 --- a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts +++ b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts @@ -1840,6 +1840,43 @@ describe('turn consumer lag recovery (#3180)', () => { assert.fail('stream ended without the terminal complete event'); }); + test('resubscribes when the live stream ends without a terminal close', async () => { + const initial = new FakeSubscription( + continuitySnapshot(), + Promise.resolve([assistantMessage('turn-1', 'Hello')]), + ); + const replacement = new FakeSubscription( + continuitySnapshot({ projectionRevision: 2 }), + Promise.resolve([assistantMessage('turn-1', 'Hello world')]), + 'subscription-2', + ); + const connection = new FakeConnection([initial, replacement], true); + const driver = createRuntimeHostMakaSessionDriver({ + connection: connection.value, + cwd: '/tmp', + llmConnectionSlug: 'openai-main', + model: 'gpt-5', + now: () => 50, + }); + const switched = await driver.switchSession('session-1'); + assert.ok(switched.activeTurn); + const transcript = deferred(); + driver.subscribeTranscriptReplacements!((_sessionId, _turnId, messages, reason) => { + assert.equal(reason, 'reconnect'); + transcript.resolve(messages); + }); + + // A clean iterator end with no subscription.closed frame — e.g. the Host + // evicted the subscription as a slow consumer while the channel was still + // buffering the catch-up transcript — used to fail the channel + // permanently. It must resubscribe and continue the live stream instead. + await initial.close(); + assert.deepEqual(await transcript.promise, [assistantMessage('turn-1', 'Hello world')]); + assert.equal(connection.openedSubscriptions, 2); + replacement.push(deltaFrame(1, 'turn-1', 11, '!', 'subscription-2')); + assert.equal((await nextEvent(switched.activeTurn.events)).text, '!'); + }); + test('re-arms lag detection exactly at the hysteresis watermark', async () => { const initial = new FakeSubscription( continuitySnapshot(), diff --git a/packages/cli/src/runtime-host-session-channel.ts b/packages/cli/src/runtime-host-session-channel.ts index 96cd21b746..ed1443edc7 100644 --- a/packages/cli/src/runtime-host-session-channel.ts +++ b/packages/cli/src/runtime-host-session-channel.ts @@ -265,7 +265,16 @@ export class RuntimeHostSessionChannel { this.#accept(frame); } } - if (!this.#closing) throw new Error('Runtime Host Session subscription ended unexpectedly'); + // A stream that ends without a subscription.closed frame is a broken + // live channel, not a terminal state: the Host may have torn the + // subscription down mid-recovery (e.g. slow_consumer eviction while the + // transcript reload was still buffering). Route it through the same + // resync recovery as an explicit close instead of killing the channel. + if (!this.#closing) + throw new RuntimeHostSubscriptionError( + 'connection_closed', + 'Runtime Host Session subscription ended unexpectedly', + ); } catch (error) { if (this.#closing || this.#subscription !== subscription) return; // A subscription retired because a turn consumer fell behind is closed diff --git a/packages/runtime-host/src/__tests__/connection-session.test.ts b/packages/runtime-host/src/__tests__/connection-session.test.ts index eea76d9a08..a8fd36fb62 100644 --- a/packages/runtime-host/src/__tests__/connection-session.test.ts +++ b/packages/runtime-host/src/__tests__/connection-session.test.ts @@ -989,11 +989,13 @@ test('evicting one slow subscription keeps sibling subscriptions and requests us }; try { + // Alternate message streams so the queued deltas cannot coalesce: this + // exercises eviction for a genuinely undrainable backlog. for (let index = 1; index <= 32; index += 1) { await coordinator.acceptRuntimeEvent( 'slow-session', 'run-slow-session', - connectionTextEvent('slow-session', index), + connectionTextEvent('slow-session', index, `message-slow-${index % 2}`), ); } await withTimeout(writeBlocked.promise, 1_000, 'slow subscription never blocked in-flight'); @@ -1422,13 +1424,13 @@ function canonicalProjection(sessionId: string): CanonicalSessionProjection { }; } -function connectionTextEvent(sessionId: string, index: number) { +function connectionTextEvent(sessionId: string, index: number, messageId?: string) { return { type: 'text_delta' as const, id: `event-${sessionId}-${index}`, turnId: `turn-${sessionId}`, ts: index, - messageId: `message-${sessionId}`, + messageId: messageId ?? `message-${sessionId}`, text: `chunk-${index}`, }; } diff --git a/packages/runtime-host/src/__tests__/session-continuity-coordinator.test.ts b/packages/runtime-host/src/__tests__/session-continuity-coordinator.test.ts index 0730b43a3b..536e831d86 100644 --- a/packages/runtime-host/src/__tests__/session-continuity-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/session-continuity-coordinator.test.ts @@ -625,8 +625,14 @@ test('slow subscriber receives a terminal eviction without delaying another subs const fast = await open(coordinator, 'connection-fast'); fastConnection.activate(fast.subscriptionId); + // Alternate streams so queued deltas cannot coalesce: this exercises the + // eviction path for a genuinely undrainable backlog. for (let index = 1; index <= 32; index += 1) { - await coordinator.acceptRuntimeEvent(SESSION_ID, 'run-1', textEvent(index)); + await coordinator.acceptRuntimeEvent( + SESSION_ID, + 'run-1', + textEvent(index, `message-${index % 2}`), + ); } slowConnection.activate(slow.subscriptionId); await waitFor(() => slowSink.frames.length === 1 && fastSink.frames.length === 32); @@ -645,6 +651,118 @@ test('slow subscriber receives a terminal eviction without delaying another subs coordinator.close(); }); +test('coalesces queued assistant deltas instead of evicting a slow subscriber', async () => { + const coordinator = new SessionContinuityCoordinator( + HOST_EPOCH, + async () => canonical(), + new SessionAdmissionGate(), + ); + const slowSink = new RecordingSink(); + const fastSink = new RecordingSink(); + const slowConnection = coordinator.attachConnection('connection-slow', slowSink); + const fastConnection = coordinator.attachConnection('connection-fast', fastSink); + const slow = await open(coordinator, 'connection-slow'); + const fast = await open(coordinator, 'connection-fast'); + fastConnection.activate(fast.subscriptionId); + + // A same-stream delta flood that used to overflow the 32-frame budget and + // evict the subscriber before it ever activated. + for (let index = 1; index <= 64; index += 1) { + await coordinator.acceptRuntimeEvent(SESSION_ID, 'run-1', textEvent(index)); + } + slowConnection.activate(slow.subscriptionId); + await waitFor(() => slowSink.frames.length === 1 && fastSink.frames.length === 64); + + // The lagging subscriber receives one merged, content-identical delta: no + // eviction, absolute offsets preserved, no sequence spent on absorbed + // frames. + const merged = slowSink.frames[0]; + assert.equal(merged?.kind, 'subscription.session_delta'); + if (merged?.kind !== 'subscription.session_delta') return; + assert.equal(merged.sequence, 1); + assert.equal(merged.delta.startOffset, 0); + assert.equal( + merged.delta.text, + Array.from({ length: 64 }, (_, index) => `chunk-${index + 1}`).join(''), + ); + + // The next enqueue continues the sequence exactly where the merged frame + // left it, and the absolute offset continues the stream. + await coordinator.acceptRuntimeEvent(SESSION_ID, 'run-1', textEvent(65)); + await waitFor(() => slowSink.frames.length === 2); + const next = slowSink.frames[1]; + assert.equal(next?.kind, 'subscription.session_delta'); + if (next?.kind !== 'subscription.session_delta') return; + assert.equal(next.sequence, 2); + assert.equal(next.delta.startOffset, merged.delta.text.length); + assert.equal(next.delta.text, 'chunk-65'); + coordinator.close(); +}); + +test('keeps stream, kind, and completion boundaries when coalescing deltas', async () => { + const coordinator = new SessionContinuityCoordinator( + HOST_EPOCH, + async () => canonical(), + new SessionAdmissionGate(), + ); + const sink = new RecordingSink(); + const connection = coordinator.attachConnection('connection-1', sink); + const opened = await open(coordinator, 'connection-1'); + + await coordinator.acceptRuntimeEvent(SESSION_ID, 'run-1', textEvent(1, 'message-1')); + await coordinator.acceptRuntimeEvent(SESSION_ID, 'run-1', textEvent(2, 'message-1')); + // A thinking delta on the same message is a different delta kind: no merge. + await coordinator.acceptRuntimeEvent( + SESSION_ID, + 'run-1', + thinkingEvent('thinking_delta', 'thinking-1', 'think'), + ); + // A different message stream: no merge. + await coordinator.acceptRuntimeEvent(SESSION_ID, 'run-1', textEvent(3, 'message-2')); + await coordinator.acceptRuntimeEvent(SESSION_ID, 'run-1', textEvent(4, 'message-2')); + // Completion closes the stream and must land as its own frame. + await coordinator.acceptRuntimeEvent( + SESSION_ID, + 'run-1', + textCompleteEvent('message-1', 'chunk-1chunk-2'), + ); + + connection.activate(opened.subscriptionId); + await waitFor(() => sink.frames.length === 4); + assert.deepEqual( + sink.frames.map((frame) => + frame.kind === 'subscription.session_delta' + ? { + sequence: frame.sequence, + kind: frame.delta.kind, + messageId: frame.delta.messageId, + text: frame.delta.text, + complete: frame.delta.complete === true, + } + : frame.kind, + ), + [ + { + sequence: 1, + kind: 'text', + messageId: 'message-1', + text: 'chunk-1chunk-2', + complete: false, + }, + { sequence: 2, kind: 'thinking', messageId: 'thinking-1', text: 'think', complete: false }, + { + sequence: 3, + kind: 'text', + messageId: 'message-2', + text: 'chunk-3chunk-4', + complete: false, + }, + { sequence: 4, kind: 'text', messageId: 'message-1', text: '', complete: true }, + ], + ); + coordinator.close(); +}); + test('removal closes every Session subscriber at the admitted sequence boundary', async () => { const admission = new SessionAdmissionGate(); const coordinator = new SessionContinuityCoordinator( @@ -1882,13 +2000,13 @@ function pendingInteraction() { }; } -function textEvent(index: number) { +function textEvent(index: number, messageId = 'message-1') { return { type: 'text_delta' as const, id: `event-${index}`, turnId: 'turn-1', ts: index, - messageId: 'message-1', + messageId, text: `chunk-${index}`, }; } diff --git a/packages/runtime-host/src/server/session-continuity-coordinator.ts b/packages/runtime-host/src/server/session-continuity-coordinator.ts index bc72efba25..77a663374c 100644 --- a/packages/runtime-host/src/server/session-continuity-coordinator.ts +++ b/packages/runtime-host/src/server/session-continuity-coordinator.ts @@ -1352,6 +1352,33 @@ export class SessionContinuityCoordinator implements SessionContinuityService { return; } const terminalBytes = terminalFrameByteBudget(subscriber, this.#hostEpoch); + // Assistant text/thinking floods arrive far faster than the + // one-awaited-send-at-a-time flush can drain them, and the queue budget + // exists to bound memory, not to force eviction. Fold a delta into the + // queued tail when it continues the same stream: projectors apply deltas + // by absolute startOffset, so a merged frame carries byte-identical + // content, and the absorbed frame never spends a sequence, keeping later + // frames contiguous. The in-flight head frame is never touched. + const tail = subscriber.queue[subscriber.queue.length - 1]; + if (tail && (!subscriber.pumping || subscriber.queue.length > 1)) { + const mergedText = mergeableAssistantDeltaText(tail.frame, frame); + if (mergedText !== undefined && tail.frame.kind === 'subscription.session_delta') { + const merged: SubscriptionFrame = { + ...tail.frame, + delta: { ...tail.frame.delta, text: mergedText }, + }; + const mergedEncodedBytes = encodeProtocolMessage(merged).byteLength; + if ( + subscriber.queuedBytes - tail.encodedBytes + mergedEncodedBytes + terminalBytes <= + MAX_SUBSCRIBER_QUEUED_BYTES + ) { + tail.frame = merged; + subscriber.queuedBytes += mergedEncodedBytes - tail.encodedBytes; + tail.encodedBytes = mergedEncodedBytes; + return; + } + } + } if ( subscriber.queue.length >= MAX_SUBSCRIBER_QUEUED_FRAMES - 1 || subscriber.queuedBytes + encodedBytes + terminalBytes > MAX_SUBSCRIBER_QUEUED_BYTES @@ -1731,6 +1758,33 @@ function terminalFrameByteBudget(subscriber: Subscriber, hostEpoch: string): num ); } +/** + * Returns the concatenated text when `next` continues `tail`'s assistant + * stream contiguously, making the two frames safe to ship as one. Reset and + * completion frames never merge: a reset must land on its own boundary and a + * completion closes the stream. + */ +function mergeableAssistantDeltaText( + tail: SubscriptionFrame, + next: SubscriptionFrame, +): string | undefined { + if (tail.kind !== 'subscription.session_delta' || next.kind !== 'subscription.session_delta') + return undefined; + const a = tail.delta; + const b = next.delta; + if ( + a.kind !== b.kind || + a.turnId !== b.turnId || + a.runId !== b.runId || + a.messageId !== b.messageId + ) + return undefined; + if (a.complete === true || b.complete === true || a.reset === true || b.reset === true) + return undefined; + if (a.startOffset + a.text.length !== b.startOffset) return undefined; + return a.text + b.text; +} + function immutableClone(value: T): T { return deepFreeze(structuredClone(value)); } From 59c83e1cea41fc73ea0c10966d1e4f8f65f4c6fe Mon Sep 17 00:00:00 2001 From: me2seeks Date: Tue, 18 Aug 2026 19:19:16 +0800 Subject: [PATCH 11/17] fix(runtime-host): bound coalesced session deltas Generated-by: Maka --- .../session-continuity-coordinator.test.ts | 47 +++++++++++++++++++ .../server/session-continuity-coordinator.ts | 10 +++- 2 files changed, 56 insertions(+), 1 deletion(-) diff --git a/packages/runtime-host/src/__tests__/session-continuity-coordinator.test.ts b/packages/runtime-host/src/__tests__/session-continuity-coordinator.test.ts index 536e831d86..6168f731ce 100644 --- a/packages/runtime-host/src/__tests__/session-continuity-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/session-continuity-coordinator.test.ts @@ -11,6 +11,10 @@ import { SESSION_TRANSCRIPT_PAGE_MAX_BYTES, type SubscriptionFrame, } from '../protocol/index.js'; +import { + decodeSubscriptionFrame, + SESSION_LIVE_DELTA_MAX_BYTES, +} from '../protocol/session-continuity.js'; import type { ConnectionContext } from '../server/operation-dispatcher.js'; import { type CanonicalSessionProjection, @@ -763,6 +767,49 @@ test('keeps stream, kind, and completion boundaries when coalescing deltas', asy coordinator.close(); }); +test('keeps coalesced deltas within the protocol text and frame limits', async () => { + const coordinator = new SessionContinuityCoordinator( + HOST_EPOCH, + async () => canonical(), + new SessionAdmissionGate(), + ); + const sink = new RecordingSink(); + const connection = coordinator.attachConnection('connection-1', sink); + const opened = await open(coordinator, 'connection-1'); + + // Each delta is individually protocol-valid, but merging the two would + // push the text past SESSION_LIVE_DELTA_MAX_BYTES: the second must stay + // its own frame so the receiver-side decoder does not reject it. + const firstText = 'a'.repeat(SESSION_LIVE_DELTA_MAX_BYTES - 1024); + const secondText = 'b'.repeat(SESSION_LIVE_DELTA_MAX_BYTES - 1024); + await coordinator.acceptRuntimeEvent(SESSION_ID, 'run-1', { ...textEvent(1), text: firstText }); + await coordinator.acceptRuntimeEvent(SESSION_ID, 'run-1', { ...textEvent(2), text: secondText }); + // A small continuation still merges into the new tail. + await coordinator.acceptRuntimeEvent(SESSION_ID, 'run-1', textEvent(3)); + + connection.activate(opened.subscriptionId); + await waitFor(() => sink.frames.length === 2); + + const first = sink.frames[0]; + assert.equal(first?.kind, 'subscription.session_delta'); + if (first?.kind !== 'subscription.session_delta') return; + assert.equal(first.sequence, 1); + assert.equal(first.delta.startOffset, 0); + assert.equal(first.delta.text, firstText); + + const second = sink.frames[1]; + assert.equal(second?.kind, 'subscription.session_delta'); + if (second?.kind !== 'subscription.session_delta') return; + assert.equal(second.sequence, 2); + assert.equal(second.delta.startOffset, firstText.length); + assert.equal(second.delta.text, secondText + 'chunk-3'); + + // Every emitted frame passes the receiver-side decoder, including its + // 16 KiB delta-text and 64 KiB frame limits. + for (const frame of sink.frames) decodeSubscriptionFrame(frame); + coordinator.close(); +}); + test('removal closes every Session subscriber at the admitted sequence boundary', async () => { const admission = new SessionAdmissionGate(); const coordinator = new SessionContinuityCoordinator( diff --git a/packages/runtime-host/src/server/session-continuity-coordinator.ts b/packages/runtime-host/src/server/session-continuity-coordinator.ts index 77a663374c..02d50c741b 100644 --- a/packages/runtime-host/src/server/session-continuity-coordinator.ts +++ b/packages/runtime-host/src/server/session-continuity-coordinator.ts @@ -1368,9 +1368,17 @@ export class SessionContinuityCoordinator implements SessionContinuityService { delta: { ...tail.frame.delta, text: mergedText }, }; const mergedEncodedBytes = encodeProtocolMessage(merged).byteLength; + // Merging must preserve the wire invariants the split path + // guarantees per frame: the decoder rejects a delta text beyond + // SESSION_LIVE_DELTA_MAX_BYTES and any subscription frame beyond + // SESSION_SUBSCRIPTION_FRAME_MAX_BYTES, so an oversized merge would + // break the very subscription coalescing tries to preserve. Keep + // the next delta as its own frame instead. if ( + Buffer.byteLength(mergedText, 'utf8') <= SESSION_LIVE_DELTA_MAX_BYTES && + mergedEncodedBytes <= SESSION_SUBSCRIPTION_FRAME_MAX_BYTES && subscriber.queuedBytes - tail.encodedBytes + mergedEncodedBytes + terminalBytes <= - MAX_SUBSCRIBER_QUEUED_BYTES + MAX_SUBSCRIBER_QUEUED_BYTES ) { tail.frame = merged; subscriber.queuedBytes += mergedEncodedBytes - tail.encodedBytes; From 5224d45a800aa55c2ed59343df09d310eb78ba78 Mon Sep 17 00:00:00 2001 From: me2seeks Date: Tue, 18 Aug 2026 19:19:16 +0800 Subject: [PATCH 12/17] fix(cli): retain tool results under lag Generated-by: Maka --- .../runtime-host-session-driver.test.ts | 27 +++++++++++++++++ .../cli/src/runtime-host-session-channel.ts | 30 ++++++++++++------- 2 files changed, 47 insertions(+), 10 deletions(-) diff --git a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts index a9f6565176..fa1d8fb45c 100644 --- a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts +++ b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts @@ -1807,6 +1807,33 @@ describe('turn consumer lag recovery (#3180)', () => { ); }); + test('admits a tool result when the lagged backlog holds no deltas', async () => { + const { initial, replacement, connection, driver, resynced } = lagRecoveryFixture(); + const switched = await driver.switchSession('session-1'); + assert.ok(switched.activeTurn); + + // Fill the bound with non-delta events: nothing sheddable to evict. + await floodToolStream(initial, 1_100); + await waitForSubscriptions(connection, 2); + await resynced.promise; + + // The tool result is the authoritative terminal outcome for its tool and + // must land even though no delta can be evicted; otherwise the live tool + // card stays running until the durable transcript heals it. + replacement.push(toolResultFrame(1, 'subscription-2')); + replacement.push(projectionFrame(2, completedTurn('turn-1', 'run-1'), 2, 'subscription-2')); + await delay(0); + + let sawToolResult = false; + const iterator = switched.activeTurn.events[Symbol.asyncIterator](); + for (let index = 0; index < 1_200; index += 1) { + const result = await iterator.next(); + if (result.done) break; + if ((result.value as { type?: string }).type === 'tool_result') sawToolResult = true; + } + assert.ok(sawToolResult, 'tool_result was admitted over a non-delta backlog'); + }); + test('sheds lagged tool output deltas so the tool result and terminal outcome land', async () => { const { initial, replacement, connection, driver, resynced } = lagRecoveryFixture(); const switched = await driver.switchSession('session-1'); diff --git a/packages/cli/src/runtime-host-session-channel.ts b/packages/cli/src/runtime-host-session-channel.ts index ed1443edc7..950559fd05 100644 --- a/packages/cli/src/runtime-host-session-channel.ts +++ b/packages/cli/src/runtime-host-session-channel.ts @@ -626,16 +626,18 @@ class SessionEventQueue implements AsyncIterable, AsyncIterator Date: Tue, 18 Aug 2026 21:51:46 +0800 Subject: [PATCH 13/17] fix(cli): make session recovery a bounded sequence cut Generated-by: Maka --- .../runtime-host-session-driver.test.ts | 89 +++++++++++++++++-- .../cli/src/runtime-host-session-channel.ts | 79 ++++++++++++---- 2 files changed, 143 insertions(+), 25 deletions(-) diff --git a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts index fa1d8fb45c..5935f489d1 100644 --- a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts +++ b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts @@ -1807,6 +1807,26 @@ describe('turn consumer lag recovery (#3180)', () => { ); }); + test('drops the entire pre-resync tool backlog at the canonical cut', async () => { + const { initial, replacement, connection, driver, resynced } = lagRecoveryFixture(); + const switched = await driver.switchSession('session-1'); + assert.ok(switched.activeTurn); + + await floodToolStream(initial, 1_100); + await waitForSubscriptions(connection, 2); + await resynced.promise; + + replacement.push(toolStartFrame(1, 9_000, 'subscription-2')); + replacement.push(projectionFrame(2, completedTurn('turn-1', 'run-1'), 2, 'subscription-2')); + await delay(0); + + const iterator = switched.activeTurn.events[Symbol.asyncIterator](); + const first = await iterator.next(); + assert.equal(first.done, false); + assert.equal(first.value.type, 'tool_start'); + if (first.value.type === 'tool_start') assert.equal(first.value.toolUseId, 'tool-9000'); + }); + test('admits a tool result when the lagged backlog holds no deltas', async () => { const { initial, replacement, connection, driver, resynced } = lagRecoveryFixture(); const switched = await driver.switchSession('session-1'); @@ -1904,6 +1924,51 @@ describe('turn consumer lag recovery (#3180)', () => { assert.equal((await nextEvent(switched.activeTurn.events)).text, '!'); }); + test('backs off several immediate clean-EOF replacements before recovering', async () => { + const initial = new FakeSubscription( + continuitySnapshot(), + Promise.resolve([assistantMessage('turn-1', 'Hello')]), + ); + const ended = [2, 3, 4].map( + (index) => + new FakeSubscription( + continuitySnapshot({ projectionRevision: index }), + Promise.resolve([assistantMessage('turn-1', 'Hello')]), + `subscription-${index}`, + ), + ); + for (const subscription of ended) await subscription.close(); + const stable = new FakeSubscription( + continuitySnapshot({ projectionRevision: 5 }), + Promise.resolve([assistantMessage('turn-1', 'Hello world')]), + 'subscription-5', + ); + const connection = new FakeConnection([initial, ...ended, stable], true); + const driver = createRuntimeHostMakaSessionDriver({ + connection: connection.value, + cwd: '/tmp', + llmConnectionSlug: 'openai-main', + model: 'gpt-5', + now: () => 50, + }); + const switched = await driver.switchSession('session-1'); + assert.ok(switched.activeTurn); + const resynced = deferred(); + driver.subscribeTranscriptReplacements!((_sessionId, _turnId, _messages, reason) => { + if (reason === 'reconnect') resynced.resolve(); + }); + + await initial.close(); + await waitForSubscriptions(connection, 2); + await delay(5); + assert.equal(connection.openedSubscriptions, 2, 'the first repeated EOF is backoff-gated'); + + await resynced.promise; + assert.equal(connection.openedSubscriptions, 5); + stable.push(deltaFrame(1, 'turn-1', 11, '!', 'subscription-5')); + assert.equal((await nextEvent(switched.activeTurn.events)).text, '!'); + }); + test('re-arms lag detection exactly at the hysteresis watermark', async () => { const initial = new FakeSubscription( continuitySnapshot(), @@ -1934,11 +1999,15 @@ describe('turn consumer lag recovery (#3180)', () => { const switched = await driver.switchSession('session-1'); assert.ok(switched.activeTurn); - // Latch the lag flag with a full non-delta backlog: all 1_024 queued - // events stay because nothing is sheddable. + // Latch the lag flag with a non-delta backlog. The canonical cut clears + // every pre-cut event, then a still-wedged consumer fills again without + // triggering a resubscribe loop. await floodToolStream(initial, 1_100); await waitForSubscriptions(connection, 2); await waitFor(() => resyncs === 1); + await floodToolStream(second, 1_100, 'subscription-2', 1); + await delay(20); + assert.equal(connection.openedSubscriptions, 2, 'the post-cut lag latch stayed armed'); // Draining to one event above the watermark (513 pending) must NOT // re-arm: a fresh overflow on the still-latched queue is the same lag @@ -1948,7 +2017,7 @@ describe('turn consumer lag recovery (#3180)', () => { for (let index = 0; index < 511; index += 1) { assert.equal((await iterator.next()).done, false); } - await floodToolStream(second, 600, 'subscription-2', 2_000); + await floodToolStream(second, 600, 'subscription-2', 1_101); await delay(20); assert.equal(connection.openedSubscriptions, 2, 'lag latch held above the watermark'); @@ -1957,7 +2026,7 @@ describe('turn consumer lag recovery (#3180)', () => { for (let index = 0; index < 512; index += 1) { assert.equal((await iterator.next()).done, false); } - await floodToolStream(second, 600, 'subscription-2', 3_000); + await floodToolStream(second, 600, 'subscription-2', 1_701); await waitForSubscriptions(connection, 3); await waitFor(() => resyncs === 2); }); @@ -1992,11 +2061,13 @@ describe('turn consumer lag recovery (#3180)', () => { const switched = await driver.switchSession('session-1'); assert.ok(switched.activeTurn); - // First lag episode over a non-delta backlog: nothing to compact, and the - // latch stays on while the consumer remains behind. + // First lag episode over a non-delta backlog. The canonical cut clears the + // retired subscription's events; a still-wedged consumer can fill again + // without immediately looping recovery. await floodToolStream(initial, 1_100); await waitForSubscriptions(connection, 2); await waitFor(() => resyncs === 1); + await floodToolStream(second, 1_100, 'subscription-2', 1); // The consumer drains past the hysteresis watermark, re-arming lag // detection, and fresh output flows again. One hundred events stay queued @@ -2006,9 +2077,9 @@ describe('turn consumer lag recovery (#3180)', () => { const result = await iterator.next(); assert.equal(result.done, false); } - second.push(deltaFrame(1, 'turn-1', 5, ' world', 'subscription-2')); + second.push(deltaFrame(1_101, 'turn-1', 5, ' world', 'subscription-2')); for (let index = 0; index < 100; index += 1) { - second.push(toolStartFrame(2 + index, 2_000 + index, 'subscription-2')); + second.push(toolStartFrame(1_102 + index, 2_000 + index, 'subscription-2')); } await delay(0); let fresh = ''; @@ -2021,7 +2092,7 @@ describe('turn consumer lag recovery (#3180)', () => { // A second lag episode is a new episode, not a dead latch: it triggers a // fresh canonical resync. The stream stays contiguous on `second`. - await floodToolStream(second, 1_100, 'subscription-2', 102); + await floodToolStream(second, 1_100, 'subscription-2', 1_202); await waitForSubscriptions(connection, 3); await waitFor(() => resyncs === 2); diff --git a/packages/cli/src/runtime-host-session-channel.ts b/packages/cli/src/runtime-host-session-channel.ts index 950559fd05..8f33d29c1e 100644 --- a/packages/cli/src/runtime-host-session-channel.ts +++ b/packages/cli/src/runtime-host-session-channel.ts @@ -25,6 +25,10 @@ import type { MakaPreparedSessionTurn } from './session-driver.js'; const MAX_PENDING_FRAMES = 512; const MAX_PENDING_EVENTS_PER_TURN = 1_024; const LAG_REARM_PENDING_EVENTS = MAX_PENDING_EVENTS_PER_TURN / 2; +const MAX_RECOVERY_ATTEMPTS_WITHOUT_LIVE_FRAME = 8; +const RECOVERY_BACKOFF_INITIAL_MS = 25; +const RECOVERY_BACKOFF_MAX_MS = 500; +const RECOVERY_STABLE_AFTER_MS = 1_000; export interface RuntimeHostSessionChannelOpenResult { channel: RuntimeHostSessionChannel; @@ -75,6 +79,8 @@ export class RuntimeHostSessionChannel { #closing = false; #failure: Error | undefined; #recoveryTask: Promise | undefined; + #recoveryAttemptsWithoutLiveFrame = 0; + #recoveryStableTimer: ReturnType | undefined; private constructor( subscription: RuntimeHostSessionSubscription, @@ -244,6 +250,7 @@ export class RuntimeHostSessionChannel { async close(): Promise { if (this.#closing) return; this.#closing = true; + this.#clearRecoveryStableTimer(); this.#pendingStartedTurns.clear(); for (const queue of this.#turns.values()) queue.finish(); await this.#subscription.close(); @@ -253,6 +260,7 @@ export class RuntimeHostSessionChannel { try { for await (const frame of subscription) { if (this.#closing || this.#subscription !== subscription) return; + if (frame.kind !== 'subscription.closed') this.#markRecoveryStable(subscription); if (!this.#ready) { if (this.#pendingFrames.length >= MAX_PENDING_FRAMES) { throw new RuntimeHostSubscriptionError( @@ -306,6 +314,7 @@ export class RuntimeHostSessionChannel { #scheduleRecovery(failed: RuntimeHostSessionSubscription): void { if (this.#closing || this.#failure || this.#subscription !== failed || this.#recoveryTask) return; + this.#clearRecoveryStableTimer(); const task = this.#recover(failed); this.#recoveryTask = task; void task @@ -321,6 +330,8 @@ export class RuntimeHostSessionChannel { let previous = failed; while (!this.#closing && !this.#failure && this.#subscription === previous) { await previous.close().catch(() => undefined); + await this.#waitForRecoveryAttempt(); + if (this.#closing || this.#failure || this.#subscription !== previous) return; let replacement: RuntimeHostSessionSubscription; try { replacement = await this.#connection.openSessionSubscription({ @@ -355,6 +366,7 @@ export class RuntimeHostSessionChannel { this.#ready = true; for (const frame of this.#pendingFrames.splice(0)) this.#accept(frame); if (replacedLiveState) this.#onRecovered(); + this.#scheduleRecoveryStable(replacement); return; } catch (error) { if (!this.#canRecover(error)) throw error; @@ -379,10 +391,11 @@ export class RuntimeHostSessionChannel { this.#now, this.#subscription.activeAssistantStreams, ); - // Deltas a lagging consumer has not seen are superseded by this canonical - // resync; keeping them would shed the fresh post-recovery stream behind - // them. - for (const queue of this.#turns.values()) queue.shedLaggedDeltas(); + // A canonical replacement is a sequence cut. No queued event from the + // retired subscription may replay after the transcript/snapshot has + // established newer state; active, terminal, and interaction state is + // seeded again below from the replacement authority. + for (const queue of this.#turns.values()) queue.cutBacklog(); if (!replacedLiveState) { for (const event of this.#projector.seedActive(false)) this.#emit(event); return false; @@ -453,6 +466,46 @@ export class RuntimeHostSessionChannel { return true; } + async #waitForRecoveryAttempt(): Promise { + if (this.#recoveryAttemptsWithoutLiveFrame >= MAX_RECOVERY_ATTEMPTS_WITHOUT_LIVE_FRAME) { + throw new RuntimeHostSubscriptionError( + 'connection_closed', + 'Runtime Host Session subscription recovery exhausted its retry budget', + ); + } + if (this.#recoveryAttemptsWithoutLiveFrame > 0) { + const delayMs = Math.min( + RECOVERY_BACKOFF_INITIAL_MS * 2 ** (this.#recoveryAttemptsWithoutLiveFrame - 1), + RECOVERY_BACKOFF_MAX_MS, + ); + await new Promise((resolve) => setTimeout(resolve, delayMs)); + } + this.#recoveryAttemptsWithoutLiveFrame += 1; + } + + #markRecoveryStable(subscription: RuntimeHostSessionSubscription): void { + if (this.#subscription !== subscription) return; + this.#clearRecoveryStableTimer(); + this.#recoveryAttemptsWithoutLiveFrame = 0; + } + + #scheduleRecoveryStable(subscription: RuntimeHostSessionSubscription): void { + this.#clearRecoveryStableTimer(); + const timer = setTimeout(() => { + if (!this.#closing && this.#subscription === subscription && this.#ready) { + this.#recoveryAttemptsWithoutLiveFrame = 0; + } + if (this.#recoveryStableTimer === timer) this.#recoveryStableTimer = undefined; + }, RECOVERY_STABLE_AFTER_MS); + timer.unref?.(); + this.#recoveryStableTimer = timer; + } + + #clearRecoveryStableTimer(): void { + if (this.#recoveryStableTimer !== undefined) clearTimeout(this.#recoveryStableTimer); + this.#recoveryStableTimer = undefined; + } + #canRecover(error: unknown): boolean { if (!isRuntimeHostReconnectingConnection(this.#connection)) return false; if (error instanceof RuntimeHostRequestInterruptedError) { @@ -598,6 +651,9 @@ class SessionEventQueue implements AsyncIterable, AsyncIterator { this.#waiting = { resolve, reject }; }); @@ -646,18 +702,9 @@ class SessionEventQueue implements AsyncIterable, AsyncIterator= 0; index -= 1) { - if (isSheddableDelta(this.#items[index]!)) this.#items.splice(index, 1); - } + /** Drop the entire unseen pre-cut backlog after canonical replacement. */ + cutBacklog(): void { + this.#items.length = 0; } #noteLag(): void { From 13c0e1b7927c53adafcb718debfa48c5313a9d74 Mon Sep 17 00:00:00 2001 From: me2seeks Date: Wed, 19 Aug 2026 00:41:43 +0800 Subject: [PATCH 14/17] fix(cli): preserve bounded recovery outcomes --- .../runtime-host-session-driver.test.ts | 74 +++++++++++++++++++ .../cli/src/runtime-host-session-channel.ts | 22 ++---- 2 files changed, 82 insertions(+), 14 deletions(-) diff --git a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts index 5935f489d1..d2c728351c 100644 --- a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts +++ b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts @@ -1969,6 +1969,80 @@ describe('turn consumer lag recovery (#3180)', () => { assert.equal((await nextEvent(switched.activeTurn.events)).text, '!'); }); + for (const [name, replacementRoot] of [ + ['the same terminal turn', completedTurn('turn-1', 'run-1')], + ['a successor turn', runningTurn('turn-2', 'run-2')], + ] as const) { + test(`preserves an unconsumed terminal event across a replacement with ${name}`, async () => { + const initial = new FakeSubscription( + continuitySnapshot(), + Promise.resolve([assistantMessage('turn-1', 'Hello')]), + ); + const replacement = new FakeSubscription( + continuitySnapshot({ projectionRevision: 3, rootTurn: replacementRoot }), + Promise.resolve([ + assistantMessage('turn-1', 'Hello'), + turnStateMessage('turn-1', 'completed'), + ...(replacementRoot.turnId === 'turn-2' ? [userMessage('turn-2', 'Continue')] : []), + ]), + 'subscription-2', + ); + const connection = new FakeConnection([initial, replacement], true); + const driver = createRuntimeHostMakaSessionDriver({ + connection: connection.value, + cwd: '/tmp', + llmConnectionSlug: 'openai-main', + model: 'gpt-5', + now: () => 50, + }); + const switched = await driver.switchSession('session-1'); + assert.ok(switched.activeTurn); + + initial.push(projectionFrame(1, completedTurn('turn-1', 'run-1'), 2)); + await delay(0); + initial.fail(new RuntimeHostSubscriptionError('connection_closed', 'connection lost')); + await waitForSubscriptions(connection, 2); + + assert.equal((await nextEvent(switched.activeTurn.events)).type, 'complete'); + assert.equal((await switched.activeTurn.events[Symbol.asyncIterator]().next()).done, true); + }); + } + + test('exhausts recovery after repeated one-frame clean-EOF replacements', async () => { + const initial = new FakeSubscription( + continuitySnapshot(), + Promise.resolve([assistantMessage('turn-1', 'Hello')]), + ); + const ended = Array.from({ length: 8 }, (_, index) => { + const subscription = new FakeSubscription( + continuitySnapshot({ projectionRevision: index + 2 }), + Promise.resolve([assistantMessage('turn-1', 'Hello')]), + `subscription-${index + 2}`, + ); + subscription.push(deltaFrame(1, 'turn-1', 5, String(index), `subscription-${index + 2}`)); + return subscription; + }); + for (const subscription of ended) await subscription.close(); + const connection = new FakeConnection([initial, ...ended], true); + const driver = createRuntimeHostMakaSessionDriver({ + connection: connection.value, + cwd: '/tmp', + llmConnectionSlug: 'openai-main', + model: 'gpt-5', + now: () => 50, + }); + const switched = await driver.switchSession('session-1'); + assert.ok(switched.activeTurn); + + await initial.close(); + await assert.rejects(async () => { + for await (const _event of switched.activeTurn!.events) { + // Drain each replacement's single live frame until recovery fails. + } + }, /recovery exhausted its retry budget/u); + assert.equal(connection.openedSubscriptions, 9); + }); + test('re-arms lag detection exactly at the hysteresis watermark', async () => { const initial = new FakeSubscription( continuitySnapshot(), diff --git a/packages/cli/src/runtime-host-session-channel.ts b/packages/cli/src/runtime-host-session-channel.ts index 8f33d29c1e..67d2244ed3 100644 --- a/packages/cli/src/runtime-host-session-channel.ts +++ b/packages/cli/src/runtime-host-session-channel.ts @@ -260,7 +260,6 @@ export class RuntimeHostSessionChannel { try { for await (const frame of subscription) { if (this.#closing || this.#subscription !== subscription) return; - if (frame.kind !== 'subscription.closed') this.#markRecoveryStable(subscription); if (!this.#ready) { if (this.#pendingFrames.length >= MAX_PENDING_FRAMES) { throw new RuntimeHostSubscriptionError( @@ -483,12 +482,6 @@ export class RuntimeHostSessionChannel { this.#recoveryAttemptsWithoutLiveFrame += 1; } - #markRecoveryStable(subscription: RuntimeHostSessionSubscription): void { - if (this.#subscription !== subscription) return; - this.#clearRecoveryStableTimer(); - this.#recoveryAttemptsWithoutLiveFrame = 0; - } - #scheduleRecoveryStable(subscription: RuntimeHostSessionSubscription): void { this.#clearRecoveryStableTimer(); const timer = setTimeout(() => { @@ -702,9 +695,11 @@ class SessionEventQueue implements AsyncIterable, AsyncIterator Date: Wed, 19 Aug 2026 08:56:34 +0800 Subject: [PATCH 15/17] test(cli): exercise post-resync backlog admission --- .../runtime-host-session-driver.test.ts | 27 ++++++++++++++----- 1 file changed, 20 insertions(+), 7 deletions(-) diff --git a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts index d2c728351c..15bbc6eebf 100644 --- a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts +++ b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts @@ -1669,11 +1669,12 @@ describe('turn consumer lag recovery (#3180)', () => { subscription: InstanceType, count: number, startOffset: number, + subscriptionId = 'subscription-1', ): Promise { let offset = startOffset; for (let index = 0; index < count; index += 1) { const text = `x${String(index).padStart(4, '0')}`; - subscription.push(deltaFrame(index + 1, 'turn-1', offset, text)); + subscription.push(deltaFrame(index + 1, 'turn-1', offset, text, subscriptionId)); offset += text.length; if (index % 64 === 63) await delay(0); } @@ -1779,7 +1780,10 @@ describe('turn consumer lag recovery (#3180)', () => { await waitForSubscriptions(connection, 2); await resynced.promise; - replacement.push(projectionFrame(1, completedTurn('turn-1', 'run-1'), 2, 'subscription-2')); + await floodTurnStream(replacement, 1_024, 5, 'subscription-2'); + replacement.push( + projectionFrame(1_025, completedTurn('turn-1', 'run-1'), 2, 'subscription-2'), + ); await delay(0); assert.ok( await drainUntilDone(switched.activeTurn.events), @@ -1797,9 +1801,12 @@ describe('turn consumer lag recovery (#3180)', () => { await waitForSubscriptions(connection, 2); await resynced.promise; + await floodToolStream(replacement, 1_024, 'subscription-2'); // The terminal outcome must land even though no delta can be evicted; // process the frame before draining so the backlog is still full. - replacement.push(projectionFrame(1, completedTurn('turn-1', 'run-1'), 2, 'subscription-2')); + replacement.push( + projectionFrame(1_025, completedTurn('turn-1', 'run-1'), 2, 'subscription-2'), + ); await delay(0); assert.ok( await drainUntilDone(switched.activeTurn.events), @@ -1837,11 +1844,14 @@ describe('turn consumer lag recovery (#3180)', () => { await waitForSubscriptions(connection, 2); await resynced.promise; + await floodToolStream(replacement, 1_024, 'subscription-2'); // The tool result is the authoritative terminal outcome for its tool and // must land even though no delta can be evicted; otherwise the live tool // card stays running until the durable transcript heals it. - replacement.push(toolResultFrame(1, 'subscription-2')); - replacement.push(projectionFrame(2, completedTurn('turn-1', 'run-1'), 2, 'subscription-2')); + replacement.push(toolResultFrame(1_025, 'subscription-2')); + replacement.push( + projectionFrame(1_026, completedTurn('turn-1', 'run-1'), 2, 'subscription-2'), + ); await delay(0); let sawToolResult = false; @@ -1865,12 +1875,15 @@ describe('turn consumer lag recovery (#3180)', () => { await waitForSubscriptions(connection, 2); await resynced.promise; + await floodToolOutput(replacement, 1_024, 'subscription-2'); // The canonical resync compacts the unseen tool deltas, so the tool // result lands instead of being dropped behind a full non-delta backlog // (which would leave the live card stuck at "running" until the durable // transcript heals it). - replacement.push(toolResultFrame(1, 'subscription-2')); - replacement.push(projectionFrame(2, completedTurn('turn-1', 'run-1'), 2, 'subscription-2')); + replacement.push(toolResultFrame(1_025, 'subscription-2')); + replacement.push( + projectionFrame(1_026, completedTurn('turn-1', 'run-1'), 2, 'subscription-2'), + ); await delay(0); let sawToolResult = false; From adb4ae32b2a909582e102e447df202fc3125be44 Mon Sep 17 00:00:00 2001 From: me2seeks Date: Wed, 19 Aug 2026 09:16:52 +0800 Subject: [PATCH 16/17] style(cli): format lag admission coverage --- .../runtime-host-session-driver.test.ts | 16 ++++------------ 1 file changed, 4 insertions(+), 12 deletions(-) diff --git a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts index 15bbc6eebf..b7691d0dbe 100644 --- a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts +++ b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts @@ -1781,9 +1781,7 @@ describe('turn consumer lag recovery (#3180)', () => { await resynced.promise; await floodTurnStream(replacement, 1_024, 5, 'subscription-2'); - replacement.push( - projectionFrame(1_025, completedTurn('turn-1', 'run-1'), 2, 'subscription-2'), - ); + replacement.push(projectionFrame(1_025, completedTurn('turn-1', 'run-1'), 2, 'subscription-2')); await delay(0); assert.ok( await drainUntilDone(switched.activeTurn.events), @@ -1804,9 +1802,7 @@ describe('turn consumer lag recovery (#3180)', () => { await floodToolStream(replacement, 1_024, 'subscription-2'); // The terminal outcome must land even though no delta can be evicted; // process the frame before draining so the backlog is still full. - replacement.push( - projectionFrame(1_025, completedTurn('turn-1', 'run-1'), 2, 'subscription-2'), - ); + replacement.push(projectionFrame(1_025, completedTurn('turn-1', 'run-1'), 2, 'subscription-2')); await delay(0); assert.ok( await drainUntilDone(switched.activeTurn.events), @@ -1849,9 +1845,7 @@ describe('turn consumer lag recovery (#3180)', () => { // must land even though no delta can be evicted; otherwise the live tool // card stays running until the durable transcript heals it. replacement.push(toolResultFrame(1_025, 'subscription-2')); - replacement.push( - projectionFrame(1_026, completedTurn('turn-1', 'run-1'), 2, 'subscription-2'), - ); + replacement.push(projectionFrame(1_026, completedTurn('turn-1', 'run-1'), 2, 'subscription-2')); await delay(0); let sawToolResult = false; @@ -1881,9 +1875,7 @@ describe('turn consumer lag recovery (#3180)', () => { // (which would leave the live card stuck at "running" until the durable // transcript heals it). replacement.push(toolResultFrame(1_025, 'subscription-2')); - replacement.push( - projectionFrame(1_026, completedTurn('turn-1', 'run-1'), 2, 'subscription-2'), - ); + replacement.push(projectionFrame(1_026, completedTurn('turn-1', 'run-1'), 2, 'subscription-2')); await delay(0); let sawToolResult = false; From 930fdeaf326e82aa459a38934c2c9718fdbe2202 Mon Sep 17 00:00:00 2001 From: me2seeks Date: Wed, 19 Aug 2026 15:35:49 +0800 Subject: [PATCH 17/17] fix(cli): close remaining lag recovery boundaries Start recovery stability only after a post-hydration live frame, recover buffered slow-consumer closure during initial hydration, and guarantee assistant completion admission over saturated control backlogs. Generated-by: Maka --- .../runtime-host-session-driver.test.ts | 128 ++++++++++++++++++ .../cli/src/runtime-host-session-channel.ts | 40 +++++- 2 files changed, 162 insertions(+), 6 deletions(-) diff --git a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts index b7691d0dbe..ec07c74f5a 100644 --- a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts +++ b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts @@ -1543,6 +1543,31 @@ function deltaFrame( }; } +function textCompleteFrame( + sequence: number, + turnId: string, + startOffset: number, + text: string, + subscriptionId = 'subscription-1', +): SubscriptionFrame { + return { + kind: 'subscription.session_delta', + hostEpoch: 'host-1', + subscriptionId, + sequence, + sessionId: 'session-1', + delta: { + kind: 'text', + turnId, + runId: 'run-1', + messageId: `message-${turnId}`, + startOffset, + text, + complete: true, + }, + }; +} + function thinkingFrame( sequence: number, messageId: string, @@ -1810,6 +1835,30 @@ describe('turn consumer lag recovery (#3180)', () => { ); }); + test('admits assistant completion before the terminal outcome over a non-delta backlog', async () => { + const { initial, replacement, connection, driver, resynced } = lagRecoveryFixture(); + const switched = await driver.switchSession('session-1'); + assert.ok(switched.activeTurn); + + await floodToolStream(initial, 1_100); + await waitForSubscriptions(connection, 2); + await resynced.promise; + + await floodToolStream(replacement, 1_024, 'subscription-2'); + replacement.push(textCompleteFrame(1_025, 'turn-1', 5, ' final answer', 'subscription-2')); + replacement.push(projectionFrame(1_026, completedTurn('turn-1', 'run-1'), 2, 'subscription-2')); + await delay(0); + + let finalOutput: string | undefined; + let completed = false; + for await (const event of switched.activeTurn.events) { + if (event.type === 'text_complete') finalOutput = event.text; + if (event.type === 'complete') completed = true; + } + assert.equal(finalOutput, 'Hello final answer'); + assert.equal(completed, true); + }); + test('drops the entire pre-resync tool backlog at the canonical cut', async () => { const { initial, replacement, connection, driver, resynced } = lagRecoveryFixture(); const switched = await driver.switchSession('session-1'); @@ -1929,6 +1978,42 @@ describe('turn consumer lag recovery (#3180)', () => { assert.equal((await nextEvent(switched.activeTurn.events)).text, '!'); }); + test('recovers when slow-consumer closure is buffered during initial hydration', async () => { + const transcript = deferred(); + const initial = new FakeSubscription(continuitySnapshot(), transcript.promise); + const replacement = new FakeSubscription( + continuitySnapshot({ projectionRevision: 2 }), + Promise.resolve([assistantMessage('turn-1', 'Hello world')]), + 'subscription-2', + ); + const connection = new FakeConnection([initial, replacement], true); + const driver = createRuntimeHostMakaSessionDriver({ + connection: connection.value, + cwd: '/tmp', + llmConnectionSlug: 'openai-main', + model: 'gpt-5', + now: () => 50, + }); + + const switching = driver.switchSession('session-1'); + await waitFor(() => initial.nextCalls === 1); + initial.push({ + kind: 'subscription.closed', + hostEpoch: 'host-1', + subscriptionId: 'subscription-1', + sequence: 1, + reason: 'slow_consumer', + }); + await waitFor(() => initial.nextCalls === 2); + transcript.resolve([assistantMessage('turn-1', 'Hello')]); + + const switched = await switching; + assert.ok(switched.activeTurn); + assert.equal(connection.openedSubscriptions, 2); + replacement.push(deltaFrame(1, 'turn-1', 11, '!', 'subscription-2')); + assert.equal((await nextEvent(switched.activeTurn.events)).text, '!'); + }); + test('backs off several immediate clean-EOF replacements before recovering', async () => { const initial = new FakeSubscription( continuitySnapshot(), @@ -2048,6 +2133,49 @@ describe('turn consumer lag recovery (#3180)', () => { assert.equal(connection.openedSubscriptions, 9); }); + test('does not reset recovery after a silent replacement outlives the stability window', async () => { + const initial = new FakeSubscription( + continuitySnapshot(), + Promise.resolve([assistantMessage('turn-1', 'Hello')]), + ); + const silent = new FakeSubscription( + continuitySnapshot({ projectionRevision: 2 }), + Promise.resolve([assistantMessage('turn-1', 'Hello')]), + 'subscription-2', + ); + const ended = Array.from({ length: 7 }, (_, index) => { + const subscription = new FakeSubscription( + continuitySnapshot({ projectionRevision: index + 3 }), + Promise.resolve([assistantMessage('turn-1', 'Hello')]), + `subscription-${index + 3}`, + ); + void subscription.close(); + return subscription; + }); + const connection = new FakeConnection([initial, silent, ...ended], true); + const driver = createRuntimeHostMakaSessionDriver({ + connection: connection.value, + cwd: '/tmp', + llmConnectionSlug: 'openai-main', + model: 'gpt-5', + now: () => 50, + }); + const switched = await driver.switchSession('session-1'); + assert.ok(switched.activeTurn); + + await initial.close(); + await waitForSubscriptions(connection, 2); + await delay(1_100); + await silent.close(); + + await assert.rejects(async () => { + for await (const _event of switched.activeTurn!.events) { + // A silent hydrated subscription is not evidence of live stability. + } + }, /recovery exhausted its retry budget/u); + assert.equal(connection.openedSubscriptions, 9); + }); + test('re-arms lag detection exactly at the hysteresis watermark', async () => { const initial = new FakeSubscription( continuitySnapshot(), diff --git a/packages/cli/src/runtime-host-session-channel.ts b/packages/cli/src/runtime-host-session-channel.ts index 67d2244ed3..5acf02b167 100644 --- a/packages/cli/src/runtime-host-session-channel.ts +++ b/packages/cli/src/runtime-host-session-channel.ts @@ -80,6 +80,7 @@ export class RuntimeHostSessionChannel { #failure: Error | undefined; #recoveryTask: Promise | undefined; #recoveryAttemptsWithoutLiveFrame = 0; + #recoveryAwaitingLiveFrame: RuntimeHostSessionSubscription | undefined; #recoveryStableTimer: ReturnType | undefined; private constructor( @@ -155,7 +156,17 @@ export class RuntimeHostSessionChannel { ); for (const event of this.#projector.seedActive(false)) this.#emit(event); this.#ready = true; - for (const frame of this.#pendingFrames.splice(0)) this.#accept(frame); + try { + for (const frame of this.#pendingFrames.splice(0)) this.#accept(frame); + } catch (error) { + if (!this.#canRecover(error)) throw error; + this.#failedSubscriptions.add(subscription); + await this.#recover(subscription); + if (!this.#ready) { + throw this.#failure ?? new Error('Runtime Host Session recovery ended before hydration'); + } + return true; + } return false; } @@ -251,6 +262,7 @@ export class RuntimeHostSessionChannel { if (this.#closing) return; this.#closing = true; this.#clearRecoveryStableTimer(); + this.#recoveryAwaitingLiveFrame = undefined; this.#pendingStartedTurns.clear(); for (const queue of this.#turns.values()) queue.finish(); await this.#subscription.close(); @@ -270,6 +282,7 @@ export class RuntimeHostSessionChannel { this.#pendingFrames.push(frame); } else { this.#accept(frame); + if (frame.kind !== 'subscription.closed') this.#observeRecoveryLiveFrame(subscription); } } // A stream that ends without a subscription.closed frame is a broken @@ -328,6 +341,9 @@ export class RuntimeHostSessionChannel { async #recover(failed: RuntimeHostSessionSubscription): Promise { let previous = failed; while (!this.#closing && !this.#failure && this.#subscription === previous) { + if (this.#recoveryAwaitingLiveFrame === previous) { + this.#recoveryAwaitingLiveFrame = undefined; + } await previous.close().catch(() => undefined); await this.#waitForRecoveryAttempt(); if (this.#closing || this.#failure || this.#subscription !== previous) return; @@ -362,10 +378,10 @@ export class RuntimeHostSessionChannel { } if (this.#closing || this.#failure || this.#subscription !== replacement) return; const replacedLiveState = this.#acceptCanonicalReplacement(messages); + this.#recoveryAwaitingLiveFrame = replacement; this.#ready = true; for (const frame of this.#pendingFrames.splice(0)) this.#accept(frame); if (replacedLiveState) this.#onRecovered(); - this.#scheduleRecoveryStable(replacement); return; } catch (error) { if (!this.#canRecover(error)) throw error; @@ -494,6 +510,12 @@ export class RuntimeHostSessionChannel { this.#recoveryStableTimer = timer; } + #observeRecoveryLiveFrame(subscription: RuntimeHostSessionSubscription): void { + if (this.#recoveryAwaitingLiveFrame !== subscription) return; + this.#recoveryAwaitingLiveFrame = undefined; + this.#scheduleRecoveryStable(subscription); + } + #clearRecoveryStableTimer(): void { if (this.#recoveryStableTimer !== undefined) clearTimeout(this.#recoveryStableTimer); this.#recoveryStableTimer = undefined; @@ -740,10 +762,16 @@ function isSheddableDelta(event: SessionEvent): boolean { } function isGuaranteedOutcome(event: SessionEvent): boolean { - // complete/abort/error close the turn; tool_result is the authoritative - // terminal result for its tool — losing it leaves the live tool card - // running until the durable transcript heals it on a later reload. - return isTurnTerminalOutcome(event) || event.type === 'tool_result'; + // complete/abort/error close the turn; text/thinking completion carries the + // authoritative assistant accumulator; tool_result is the authoritative + // terminal result for its tool. Losing any of them leaves a consumer with + // an incomplete outcome even though the projector has already settled it. + return ( + isTurnTerminalOutcome(event) || + event.type === 'text_complete' || + event.type === 'thinking_complete' || + event.type === 'tool_result' + ); } function isTurnTerminalOutcome(event: SessionEvent): boolean {