fix(runtime,desktop): order same-revision catalog reads by a run epoch

apps/desktop/src/main/__tests__/app-shell-session-ui-state.test.ts

@@ -373,6 +373,85 @@ describe('shellSessionRowEqual', () => {
     assert.equal(shellSessionRowEqual(future, row), false);
   });
 
+  it('orders same-revision rows by the live run epoch (#5713)', async () => {
+    const { root } = installReactRenderer();
+    try {
+      const catalog = createSessionCatalogController();
+      catalog.commitSessions([{
+        ...row,
+        revision: 5,
+        runningTurnIds: ['turn-1'],
+        runHostGeneration: 'host-1',
+        runEpoch: 2,
+      }]);
+
+      // A read taken before the turn started lands after the running patch:
+      // same revision, same host generation, older epoch — it must not flip
+      // the row back to idle.
+      await act(async () => {
+        catalog.commitPatch(row.id, {
+          ...row,
+          revision: 5,
+          runningTurnIds: [],
+          runHostGeneration: 'host-1',
+          runEpoch: 1,
+        });
+      });
+      assert.deepEqual(
+        selectSessionById(catalog.getState(), row.id)?.runningTurnIds,
+        ['turn-1'],
+        'the older live state must not overwrite the newer',
+      );
+
+      // A genuinely newer epoch updates the row even at the same revision.
+      await act(async () => {
+        catalog.commitPatch(row.id, {
+          ...row,
+          revision: 5,
+          runningTurnIds: [],
+          runHostGeneration: 'host-1',
+          runEpoch: 3,
+        });
+      });
+      assert.deepEqual(selectSessionById(catalog.getState(), row.id)?.runningTurnIds, []);
+
+      // A Host restart is a new generation: the previous host is gone, so
+      // its row cannot out-rank the restarted host's first read, whatever
+      // each side's epoch counter reads — a wall clock is not monotonic
+      // across processes (#5713 review round two).
+      await act(async () => {
+        catalog.commitPatch(row.id, {
+          ...row,
+          revision: 5,
+          runningTurnIds: ['turn-2'],
+          runHostGeneration: 'host-2',
+          runEpoch: 1,
+        });
+      });
+      assert.deepEqual(
+        selectSessionById(catalog.getState(), row.id)?.runningTurnIds,
+        ['turn-2'],
+        'the restarted host must take over the row',
+      );
+
+      // Within the restarted generation the counter orders reads again.
+      await act(async () => {
+        catalog.commitPatch(row.id, {
+          ...row,
+          revision: 5,
+          runningTurnIds: ['turn-2'],
+          runHostGeneration: 'host-2',
+          runEpoch: 0,
+        });
+      });
+      assert.deepEqual(
+        selectSessionById(catalog.getState(), row.id)?.runningTurnIds,
+        ['turn-2'],
+        'the older read of the restarted generation must not win',
+      );
+    } finally { cleanupFakeDom(); }
+  });
+
   it('keeps a catalog row subscriber mounted through rail-only patches', async () => {
     const { root } = installReactRenderer();
     try {

apps/desktop/src/renderer/application/contracts/session-catalog/session-catalog-state.ts

@@ -188,9 +188,39 @@ export function waitForCatalogSession(
   });
 }
 
-/** A committed row at a newer revision is authoritative over an older snapshot of it. */
+/**
+ * A committed row at a newer revision is authoritative over an older snapshot
+ * of it. Equal revisions tie on the live run state's own order: a turn
+ * starting or ending does not move `revision`, so two same-revision reads can
+ * disagree about `runningTurnIds` — the run epoch says which observation is
+ * older, and the stale one must not overwrite the fresher (#5713).
+ *
+ * The epoch counter only orders observations of one Host generation.
+ * Generations themselves are not ordered, so a read from a different
+ * generation is never stale: a restarted Host must take the row over from its
+ * predecessor whatever the two counters read (#5713 review). A successful
+ * cross-generation response cannot exist on the wire, either: closing a
+ * connection rejects every in-flight request with `connection_lost`
+ * (client/connection.ts), so a lagging predecessor read never delivers after
+ * the successor's row has landed.
+ */
 function isStaleSummary(prior: DesktopSessionSummary, next: DesktopSessionSummary): boolean {
-  return prior.revision > next.revision;
+  if (prior.revision !== next.revision) return prior.revision > next.revision;
+  const priorGeneration = prior.runHostGeneration;
+  const nextGeneration = next.runHostGeneration;
+  if (
+    priorGeneration !== undefined &&
+    nextGeneration !== undefined &&
+    priorGeneration !== nextGeneration
+  ) {
+    return false;
+  }
+  const priorEpoch = prior.runEpoch;
+  const nextEpoch = next.runEpoch;
+  if (priorEpoch === undefined || nextEpoch === undefined || priorEpoch === nextEpoch) {
+    return false;
+  }
+  return priorEpoch > nextEpoch;
 }
 
 export const selectSessions = (state: SessionCatalogState): readonly DesktopSessionSummary[] =>

packages/core/src/session.ts

@@ -403,6 +403,26 @@ export interface SessionSummary {
    * the header alone and omits it.
    */
   runningTurnIds?: string[];
+  /**
+   * Bumped by the runtime each time a turn of this session starts or ends.
+   * `revision` does not move for those transitions, so two same-revision
+   * summaries can disagree about `runningTurnIds` — the epoch orders them:
+   * the higher epoch is the newer observation (#5713). Present alongside
+   * `runningTurnIds` under the same population rules.
+   *
+   * The counter restarts at zero with a fresh Host process, so it only orders
+   * observations of one host generation: summaries whose `runHostGeneration`
+   * differs are not comparable by epoch, and the newer generation's host owns
+   * the row outright.
+   */
+  runEpoch?: number;
+  /**
+   * Identifies the Host process generation that produced this live-run
+   * observation. Summaries from different generations are not ordered by
+   * `runEpoch` — a restarted Host supersedes every observation its
+   * predecessor published, whatever the epoch counters read (#5713).
+   */
+  runHostGeneration?: string;
   parentSessionId?: string;
   branchOfTurnId?: string;
   subagent?: SessionSubagentProjection;

packages/runtime-host/src/__tests__/authenticated-websocket.test.ts

@@ -58,6 +58,7 @@ const PROTOCOL = {
 const KNOWN_EMPTY_LIVE_RUN_STATE = {
   schemaVersion: SESSION_CATALOG_LIVE_RUN_STATE_SCHEMA_VERSION,
   runningTurnIds: [],
+  runEpoch: 0,
 } as const;
 
 async function configureTestModel(local: RuntimeHostConnection): Promise<void> {
@@ -309,15 +310,25 @@ test('one Local IPC owner and one authenticated WebSocket Client control the sam
     if (closed.value?.kind === 'subscription.closed') {
       assert.equal(closed.value.reason, 'access_revoked');
     }
+    const sharedRead = await remote.request('session.catalog.query', {
+      kind: 'get',
+      sessionId: 'shared-session',
+    });
+    assert.equal(sharedRead.kind, 'session');
+    const sharedSession = sharedRead.kind === 'session' ? sharedRead.session : null;
+    assert.ok(sharedSession && !('kind' in sharedSession));
+    if ('kind' in sharedSession) assert.fail('the shared row must be a current projection');
+    // The host generation is unique per Host process, so assert its shape and
+    // compare the known-empty remainder.
+    const { hostGeneration: sharedGeneration, ...sharedKnownEmpty } =
+      sharedSession.liveRunState ?? {};
+    if (typeof sharedGeneration !== 'string' || sharedGeneration.length === 0) {
+      assert.fail('the host generation must be a non-empty string');
+    }
+    assert.deepEqual(sharedKnownEmpty, KNOWN_EMPTY_LIVE_RUN_STATE);
     assert.deepEqual(
-      await remote.request('session.catalog.query', {
-        kind: 'get',
-        sessionId: 'shared-session',
-      }),
-      {
-        kind: 'session',
-        session: { ...created, liveRunState: KNOWN_EMPTY_LIVE_RUN_STATE },
-      },
+      { ...sharedSession, liveRunState: KNOWN_EMPTY_LIVE_RUN_STATE },
+      { ...created, liveRunState: KNOWN_EMPTY_LIVE_RUN_STATE },
     );
 
     const catalogChanged = new Promise<string>((resolve) => {
@@ -330,16 +341,23 @@ test('one Local IPC owner and one authenticated WebSocket Client control the sam
     });
     assert.equal(renamed.kind, 'committed');
     assert.equal(await catalogChanged, 'shared-session');
+    const localRead = await local.request('session.catalog.query', {
+      kind: 'get',
+      sessionId: 'shared-session',
+    });
+    assert.equal(localRead.kind, 'session');
+    const localSession = localRead.kind === 'session' ? localRead.session : null;
+    assert.ok(localSession && !('kind' in localSession));
+    if ('kind' in localSession) assert.fail('the local row must be a current projection');
+    const { hostGeneration: localGeneration, ...localKnownEmpty } = localSession.liveRunState ?? {};
+    if (typeof localGeneration !== 'string' || localGeneration.length === 0) {
+      assert.fail('the host generation must be a non-empty string');
+    }
+    assert.deepEqual(localKnownEmpty, KNOWN_EMPTY_LIVE_RUN_STATE);
     assert.deepEqual(
-      await local.request('session.catalog.query', {
-        kind: 'get',
-        sessionId: 'shared-session',
-      }),
+      { ...localSession, liveRunState: KNOWN_EMPTY_LIVE_RUN_STATE },
       renamed.kind === 'committed'
-        ? {
-            kind: 'session',
-            session: { ...renamed.session, liveRunState: KNOWN_EMPTY_LIVE_RUN_STATE },
-          }
+        ? { ...renamed.session, liveRunState: KNOWN_EMPTY_LIVE_RUN_STATE }
         : assert.fail('Remote Session rename did not commit'),
     );
 
@@ -616,7 +634,29 @@ test('an authenticated WebSocket Client reconnects after service restart to cano
       websocket: { host: '127.0.0.1', port },
     });
 
-    assert.deepEqual(await recovered, expected);
+    const canonical = await recovered;
+    assert.equal(canonical.kind, 'session');
+    assert.equal(expected.kind, 'session');
+    const beforeSession = expected.kind === 'session' ? expected.session : null;
+    const afterSession = canonical.kind === 'session' ? canonical.session : null;
+    assert.ok(beforeSession && afterSession);
+    if ('kind' in beforeSession || 'kind' in afterSession) {
+      assert.fail('the canonical reads must be current projections');
+    }
+    const beforeLive = beforeSession.liveRunState;
+    const afterLive = afterSession.liveRunState;
+    assert.ok(beforeLive && afterLive);
+    // The restart must change the host generation — the wire signal that lets
+    // clients tell the restarted Host's reads apart from its predecessor's
+    // (#5713).
+    assert.notEqual(afterLive.hostGeneration, beforeLive.hostGeneration);
+    assert.deepEqual(
+      {
+        ...afterSession,
+        liveRunState: { ...afterLive, hostGeneration: beforeLive.hostGeneration },
+      },
+      beforeSession,
+    );
     assert.notEqual(remote.hostEpoch, firstHostEpoch);
   } finally {
     await Promise.allSettled([remote?.close(), local?.close()]);

packages/runtime-host/src/__tests__/session-catalog-coordinator.test.ts

@@ -314,6 +314,8 @@ test('catalog queries project known-empty and running state from Runtime authori
   assert.deepEqual(emptyOutcome.result.session.liveRunState, {
     schemaVersion: 1,
     runningTurnIds: [],
+    runEpoch: 0,
+    hostGeneration: 'test-host-generation',
   });
 
   runningTurnIds = ['turn-live'];
@@ -332,6 +334,8 @@ test('catalog queries project known-empty and running state from Runtime authori
   assert.deepEqual(session.liveRunState, {
     schemaVersion: 1,
     runningTurnIds: ['turn-live'],
+    runEpoch: 0,
+    hostGeneration: 'test-host-generation',
   });
 });
 
@@ -379,6 +383,8 @@ test('catalog queries de-duplicate Runtime live turn ids in stable order', async
   assert.deepEqual(outcome.result.session.liveRunState, {
     schemaVersion: 1,
     runningTurnIds: ['turn-a', 'turn-b'],
+    runEpoch: 0,
+    hostGeneration: 'test-host-generation',
   });
 });
 
@@ -2202,6 +2208,8 @@ function createFixture(
   const runtimePolicy = options.runtimePolicy ?? runtimePolicyFixture(options.connection ?? {});
   const manager: ConfigurationAuthority = {
     runningTurnIds: () => [],
+    sessionRunEpoch: () => 0,
+    sessionHostGeneration: () => 'test-host-generation',
     transitionSessionConfiguration: async (_sessionId, input) => {
       header = {
         ...header,