diff --git a/apps/desktop/src/main/__tests__/desktop-transcript-range-store.test.ts b/apps/desktop/src/main/__tests__/desktop-transcript-range-store.test.ts index 308b9dce42..0c1428d8ce 100644 --- a/apps/desktop/src/main/__tests__/desktop-transcript-range-store.test.ts +++ b/apps/desktop/src/main/__tests__/desktop-transcript-range-store.test.ts @@ -256,6 +256,305 @@ test('keeps a bounded contiguous window while moving between history and the tai assert.ok(replica.residentBytes <= maxResidentBytes); }); +test('delivers a mid-session tail append even while a history window is resident', async () => { + // Reproduces the "active session does not show the newest message until you + // switch away and back" bug. Once the resident window has been trimmed off + // the tail (hasNewer === true, e.g. after loading older history), a Host + // `transcript_advanced` for a freshly persisted message must still reach an + // already-open consumer. Before the fix, `advance()` short-circuited on + // hasNewer and published an empty change, so the append was silently dropped + // and only a fresh subscription (session switch) re-read it. + const messages = [0, 1, 2, 3, 4].map((sequence) => ({ + identity: sequence, + message: assistantMessage(String(sequence), `assistant-${sequence}`), + })); + const appended = { identity: 5, message: assistantMessage('5', 'assistant-5') }; + const page = (nextCursor: string | null) => ({ + kind: 'page' as const, + sessionId: 'session-1', + source: 'durable' as const, + direction: 'older' as const, + throughSequence: 4, + rawBytes: 1, + fragments: [], + nextCursor, + }); + const bootstrapPage = page('older'); + const olderPage = page('older'); + // The tail reload after the append: a fresh newest window ending at seq 5, + // with older history still available below it. + const tailPage = { ...page('older'), throughSequence: 5 }; + const changes: { durableUpserts: readonly { sequence: number }[] }[] = []; + const handle = runtimeHostSessionFixture({ + snapshot: continuitySnapshot(), + transcript: Promise.resolve([]), + events: { async *[Symbol.asyncIterator]() {} }, + transcriptBootstrap: { + throughSequence: 4, + overlayMessageCount: 0, + durable: bootstrapPage, + overlay: { ...page(null), source: 'overlay' }, + }, + loadTranscriptOverlay: async () => [], + decodeTranscriptPage: async (candidate) => candidate === bootstrapPage + ? { messages: messages.slice(3), nextCursor: 'older' } + : candidate === tailPage + ? { messages: [appended], nextCursor: 'older' } + : { messages: messages.slice(2, 4), nextCursor: 'older' }, + loadTranscriptPage: async (input) => input.throughSequence === 5 ? tailPage : olderPage, + async close() {}, + }); + const maxResidentBytes = Buffer.byteLength(JSON.stringify(messages[0]!.message), 'utf8') + 1; + const replica = await DesktopTranscriptReplica.prepare(handle, { + maxResidentBytes, + onChange: (_replica, change) => changes.push(change), + }); + + await replica.loadBefore(4, 128 * 1024); + assert.equal(replica.snapshot().hasNewer, true); + changes.splice(0); + + // The Host persists a new assistant message (sequence 5) and advances. + await replica.advance(5); + + const upserts = changes.flatMap((change) => change.durableUpserts.map(({ sequence }) => sequence)); + assert.ok(upserts.includes(5), 'the tail append must be delivered to open consumers'); + assert.equal(replica.durableThrough, 5); +}); + +test('does not resurrect a discarded replica when a tail re-anchor is in flight', async () => { + // Guards the concurrency edge introduced by re-anchoring on `hasNewer`: the + // re-anchor now awaits a page load, and `discard()` (memory reclaim for a + // non-visible session) can run during that await. When the page resolves the + // replica must stay non-resident and empty — repopulating durable state here + // would undo the eviction and blow the memory bound. + const messages = [0, 1, 2, 3, 4].map((sequence) => ({ + identity: sequence, + message: assistantMessage(String(sequence), `assistant-${sequence}`), + })); + const appended = { identity: 5, message: assistantMessage('5', 'assistant-5') }; + const page = (nextCursor: string | null) => ({ + kind: 'page' as const, + sessionId: 'session-1', + source: 'durable' as const, + direction: 'older' as const, + throughSequence: 4, + rawBytes: 1, + fragments: [], + nextCursor, + }); + const bootstrapPage = page('older'); + const olderPage = page('older'); + const tailPage = { ...page('older'), throughSequence: 5 }; + let releaseTail: () => void = () => {}; + const tailGate = new Promise((resolve) => { + releaseTail = resolve; + }); + let signalTailEntered: () => void = () => {}; + const tailEntered = new Promise((resolve) => { + signalTailEntered = resolve; + }); + const changes: { durableUpserts: readonly { sequence: number }[] }[] = []; + const handle = runtimeHostSessionFixture({ + snapshot: continuitySnapshot(), + transcript: Promise.resolve([]), + events: { async *[Symbol.asyncIterator]() {} }, + transcriptBootstrap: { + throughSequence: 4, + overlayMessageCount: 0, + durable: bootstrapPage, + overlay: { ...page(null), source: 'overlay' }, + }, + loadTranscriptOverlay: async () => [], + decodeTranscriptPage: async (candidate) => candidate === bootstrapPage + ? { messages: messages.slice(3), nextCursor: 'older' } + : candidate === tailPage + ? { messages: [appended], nextCursor: 'older' } + : { messages: messages.slice(2, 4), nextCursor: 'older' }, + loadTranscriptPage: async (input) => { + if (input.throughSequence === 5) { + // Signal that catch-up is now parked inside the re-anchor's page await, + // so the test can `discard()` at exactly that point. + signalTailEntered(); + await tailGate; + return tailPage; + } + return olderPage; + }, + async close() {}, + }); + const maxResidentBytes = Buffer.byteLength(JSON.stringify(messages[0]!.message), 'utf8') + 1; + const replica = await DesktopTranscriptReplica.prepare(handle, { + maxResidentBytes, + onChange: (_replica, change) => changes.push(change), + }); + + await replica.loadBefore(4, 128 * 1024); + assert.equal(replica.snapshot().hasNewer, true); + changes.splice(0); + + // Start the tail re-anchor; wait until catch-up is parked inside its page + // await, then reclaim memory before the page resolves. + const advancing = replica.advance(5); + await tailEntered; + replica.discard(); + assert.equal(replica.resident, false); + releaseTail(); + await advancing; + + const upserts = changes.flatMap((change) => change.durableUpserts.map(({ sequence }) => sequence)); + assert.ok(!upserts.includes(5), 'a discarded replica must not be repopulated by an in-flight re-anchor'); + assert.equal(replica.resident, false); + assert.equal(replica.residentBytes, 0); +}); + +test('does not resurrect a discarded replica when a history load is in flight', async () => { + // Same post-await `#resident` invariant, exercised through `loadBefore`: a + // history page is in flight when `discard()` reclaims the replica. The + // resolved older page must not repopulate durable state or publish. + const messages = [0, 1, 2, 3, 4].map((sequence) => ({ + identity: sequence, + message: assistantMessage(String(sequence), `assistant-${sequence}`), + })); + const page = (nextCursor: string | null) => ({ + kind: 'page' as const, + sessionId: 'session-1', + source: 'durable' as const, + direction: 'older' as const, + throughSequence: 4, + rawBytes: 1, + fragments: [], + nextCursor, + }); + const bootstrapPage = page('older'); + const olderPage = page(null); + let releaseOlder: () => void = () => {}; + const olderGate = new Promise((resolve) => { + releaseOlder = resolve; + }); + let signalEntered: () => void = () => {}; + const olderEntered = new Promise((resolve) => { + signalEntered = resolve; + }); + const changes: { durableUpserts: readonly { sequence: number }[] }[] = []; + const handle = runtimeHostSessionFixture({ + snapshot: continuitySnapshot(), + transcript: Promise.resolve([]), + events: { async *[Symbol.asyncIterator]() {} }, + transcriptBootstrap: { + throughSequence: 4, + overlayMessageCount: 0, + durable: bootstrapPage, + overlay: { ...page(null), source: 'overlay' }, + }, + loadTranscriptOverlay: async () => [], + decodeTranscriptPage: async (candidate) => candidate === bootstrapPage + ? { messages: messages.slice(4), nextCursor: 'older' } + : { messages: messages.slice(2, 4), nextCursor: null }, + loadTranscriptPage: async () => { + signalEntered(); + await olderGate; + return olderPage; + }, + async close() {}, + }); + const replica = await DesktopTranscriptReplica.prepare(handle, { + maxResidentBytes: 1024 * 1024, + onChange: (_replica, change) => changes.push(change), + }); + + // Load older history; reclaim memory while its page is pending. + const loading = replica.loadBefore(4, 128 * 1024); + await olderEntered; + replica.discard(); + assert.equal(replica.resident, false); + releaseOlder(); + await loading; + + assert.equal(changes.length, 0, 'a discarded replica must not publish an in-flight history page'); + assert.equal(replica.resident, false); + assert.equal(replica.residentBytes, 0); +}); + +test('does not drive a discarded replica terminal when a contiguous catch-up is in flight', async () => { + // Same post-await `#resident` invariant on the ordinary contiguous catch-up + // path: another observed session's LRU `discard()` reclaims this replica while + // a `direction: 'newer'` page is pending. The per-page callback already returns + // early, but without the post-loop guard the watermark check would throw + // `correlation_changed` and drive the session terminal. A discarded replica has + // no watermark to meet — catch-up must return cleanly, not reject. + const messages = [0, 1, 2, 3, 4].map((sequence) => ({ + identity: sequence, + message: assistantMessage(String(sequence), `assistant-${sequence}`), + })); + const appended = { identity: 5, message: assistantMessage('5', 'assistant-5') }; + const page = (nextCursor: string | null, throughSequence: number) => ({ + kind: 'page' as const, + sessionId: 'session-1', + source: 'durable' as const, + direction: 'newer' as const, + throughSequence, + rawBytes: 1, + fragments: [], + nextCursor, + }); + const bootstrapPage = page(null, 4); + const newerPage = page(null, 5); + let releaseNewer: () => void = () => {}; + const newerGate = new Promise((resolve) => { + releaseNewer = resolve; + }); + let signalEntered: () => void = () => {}; + const newerEntered = new Promise((resolve) => { + signalEntered = resolve; + }); + const changes: { durableUpserts: readonly { sequence: number }[] }[] = []; + const handle = runtimeHostSessionFixture({ + snapshot: continuitySnapshot(), + transcript: Promise.resolve([]), + events: { async *[Symbol.asyncIterator]() {} }, + transcriptBootstrap: { + throughSequence: 4, + overlayMessageCount: 0, + durable: bootstrapPage, + overlay: { ...page(null, 4), source: 'overlay' }, + }, + loadTranscriptOverlay: async () => [], + decodeTranscriptPage: async (candidate) => candidate === bootstrapPage + ? { messages, nextCursor: null } + : { messages: [appended], nextCursor: null }, + loadTranscriptPage: async () => { + // Park catch-up inside the contiguous newer-page await so the test can + // reclaim memory at exactly that point. + signalEntered(); + await newerGate; + return newerPage; + }, + async close() {}, + }); + const replica = await DesktopTranscriptReplica.prepare(handle, { + maxResidentBytes: 1024 * 1024, + onChange: (_replica, change) => changes.push(change), + }); + // A large budget keeps the whole bootstrap resident, so the tail is contiguous + // (`hasNewer` false) and `advance` takes the paged catch-up, not the re-anchor. + assert.equal(replica.snapshot().hasNewer, false); + + // Advance the watermark contiguously; reclaim memory while the newer page is + // pending. Before the fix `advancing` rejects with `correlation_changed`. + const advancing = replica.advance(5); + await newerEntered; + replica.discard(); + assert.equal(replica.resident, false); + releaseNewer(); + await advancing; + + const upserts = changes.flatMap((change) => change.durableUpserts.map(({ sequence }) => sequence)); + assert.ok(!upserts.includes(5), 'a discarded replica must not be repopulated by an in-flight catch-up'); + assert.equal(replica.resident, false); + assert.equal(replica.residentBytes, 0); +}); + test('loads a history target with newer messages available below it', async () => { const messages = [0, 1, 2, 3, 4].map((sequence) => ({ identity: sequence, diff --git a/apps/desktop/src/main/desktop-transcript-replica.ts b/apps/desktop/src/main/desktop-transcript-replica.ts index b7408ca4ac..c41b994945 100644 --- a/apps/desktop/src/main/desktop-transcript-replica.ts +++ b/apps/desktop/src/main/desktop-transcript-replica.ts @@ -233,6 +233,11 @@ export class DesktopTranscriptReplica { }); await this.#withDecodedPage(page, (decoded) => { this.#assertOpen(); + // Same post-await `#resident` invariant as `#replaceWithRange` and the + // paged catch-up: a concurrent `discard()` may have reclaimed this + // replica while the older page was in flight. Installing the page here + // would repopulate durable state and undo the eviction. + if (!this.#resident) return; this.#acceptRange(decoded.messages); if ( anchor !== null && @@ -275,6 +280,13 @@ export class DesktopTranscriptReplica { }); await this.#withDecodedPage(page, (decoded) => { this.#assertOpen(); + // `#resident` can flip to false across the `await` above (a concurrent + // `discard()` reclaims memory for a non-visible session while the page is + // in flight). Re-anchoring here would repopulate durable state and undo + // the eviction, resurrecting a deliberately discarded replica past its + // memory budget. The paged catch-up guards its own post-await callback + // the same way; mirror it before mutating or publishing. + if (!this.#resident) return; this.#acceptRange(decoded.messages); if ( decoded.messages.length > 0 && @@ -359,8 +371,16 @@ export class DesktopTranscriptReplica { return; } if (this.#hasNewer) { - this.#durableThrough = target; - this.#publish([], [], []); + // The resident window was trimmed off the tail (e.g. after loading + // older history, `#evictToBudget(..., 'newest')` set `#hasNewer`), so + // `target` cannot be appended contiguously to what is resident. Bumping + // the watermark and publishing an empty change here silently dropped + // the freshly persisted message: an already-open consumer never learned + // about it and only a fresh subscription (the user switching sessions + // and back) re-read it. Re-anchor to the newest window instead — the + // same recovery a fresh subscription performs — so the append reaches + // open consumers live. + await this.#replaceWithRange(target, target, 512 * 1024); return; } let cursor: string | null = null; @@ -398,6 +418,14 @@ export class DesktopTranscriptReplica { cursor = decoded.nextCursor; }); } while (cursor !== null); + // A concurrent `discard()` (LRU reclaim for another observed session) can + // flip `#resident` to false across any page `await` above. The per-page + // callback already returns early in that case, so `expectedSequence` is + // left short of the watermark. Without this guard the check below would + // turn a benign memory reclaim into a fatal `correlation_changed` that + // drives the session terminal. A discarded replica has no watermark to + // meet, so return cleanly and let a later resume re-catch-up. + if (!this.#resident) return; if (expectedSequence !== target + 1) { throw correlationError('Desktop transcript catch-up ended before its watermark'); }