fix(desktop): deliver mid-session tail append after loading history

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<void>((resolve) => {
+    releaseTail = resolve;
+  });
+  let signalTailEntered: () => void = () => {};
+  const tailEntered = new Promise<void>((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<void>((resolve) => {
+    releaseOlder = resolve;
+  });
+  let signalEntered: () => void = () => {};
+  const olderEntered = new Promise<void>((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<void>((resolve) => {
+    releaseNewer = resolve;
+  });
+  let signalEntered: () => void = () => {};
+  const newerEntered = new Promise<void>((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,

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');
       }