fix(engine,client-core): unify automation echo identity and re-read when a leaf appears

Author: Zerlight

packages/client/core/src/client/event-buffer.ts

@@ -24,8 +24,7 @@ export interface SequencedAgentEvent {
   /** Client receive time (ms epoch), stamped when the event is ingested from the live stream.
    * Drives relative timestamps in the UI; absent for events replayed from a history read. */
   receivedAt?: number;
-  /** Daemon-minted `(epoch, seq)` position — the projection merge cut; absent from unstamped
-   * hosts (≤v79 daemons, the dev mock). */
+  /** Daemon-minted `(epoch, seq)` position — the projection merge cut; absent from ≤v79 hosts. */
   position?: ConversationWatermark;
   runId?: RunId;
   /** The turn the daemon attributed the event to. A live user echo carries none: it is broadcast

packages/client/core/src/conversation-store.ts

@@ -60,7 +60,7 @@ export function createConversationStore(
   if (seed !== undefined && 'items' in seed) {
     return createProjectionStore(client, sessionId, seed, options.onResync ?? noop);
   }
-  return createHistoryStore(client, sessionId, seed);
+  return createHistoryStore(client, sessionId, seed, options.onResync ?? noop);
 }
 
 /** Kinds the projection merge never drops on the watermark: their authoritative state lives in
@@ -290,6 +290,7 @@ function createHistoryStore(
   client: LinkCodeClient,
   sessionId: SessionId,
   seed: ConversationSeed | undefined,
+  onResync: (reason: ConversationResyncReason) => void,
 ): ConversationStore {
   const builder = createConversationBuilder();
   const uptoSeq = seed?.uptoSeq ?? 0;
@@ -328,6 +329,19 @@ function createHistoryStore(
   let seeded = false;
   /** Highest receive seq already examined (not necessarily folded — covered ones may be cut). */
   let consumedSeq = 0;
+  let resyncRequested = false;
+
+  const requestResync = (): void => {
+    if (resyncRequested) return;
+    resyncRequested = true;
+    queueMicrotask(() => onResync('graph'));
+  };
+
+  const noteGraph = (change: ConversationGraphChange | undefined): void => {
+    // A leaf appearing after an empty-graph / live-only read is the cutover: the owner must
+    // re-read so the next store is a projection. History-path sessions never see this.
+    if (change?.activeLeafTurnId !== undefined) requestResync();
+  };
 
   const sync = (): void => {
     if (!seeded) {
@@ -355,7 +369,17 @@ function createHistoryStore(
   };
 
   return {
-    subscribe: (onStoreChange) => client.subscribe(sessionId, onStoreChange),
+    subscribe(onStoreChange) {
+      noteGraph(client.latestGraphChange(sessionId));
+      const unsubscribeEvents = client.subscribe(sessionId, onStoreChange);
+      const unsubscribeGraph = client.subscribeGraphChanges(sessionId, (change) => {
+        noteGraph(change);
+      });
+      return () => {
+        unsubscribeEvents();
+        unsubscribeGraph();
+      };
+    },
     getSnapshot() {
       sync();
       return builder.snapshot();

packages/client/core/tests/integration/conversation-client.test.ts

@@ -66,7 +66,7 @@ describe('LinkCodeClient conversation graph API', () => {
       { seq: 1, position: { epoch: 4, seq: 7 }, runId: 'run-1', turnId: 'turn-1' },
       { seq: 2 },
     ]);
-    // An unstamped frame (≤v79 host, the dev mock) carries no position at all.
+    // An unstamped frame (≤v79 host) carries no position at all.
     expect(client.eventsSnapshot(sessionId)[1]).not.toHaveProperty('position');
     expect(seen.map((entry) => entry.position)).toEqual([{ epoch: 4, seq: 7 }, undefined]);
 

packages/client/core/tests/integration/conversation-store-projection.test.ts

@@ -197,6 +197,18 @@ describe('projection conversation store', () => {
     h.close();
   });
 
+  it('asks a live-only store to re-read when a leaf appears', async () => {
+    const h = await harness();
+    const store = createConversationStore(h.client, sessionId, undefined, {
+      onResync: (reason) => h.resyncs.push(reason),
+    });
+    store.subscribe(noop);
+    h.graphChanged(1, turn(1));
+    await tick();
+    expect(h.resyncs).toEqual(['graph']);
+    h.close();
+  });
+
   it('treats a graph move onto a leaf whose row arrived live as a plain continuation', async () => {
     const h = await harness();
     const store = h.store(seedOf([userRow(1, 'first')], { epoch: 1, seq: 1 }, 3));

packages/client/workbench/src/mock/dev-mock-host.ts

@@ -1274,11 +1274,6 @@ export class DevMockHost {
       case 'shell-command': {
         const content = [textBlock(`$ ${input.command}`)];
         const turn = this.beginTurn(session, content, input);
-        this.emit(sessionId, {
-          type: 'user-message',
-          messageId: userRowMessageId(turn.graph.turnId),
-          content,
-        });
         settleTurn(session, turn, 'completed');
         this.sendSuccess(replyTo);
         break;
@@ -1317,11 +1312,6 @@ export class DevMockHost {
       name,
       ...(args !== undefined && { arguments: args }),
     });
-    this.emit(session.sessionId, {
-      type: 'user-message',
-      messageId: userRowMessageId(turn.graph.turnId),
-      content,
-    });
     session.status = 'running';
     this.emit(session.sessionId, { type: 'status', status: 'running' });
     if (fixture.reply === undefined) {
@@ -1365,16 +1355,11 @@ export class DevMockHost {
     const turn = this.beginTurn(session, content, p.input.type === 'prompt' ? undefined : p.input);
     this.send({ kind: 'turn.submitted', replyTo: p.clientReqId, turnId: turn.graph.turnId });
     if (p.input.type === 'prompt') {
-      const result = await this.streamMockReply(session, turn, content);
+      const result = await this.streamMockReply(session, content);
       settleTurn(session, turn, result.ok ? 'completed' : 'failed');
       return;
     }
     // Command/shell turns just echo — the mock has no directive execution behind turn.submit.
-    this.emit(p.sessionId, {
-      type: 'user-message',
-      messageId: userRowMessageId(turn.graph.turnId),
-      content,
-    });
     settleTurn(session, turn, 'completed');
   }
 
@@ -1405,6 +1390,13 @@ export class DevMockHost {
     };
     session.graphTurns.push(turn);
     session.runningTurnId = turnId;
+    // Echo before graph.changed so a subscribed projection store sees the new leaf row and
+    // treats a plain send as continuation, matching the engine dispatcher.
+    this.emit(session.sessionId, {
+      type: 'user-message',
+      messageId: userRowMessageId(turnId),
+      content,
+    });
     this.send({
       kind: 'conversation.graph.changed',
       sessionId: session.sessionId,
@@ -1420,25 +1412,19 @@ export class DevMockHost {
     content: ContentBlock[],
   ): Promise<void> {
     const turn = this.beginTurn(session, content);
-    const result = await this.streamMockReply(session, turn, content);
+    const result = await this.streamMockReply(session, content);
     settleTurn(session, turn, result.ok ? 'completed' : 'failed');
     if (result.ok) this.sendSuccess(replyTo);
     else this.sendFailure(replyTo, result.message, { reportedInConversation: true });
   }
 
   private async streamMockReply(
     session: MockSession,
-    turn: MockTurn,
     content: ContentBlock[],
   ): Promise<{ ok: true } | { ok: false; message: string }> {
     const text = promptText(content);
     if (text && !session.title) session.title = text.slice(0, 80);
     session.status = 'running';
-    this.emit(session.sessionId, {
-      type: 'user-message',
-      messageId: userRowMessageId(turn.graph.turnId),
-      content,
-    });
     this.emit(session.sessionId, { type: 'status', status: 'running' });
 
     // Cancel/stop bump the session epoch; a stale epoch means this turn was cancelled and the

packages/client/workbench/tests/integration/dev-mock-projection.test.ts

@@ -5,6 +5,7 @@ import {
   readConversationSeed,
 } from '@linkcode/client-core';
 import { userRowMessageId } from '@linkcode/schema';
+import { noop } from 'foxact/noop';
 import { nullthrow } from 'foxts/guard';
 import { wait } from 'foxts/wait';
 import { describe, expect, it } from 'vitest';
@@ -23,7 +24,15 @@ describe('dev mock projection seeding', () => {
     // No turn rows and no transcript: nothing to seed, the store runs live-only.
     await expect(readConversationSeed(client, source)).resolves.toBeUndefined();
 
+    const liveResyncs: ConversationResyncReason[] = [];
+    createConversationStore(client, sessionId, undefined, {
+      onResync: (reason) => liveResyncs.push(reason),
+    }).subscribe(noop);
+
     await client.promptText(sessionId, 'Hello mocked daemon');
+    await wait(10);
+    expect(liveResyncs).toEqual(['graph']);
+
     const change = nullthrow(client.latestGraphChange(sessionId));
     const leaf = nullthrow(change.activeLeafTurnId);
 
@@ -38,6 +47,7 @@ describe('dev mock projection seeding', () => {
     const store = createConversationStore(client, sessionId, seed, {
       onResync: (reason) => resyncs.push(reason),
     });
+    store.subscribe(noop);
     const items = store.getSnapshot().items;
     // One user row under the turn identity — the live echo folded once, never twice.
     expect(items.filter((item) => item.kind === 'message' && item.role === 'user')).toEqual([
@@ -48,6 +58,8 @@ describe('dev mock projection seeding', () => {
 
     // A turn without output renders the prompt-only placeholder under its row.
     await client.runShellCommand(sessionId, 'ls');
+    await wait(10);
+    expect(resyncs).toEqual([]);
     const reseed = await readConversationSeed(client, source);
     if (reseed === undefined || !('items' in reseed)) throw new Error('expected a projection seed');
     const shellLeaf = nullthrow(client.latestGraphChange(sessionId)?.activeLeafTurnId);

packages/host/engine/src/__tests__/engine-schedule.test.ts

@@ -232,6 +232,18 @@ describe('engine schedule wiring', () => {
       graphRevision: 1,
       activeLeafTurnId: turn.turnId,
     });
+    expect(
+      h.sent.filter(
+        (payload) => payload.kind === 'agent.event' && payload.event.type === 'user-message',
+      ),
+    ).toEqual([
+      expect.objectContaining({
+        event: expect.objectContaining({
+          type: 'user-message',
+          messageId: `msg-${turn.turnId}`,
+        }),
+      }),
+    ]);
   });
 
   it('reports an unknown schedule as not found', async () => {

packages/host/engine/src/session/orchestrator.ts

@@ -1,5 +1,4 @@
 import type { AdapterFactory, AgentAdapter, BrowserToolsetFactory } from '@linkcode/agent-adapter';
-import { nextMessageId } from '@linkcode/agent-adapter';
 import type {
   AgentEvent,
   AgentHistoryCapabilities,
@@ -13,6 +12,7 @@ import type {
   SessionInfo,
   SessionRecord,
 } from '@linkcode/schema';
+import { userRowMessageId } from '@linkcode/schema';
 import type { Transport } from '@linkcode/transport';
 import { createWireMessage } from '@linkcode/transport';
 import { Cause, Deferred, Effect, Exit, Scope } from 'effect';
@@ -221,9 +221,15 @@ export class SessionOrchestrator {
             input: { type: 'prompt', blocks: promptBlocksFromContent(content) },
           });
           const result = yield* Effect.sync(() => {
-            this.events.broadcast(sessionId, session, [
-              { type: 'user-message', messageId: nextMessageId(), content },
-            ]);
+            this.events.broadcast(
+              sessionId,
+              session,
+              session.trackPrompt(
+                userRowMessageId(intent.turn.turnId),
+                content,
+                intent.turn.turnId,
+              ),
+            );
             records.setTitleFromContent(sessionId, content);
           }).pipe(
             Effect.andThen(

packages/host/engine/src/session/session-input-dispatcher.ts

@@ -138,9 +138,11 @@ export class SessionInputDispatcher {
               input.type === 'command'
                 ? `/${input.name}${input.arguments ? ` ${input.arguments}` : ''}`
                 : `$ ${input.command}`;
-            events.broadcast(sessionId, session, [
-              { type: 'user-message', messageId: echoMessageId, content: [{ type: 'text', text }] },
-            ]);
+            events.broadcast(
+              sessionId,
+              session,
+              session.trackPrompt(echoMessageId, [{ type: 'text', text }], persistedTurnId),
+            );
           }
         }
         const responseInput =