fix(engine,client-core): read a failed leaf's shared prefix, announce settles, and hide unrun turns

Author: Zerlight

packages/client/core/src/client/conversation-graph-changes.ts

@@ -1,7 +1,8 @@
 import type { SessionId, TurnId } from '@linkcode/schema';
 import type { Unsubscribe } from '@linkcode/transport';
 
-/** One `conversation.graph.changed` broadcast: the graph moved its default leaf or gained shape. */
+/** One `conversation.graph.changed` broadcast: the graph moved its default leaf or gained shape
+ * (a new revision), or a turn reached its terminal state (the revision stands). */
 export interface ConversationGraphChange {
   graphRevision: number;
   activeLeafTurnId?: TurnId;
@@ -12,15 +13,16 @@ type ChangeCb = (change: ConversationGraphChange) => void;
 /**
  * Per-session register of the newest graph revision the daemon announced on this connection. Not
  * a buffer: a store holding a read at an older revision only needs to know that a newer one exists
- * and where its leaf is.
+ * and where its leaf is. Subscribers hear every announcement — a same-revision one carries a turn
+ * state they must refetch.
  */
 export class ConversationGraphChanges {
   private readonly latest = new Map<SessionId, ConversationGraphChange>();
   private readonly subscribers = new Map<SessionId, Set<ChangeCb>>();
 
   note(sessionId: SessionId, change: ConversationGraphChange): void {
     const current = this.latest.get(sessionId);
-    if (current !== undefined && current.graphRevision >= change.graphRevision) return;
+    if (current !== undefined && current.graphRevision > change.graphRevision) return;
     this.latest.set(sessionId, change);
     const subs = this.subscribers.get(sessionId);
     if (subs) for (const cb of subs) cb(change);

packages/foundation/schema/src/wire/conversation.ts

@@ -99,8 +99,9 @@ export const conversationWireVariants = [
     activeLeafTurnId: TurnIdSchema.optional(),
     turns: z.array(ConversationGraphTurnSchema),
   }),
-  /** Session-scoped broadcast: the graph changed shape or moved its default leaf; clients holding
-   * a stale snapshot revalidate via `conversation.graph.get`. */
+  /** Session-scoped broadcast: the graph gained a node or moved its default leaf (a new
+   * `graphRevision`), or a visible turn reached its terminal state (the revision stands); clients
+   * holding a stale snapshot revalidate via `conversation.graph.get`. */
   z.object({
     kind: z.literal('conversation.graph.changed'),
     sessionId: SessionIdSchema,

packages/host/engine/src/__tests__/conversation-projection.test.ts

@@ -411,12 +411,13 @@ describe('conversation projection attribution gate', () => {
     await store.saveTurn(shellTurn('turn-b2', 'turn-a', 'b', 'completed', 2));
 
     // The inactive sibling B1 carries IDENTICAL prompt text to the active B2: counts and
-    // fingerprints both pass, so only the active-lineage gate stops the mis-slice.
+    // fingerprints both pass, so only the active-lineage gate stops the mis-slice. The shared
+    // prefix reads where the active lineage reads.
     const inactive = await Effect.runPromise(
       service.read({ sessionId, leafTurnId: 'turn-b1' as TurnId }),
     );
-    expect(answers(inactive.events)).toEqual([]);
-    expect(placeholderTurnIds(inactive.events)).toEqual(['turn-a', 'turn-b1']);
+    expect(answers(inactive.events)).toEqual([['ans-a', 'turn-a']]);
+    expect(placeholderTurnIds(inactive.events)).toEqual(['turn-b1']);
 
     // The active lineage attributes normally.
     const active = await Effect.runPromise(service.read({ sessionId }));
@@ -461,15 +462,22 @@ describe('conversation projection attribution gate', () => {
       ...shellTurn('turn-b2', 'turn-a', 'b2', 'completed', 2),
       runId: forkRunId,
     });
+    // A continue from B1 that never ran: its own lineage still reads from B1's history.
+    await store.saveTurn({
+      ...shellTurn('turn-c1', 'turn-b1', 'c1', 'failed'),
+      runId: 'run-3' as RunId,
+    });
 
-    const inactive = await Effect.runPromise(
-      service.read({ sessionId, leafTurnId: 'turn-b1' as TurnId }),
-    );
-    expect(answers(inactive.events)).toEqual([
-      ['ans-a', 'turn-a'],
-      ['ans-b1', 'turn-b1'],
-    ]);
-    expect(placeholderTurnIds(inactive.events)).toEqual([]);
+    const expectOwnHistory = async (leafTurnId: TurnId) => {
+      const inactive = await Effect.runPromise(service.read({ sessionId, leafTurnId }));
+      expect(answers(inactive.events)).toEqual([
+        ['ans-a', 'turn-a'],
+        ['ans-b1', 'turn-b1'],
+      ]);
+      expect(placeholderTurnIds(inactive.events)).toEqual([]);
+    };
+    await expectOwnHistory('turn-b1' as TurnId);
+    await expectOwnHistory('turn-c1' as TurnId);
 
     const active = await Effect.runPromise(service.read({ sessionId }));
     expect(answers(active.events)).toEqual([
@@ -478,6 +486,34 @@ describe('conversation projection attribution gate', () => {
     ]);
   });
 
+  it('reads the shared prefix of a lineage whose leaf failed from the live history', async () => {
+    const { service, store } = await makeService({
+      journals: new ConversationLiveJournals(),
+      record: makeRecord('turn-l' as TurnId, true),
+      historyEvents: [
+        providerUser('u-a', 'a'),
+        providerAnswer('ans-a', 'answer a'),
+        providerUser('u-l', 'l'),
+        providerAnswer('ans-l', 'answer l'),
+      ],
+    });
+    await store.saveTurn(shellTurn('turn-a', null, 'a', 'completed'));
+    await store.saveTurn(shellTurn('turn-l', 'turn-a', 'l', 'completed', 1));
+    // An edit of L refused at dispatch: a failed sibling on the live run, nothing durable ran.
+    await store.saveTurn(shellTurn('turn-l2', 'turn-a', 'l2', 'failed', 2));
+
+    const failed = await Effect.runPromise(
+      service.read({ sessionId, leafTurnId: 'turn-l2' as TurnId }),
+    );
+    expect(answers(failed.events)).toEqual([['ans-a', 'turn-a']]);
+    expect(placeholderTurnIds(failed.events)).toEqual([]);
+    expect(
+      failed.events.flatMap((item) =>
+        'event' in item && item.event.type === 'user-message' ? [item.turnId] : [],
+      ),
+    ).toEqual(['turn-a', 'turn-l2']);
+  });
+
   it('attributes nothing when the trailing extra partition is not the in-flight prompt', async () => {
     const { service, store } = await makeService({
       journals: new ConversationLiveJournals(),

packages/host/engine/src/__tests__/engine-turn-submit.test.ts

@@ -920,12 +920,15 @@ describe('turn.submit saga', () => {
     await vi.waitFor(() => submittedTurnId(h.sent, 's1'));
     const firstTurnId = submittedTurnId(h.sent, 's1');
     await settleEngineTasks();
-    // One commit, and the turn's own stop (held until then) settled THIS turn with its checkpoint.
+    // One commit, and the turn's own stop (held until then) settled THIS turn with its checkpoint;
+    // the settle re-announces the tree at the same revision.
     expect((await h.conversationStore.listTurns(h.sessionId))[0].state).toBe('completed');
     expect(await h.conversationStore.listBindings(firstTurnId)).toEqual([
       expect.objectContaining({ checkpoint: 'after-first', capturedFrom: 'live' }),
     ]);
-    expect(h.sent.filter((p) => p.kind === 'conversation.graph.changed')).toHaveLength(1);
+    expect(
+      h.sent.flatMap((p) => (p.kind === 'conversation.graph.changed' ? [p.graphRevision] : [])),
+    ).toEqual([1, 1]);
 
     await submitPrompt(h, 's2', 'second');
     await vi.waitFor(() => expect(adapter.sentInputs).toHaveLength(2));

packages/host/engine/src/conversation/checkpoint-service.ts

@@ -15,7 +15,12 @@ import type { HistoryBranchCut, HistoryService } from '../session/history-servic
 import { promptContentFingerprint } from '../session/live-session';
 import type { SessionRecordRegistry } from '../session/session-record-registry';
 import type { CorpusAttribution } from './lineage-attribution';
-import { attributeCorpus, hasHiddenPrefix, pathToLeaf } from './lineage-attribution';
+import {
+  attributeCorpus,
+  hasHiddenPrefix,
+  pathToLeaf,
+  settledWithProvider,
+} from './lineage-attribution';
 import type { ConversationTurnService } from './turn-service';
 import { TERMINAL_TURN_STATES } from './turn-service';
 
@@ -269,8 +274,3 @@ function toCut(binding: ProviderTurnBinding): AgentHistoryBranchOptions {
 function activePath(record: SessionRecord, turns: ConversationTurn[]): ConversationTurn[] {
   return pathToLeaf(new Map(turns.map((turn) => [turn.turnId, turn])), record.activeLeafTurnId);
 }
-
-/** The path turns that expect provider rows: settled, and not failed (nothing durable ran). */
-function settledWithProvider(path: readonly ConversationTurn[]): ConversationTurn[] {
-  return path.filter((turn) => TERMINAL_TURN_STATES.has(turn.state) && turn.state !== 'failed');
-}

packages/host/engine/src/conversation/projection-service.ts

@@ -8,6 +8,7 @@ import type {
   ConversationGraphTurn,
   ConversationReadItem,
   ConversationTurn,
+  ConversationTurnState,
   ConversationWatermark,
   RunId,
   SessionId,
@@ -27,7 +28,7 @@ import { encodeLiveBranchCursor } from '../session/live-session';
 import type { SessionRecordRegistry } from '../session/session-record-registry';
 import type { ConversationCheckpointService } from './checkpoint-service';
 import type { ProviderPartition } from './lineage-attribution';
-import { pathToLeaf } from './lineage-attribution';
+import { pathToLeaf, settledWithProvider } from './lineage-attribution';
 import type { ConversationLiveJournals } from './live-journal';
 import { inflightChunkKey } from './live-journal';
 import type { ConversationTurnService } from './turn-service';
@@ -58,6 +59,8 @@ export interface ConversationReadResult {
 }
 
 const INPUT_SUMMARY_MAX_LENGTH = 140;
+/** Turns a peer can see: running or settled. */
+const VISIBLE_TURN_STATES = new Set<ConversationTurnState>(['running', ...TERMINAL_TURN_STATES]);
 const WHITESPACE_RUN_RE = /\s+/g;
 /** One page = one logical tunnel message; oversized reassembly is silently dropped by the tunnel
  * (the history-util.ts byte-budget rationale applies verbatim). */
@@ -95,6 +98,9 @@ export class ConversationProjectionService {
       sessionTurns.sort(byCreation);
       const graphTurns: ConversationGraphTurn[] = [];
       for (let i = 0, len = sessionTurns.length; i < len; i++) {
+        // A turn that has not run yet is the submitting client's alone: peers see it once it runs
+        // or fails, which is also when the tree announces it.
+        if (!VISIBLE_TURN_STATES.has(sessionTurns[i].state)) continue;
         const summary = yield* inputSummary(sessionTurns[i]);
         graphTurns.push(
           summary === undefined ? sessionTurns[i] : { ...sessionTurns[i], inputSummary: summary },
@@ -157,11 +163,9 @@ export class ConversationProjectionService {
         }
         offset = decoded.offset;
       }
-      // Positional attribution is sound only on the active lineage: a sibling lineage has the
-      // same path length by construction (and can carry identical prompt text on a retry), so an
-      // inactive-leaf read renders host rows + placeholders until per-turn bindings (CODE-632).
-      const isActiveLineage = leafTurnId !== undefined && leafTurnId === record.activeLeafTurnId;
-      const durable = yield* composeDurable(record, path, isActiveLineage);
+      const activePath =
+        leafTurnId === record.activeLeafTurnId ? path : pathToLeaf(byId, record.activeLeafTurnId);
+      const durable = yield* composeDurable(record, path, activePath);
       const { tail, watermark } = composeTail(request.sessionId, record.eventEpoch, path);
       const { events, nextOffset } = pageReadItems(
         durable,
@@ -194,36 +198,59 @@ export class ConversationProjectionService {
   private composeDurable(
     record: SessionRecord,
     path: ConversationTurn[],
-    isActiveLineage: boolean,
+    activePath: ConversationTurn[],
   ): Effect.Effect<ConversationReadItem[], OperationError> {
     const { checkpoints, records, turns } = this;
+    const hostContents = (
+      lineage: ConversationTurn[],
+    ): Effect.Effect<(ContentBlock[] | undefined)[], OperationError> =>
+      Effect.forEach(lineage, (turn) => turns.hostUserContent(turn));
     return Effect.gen(function* () {
       const items: ConversationReadItem[] = [];
-      const contents: (ContentBlock[] | undefined)[] = [];
-      for (let i = 0, len = path.length; i < len; i++) {
-        contents.push(yield* turns.hostUserContent(path[i]));
-      }
+      const contents = yield* hostContents(path);
       let attributed: ProviderPartition[] = [];
       let leading: AgentHistoryEvent[] = [];
-      // The active lineage reads the live history. An inactive lineage reads only a history of
-      // its own (its leaf run's, when that is not the live one): a sibling that shares the live
-      // history has the same path length by construction, so slicing it positionally would hand
-      // it the active lineage's rows. Reading the corpus also backfills replay bindings.
+      const settled = settledWithProvider(path);
+      const anchor = settled.at(-1);
+      const activeIds = new Set(activePath.map((turn) => turn.turnId));
       const liveHistoryId = records.historyId(record.sessionId);
-      const lineageHistoryId = isActiveLineage
-        ? liveHistoryId
-        : inactiveLineageHistoryId(record, path, liveHistoryId);
-      if (lineageHistoryId !== undefined) {
+      // Reading a corpus also backfills replay bindings on it.
+      const ownHistoryId =
+        anchor === undefined || activeIds.has(anchor.turnId)
+          ? undefined
+          : runHistoryId(record, anchor.runId);
+      if (ownHistoryId !== undefined && ownHistoryId !== liveHistoryId) {
+        // An inactive lineage on a history of its own reads it whole: rows, cursors, and bindings
+        // all live there — never the live copy a later fork made of its prefix.
         const attribution = yield* checkpoints.attributeLineage(
           record,
           path,
           contents,
-          lineageHistoryId,
+          ownHistoryId,
         );
         if (attribution !== undefined) {
           attributed = attribution.attributed;
           leading = attribution.leading;
         }
+      } else if (anchor !== undefined) {
+        // The turns a lineage shares with the active one read where the active lineage reads: the
+        // live history, verified from the start, cut to that shared prefix. Whatever lies beyond
+        // stays a placeholder — a sibling sharing the live history has the same path length by
+        // construction (and can repeat the prompt text on a retry), so slicing it positionally
+        // would hand it the active lineage's rows.
+        const shared = settled.filter((turn) => activeIds.has(turn.turnId)).length;
+        if (shared > 0) {
+          const attribution = yield* checkpoints.attributeLineage(
+            record,
+            activePath,
+            activePath === path ? contents : yield* hostContents(activePath),
+            liveHistoryId,
+          );
+          if (attribution !== undefined) {
+            attributed = attribution.attributed.slice(0, shared);
+            leading = attribution.leading;
+          }
+        }
       }
       for (let i = 0, len = leading.length; i < len; i++) {
         items.push(projectedItem(undefined, leading[i]));
@@ -467,17 +494,6 @@ function runHistoryId(record: SessionRecord, runId: RunId): AgentHistoryId | und
   return record.runs.find((run) => run.runId === runId)?.historyId;
 }
 
-function inactiveLineageHistoryId(
-  record: SessionRecord,
-  path: readonly ConversationTurn[],
-  liveHistoryId: AgentHistoryId | undefined,
-): AgentHistoryId | undefined {
-  const leaf = path.at(-1);
-  if (leaf === undefined) return;
-  const historyId = runHistoryId(record, leaf.runId);
-  return historyId === liveHistoryId ? undefined : historyId;
-}
-
 function projectedUserRow(
   turn: ConversationTurn,
   content: ContentBlock[],

packages/host/engine/src/conversation/turn-service.ts

@@ -343,22 +343,28 @@ export class ConversationTurnService {
       }
       // A failed turn keeps its ordinal and renders with a state badge, so every device must learn
       // the tree gained it — the default leaf did not move.
-      const graphRevision = this.records.commitGraphShape(sessionId);
-      const activeLeafTurnId = this.records.get(sessionId)?.activeLeafTurnId;
-      if (graphRevision !== undefined) {
-        this.transport.send(
-          createWireMessage({
-            kind: 'conversation.graph.changed',
-            sessionId,
-            graphRevision,
-            ...(activeLeafTurnId !== undefined && { activeLeafTurnId }),
-          }),
-        );
-      }
+      this.announceGraph(sessionId, true);
       return operation;
     });
   }
 
+  /** Every device refetches the tree. A node they did not have bumps the revision — the shape
+   * moved; a visible turn reaching its terminal state keeps it — only its badge changed, and a
+   * settle must not turn a peer's in-flight explicit-parent submit into a `conflict`. */
+  private announceGraph(sessionId: SessionId, gainedNode: boolean): void {
+    if (gainedNode) this.records.commitGraphShape(sessionId);
+    const record = this.records.get(sessionId);
+    if (record === undefined) return;
+    this.transport.send(
+      createWireMessage({
+        kind: 'conversation.graph.changed',
+        sessionId,
+        graphRevision: record.graphRevision,
+        ...(record.activeLeafTurnId !== undefined && { activeLeafTurnId: record.activeLeafTurnId }),
+      }),
+    );
+  }
+
   /** {@link resolveFailed} for exit paths inside a session-scoped fiber: enqueued on the engine
    * task runner so an interrupting teardown never waits behind the store write. */
   resolveFailedDetached(
@@ -544,6 +550,7 @@ export class ConversationTurnService {
     this.settledAt.set(turn.sessionId, Date.now());
     this.runTask(
       storeOperation('conversation.turn.save', () => this.store.saveTurn({ ...turn, state })).pipe(
+        Effect.tap(() => Effect.sync(() => this.announceGraph(turn.sessionId, false))),
         Effect.catch((error) =>
           Effect.logError(
             error.publicMessage,