feat(client-core): merge conversations by daemon sequence and turn identity

apps/mobile/src/components/conversation/timeline-item.tsx

@@ -140,6 +140,12 @@ export function TimelineItem({
           ) : null}
         </View>
       );
+    case 'history-unavailable':
+      return (
+        <View className="flex-row items-center justify-center gap-2 px-2">
+          <Text className="text-footnote text-muted">{t('historyUnavailable')}</Text>
+        </View>
+      );
     default:
       return null;
   }

apps/mobile/src/runtime/__tests__/use-seeded-conversation.test.ts

@@ -5,15 +5,19 @@ import type {
   MessageId,
   SessionId,
   SessionInfo,
+  TurnId,
   WirePayload,
 } from '@linkcode/schema';
+import { userRowMessageId } from '@linkcode/schema';
 import { useSeededConversation } from '@mobile/runtime/use-seeded-conversation';
 import { renderHook, waitFor } from '@testing-library/react';
 import { expect, it } from 'vitest';
+import type { ControlledTransport } from './client-test-helpers';
 import { clientWrapper, connectClient } from './client-test-helpers';
 
 const SESSION = 'session-1' as SessionId;
 const HISTORY = 'history-1' as AgentHistoryId;
+const TURN = 'turn-1' as TurnId;
 
 const SESSION_INFO: SessionInfo = {
   sessionId: SESSION,
@@ -25,7 +29,7 @@ const SESSION_INFO: SessionInfo = {
   historyId: HISTORY,
 };
 
-/** Same session with no `historyId`, so the seed read short-circuits and only attach runs. */
+/** Same session with no `historyId`: without a turn graph there is nothing left to read. */
 const NO_HISTORY: SessionInfo = { ...SESSION_INFO, historyId: undefined };
 
 function kinds(sent: readonly WirePayload[]): string[] {
@@ -40,28 +44,52 @@ async function mountSeeded(sessionId: SessionId | null, session: SessionInfo | n
   return { transport, client, view };
 }
 
+/** Answer the newest `conversation.read` as the daemon would for a session without turn rows. */
+async function answerEmptyGraph(transport: ControlledTransport): Promise<void> {
+  await waitFor(() => expect(kinds(transport.sent)).toContain('conversation.read'));
+  const read = transport.sent.findLast((payload) => payload.kind === 'conversation.read');
+  if (read?.kind !== 'conversation.read') throw new Error('no conversation.read');
+  transport.receive({
+    kind: 'conversation.read.result',
+    replyTo: read.clientReqId,
+    sessionId: SESSION,
+    graphRevision: 0,
+    watermark: { epoch: 0, seq: 0 },
+    events: [],
+  });
+}
+
 it('announces the route session before the session list resolves', async () => {
   // The screen has the id from the route immediately but `SessionInfo` only a round-trip later.
   // Waiting for it would let the connection sit in `attached` scope with nothing announced, and
   // the events dropped in that window are not recoverable from the attach replay.
   const { transport, client } = await mountSeeded(SESSION, null);
 
   expect(transport.sent).toContainEqual({ kind: 'session.attach', sessionId: SESSION });
+  expect(kinds(transport.sent)).not.toContain('conversation.read');
   expect(kinds(transport.sent)).not.toContain('history.read');
   client.dispose();
 });
 
-it('announces the session before reading its history', async () => {
+it('announces the session, reads its turn graph, and falls back to the transcript', async () => {
   const { transport, client } = await mountSeeded(SESSION, SESSION_INFO);
 
-  await waitFor(() => expect(kinds(transport.sent)).toContain('history.read'));
   // Attaching first is what asks the daemon to re-broadcast the buffered per-session state; the
-  // read then walks the transcript. Reversed, the re-broadcast would land after the seed sampled
-  // its cut.
+  // read then follows. Reversed, the re-broadcast would land after the seed sampled its cut.
+  await answerEmptyGraph(transport);
+  await waitFor(() => expect(kinds(transport.sent)).toContain('history.read'));
   expect(kinds(transport.sent).indexOf('session.attach')).toBeLessThan(
-    kinds(transport.sent).indexOf('history.read'),
+    kinds(transport.sent).indexOf('conversation.read'),
   );
-  expect(transport.sent).toContainEqual({ kind: 'session.attach', sessionId: SESSION });
+  client.dispose();
+});
+
+it('reads nothing further for a graph-less session without a transcript', async () => {
+  const { transport, client } = await mountSeeded(SESSION, NO_HISTORY);
+
+  await answerEmptyGraph(transport);
+  await waitFor(() => expect(kinds(transport.sent)).toContain('conversation.read'));
+  expect(kinds(transport.sent)).not.toContain('history.read');
   client.dispose();
 });
 
@@ -87,8 +115,55 @@ it('announces nothing when there is no session to observe', async () => {
   client.dispose();
 });
 
+it('seeds from the turn-graph projection and re-reads when the store asks to', async () => {
+  const { transport, client, view } = await mountSeeded(SESSION, SESSION_INFO);
+  await waitFor(() => expect(kinds(transport.sent)).toContain('conversation.read'));
+  const read = transport.sent.find((payload) => payload.kind === 'conversation.read');
+  if (read?.kind !== 'conversation.read') throw new Error('no conversation.read');
+  transport.receive({
+    kind: 'conversation.read.result',
+    replyTo: read.clientReqId,
+    sessionId: SESSION,
+    graphRevision: 1,
+    leafTurnId: TURN,
+    watermark: { epoch: 1, seq: 1 },
+    events: [
+      {
+        turnId: TURN,
+        event: {
+          type: 'user-message',
+          messageId: userRowMessageId(TURN),
+          content: [{ type: 'text', text: 'hi' }],
+        },
+      },
+    ],
+  });
+
+  await waitFor(() =>
+    expect(view.result.current.items).toContainEqual(
+      expect.objectContaining({ kind: 'message', role: 'user', id: userRowMessageId(TURN) }),
+    ),
+  );
+  // A projection covers the transcript; the legacy read never runs.
+  expect(kinds(transport.sent)).not.toContain('history.read');
+
+  // A relaunch (new epoch) must be re-read: the hook answers the store's request with a fresh read.
+  transport.receive({
+    kind: 'agent.event',
+    sessionId: SESSION,
+    epoch: 2,
+    seq: 1,
+    event: { type: 'status', status: 'running' },
+  });
+  await waitFor(() =>
+    expect(kinds(transport.sent).filter((kind) => kind === 'conversation.read')).toHaveLength(2),
+  );
+  client.dispose();
+});
+
 it('keeps an ask the re-broadcast delivered before the seed cut', async () => {
   const { transport, client, view } = await mountSeeded(SESSION, SESSION_INFO);
+  await answerEmptyGraph(transport);
   await waitFor(() => expect(kinds(transport.sent)).toContain('history.read'));
 
   // What `session.attach` buys: an ask raised while the thread was closed. It arrives before the

apps/mobile/src/runtime/use-seeded-conversation.ts

@@ -1,25 +1,30 @@
-import type { Conversation, ConversationSeed, ConversationSeedEvent } from '@linkcode/client-core';
-import { useConversation, useLinkCodeClient } from '@linkcode/client-core';
+import type {
+  Conversation,
+  ConversationProjectionSeed,
+  ConversationSeed,
+} from '@linkcode/client-core';
+import { readConversationSeed, useConversation, useLinkCodeClient } from '@linkcode/client-core';
 import type { SessionId, SessionInfo } from '@linkcode/schema';
 import { noop } from 'foxact/noop';
 import { useEffect } from 'foxact/use-abortable-effect';
-import { useState } from 'react';
+import { useReducer, useState } from 'react';
 
-/** Upper bound on cursor pages one seed read follows, so a buggy cursor can't loop forever. */
-const MAX_SEED_PAGES = 20;
+type Seed = ConversationProjectionSeed | ConversationSeed;
 
 /**
- * The session's conversation view-model seeded from provider history — the live `agent.event`
- * subscription only covers this connection, so a cold-opened session replays its past from
- * `history.read` (same read walk as workbench's useSeededConversation, without the SWR cache).
- * A failed read degrades to live-only; the seed is keyed by session so it never bleeds across.
+ * The session's conversation view-model seeded from the daemon: the turn-graph projection where the
+ * host serves one, the provider transcript otherwise — the live `agent.event` subscription only
+ * covers this connection (same read as workbench's useSeededConversation, without the SWR cache).
+ * A failed read degrades to live-only; the seed is keyed by session so it never bleeds across, and
+ * a projection store's resync request re-runs the read.
  */
 export function useSeededConversation(
   sessionId: SessionId | null,
   session: SessionInfo | null,
 ): Conversation {
   const client = useLinkCodeClient();
-  const [seeded, setSeeded] = useState<{ for: SessionId; seed: ConversationSeed } | null>(null);
+  const [seeded, setSeeded] = useState<{ for: SessionId; seed: Seed } | null>(null);
+  const [readGeneration, requestReread] = useReducer((n: number) => n + 1, 0);
 
   const agentKind = session?.kind;
   const cwd = session?.cwd;
@@ -30,7 +35,7 @@ export function useSeededConversation(
   // The attach replay would not recover it — it carries control state only, and an in-flight
   // reply's chunks are not in `history.read` yet either, so the turn would render truncated.
   // Announcing before the seed read is also what keeps a re-broadcast ask: it lands inside the
-  // seed's `uptoSeq` cut, which only drops what the transcript verifiably covers (CODE-35).
+  // seed's cut, which only drops what the read verifiably covers (CODE-35).
   useEffect(() => {
     if (!sessionId) return;
     client.attachSession(sessionId);
@@ -39,34 +44,19 @@ export function useSeededConversation(
 
   useEffect(
     (signal) => {
-      if (!agentKind || !historyId || !sessionId) return;
+      if (!agentKind || cwd === undefined || !sessionId) return;
       void (async () => {
-        const events: ConversationSeedEvent[] = [];
-        let cursor: string | undefined;
-        for (let page = 0; page < MAX_SEED_PAGES; page += 1) {
-          // eslint-disable-next-line no-await-in-loop -- cursor pagination: each page's cursor comes from the previous reply
-          const result = await client.readHistory(agentKind, {
-            historyId,
-            cwd,
-            cursor,
-            forceRefresh: page === 0,
-          });
-          for (let i = 0, len = result.events.length; i < len; i++) {
-            const entry = result.events[i];
-            events.push({ event: entry.event, ts: entry.ts });
-          }
-          cursor = result.cursor;
-          if (cursor === undefined) break;
-        }
-        if (signal.aborted) return;
-        setSeeded({
-          for: sessionId,
-          seed: { events, uptoSeq: client.eventSeq(sessionId) },
-        });
+        const seed = await readConversationSeed(client, { sessionId, agentKind, cwd, historyId });
+        if (seed === undefined || signal.aborted) return;
+        setSeeded({ for: sessionId, seed });
       })().catch(noop);
     },
-    [agentKind, client, cwd, historyId, sessionId],
+    [agentKind, client, cwd, historyId, sessionId, readGeneration],
   );
 
-  return useConversation(sessionId, seeded?.for === sessionId ? seeded.seed : undefined);
+  return useConversation(
+    sessionId,
+    seeded?.for === sessionId ? seeded.seed : undefined,
+    requestReread,
+  );
 }

packages/client/core/AGENTS.md

@@ -0,0 +1,45 @@
+# packages/client/core — the cross-platform data-plane client
+
+`LinkCodeClient` (session semantics over any `Transport`), the per-session `EventBuffer`, the
+conversation view-model builder, and the conversation store shared by desktop, webview, and mobile.
+Framework-agnostic except `react.tsx` (the `useSyncExternalStore` hooks).
+
+## Conversation seeding — two merge paths
+
+`createConversationStore` folds a seed, then the live `agent.event` buffer. Which seed a surface
+reads is decided once, in `readConversationSeed` (`conversation-read.ts`), and both paths must keep
+working until the compatibility floor passes v80:
+
+- **Projection path** (`ConversationProjectionSeed`): the host advertises the turn graph
+  (`client.supportsConversationGraph`, wire ≥ `CONVERSATION_GRAPH_WIRE_VERSION`) **and** the session
+  has turn rows (`conversation.read` names a `leafTurnId`). Pages are walked as one snapshot; a
+  `graphRevision`/leaf change between pages or the daemon's typed `conflict` restarts the walk.
+  Merge rule: drop live events at or below the final page's `(epoch, seq)` watermark, older epochs
+  included; fold everything above. Nothing is matched by content — the daemon mints one identity per
+  user row (`userRowMessageId`, `msg-<turnId>`) for the live echo and the read alike.
+- **History path** (`ConversationSeed`): ≤v79 hosts and sessions without turn rows (pre-existing
+  sessions until the single-lineage migration). `history.read` transcript + the connection's receive
+  cut (`uptoSeq`); user rows are matched by content because provider and host ids never converge.
+  Retires with the floor bump.
+
+Rules the projection store enforces — keep them when touching it:
+
+- **Interactive events are never dropped on the watermark** (`permission-*`, `question-*`,
+  `prompt-response-status`): their state lives in the daemon's interaction registry, a read may
+  predate them, and the builder folds repeats idempotently. Unstamped frames (the dev mock, an old
+  host) always fold.
+- **A skip in the daemon position asks for one re-read** through `onResync`: an epoch jump (a
+  relaunch, a daemon restart), a same-epoch sequence gap (frames missed while detached), or a
+  `conversation.graph.changed` revision past the read whose new leaf's row never arrived live (an
+  edit or rewrite from any device, a stale read). The request fires at most once per store, via a
+  microtask so it never runs inside a render; folding continues meanwhile. A persisted seed carries
+  no watermark, supersedes nothing, and takes its baseline from the first stamped event.
+- **A stamped repeat stays in the `EventBuffer`** (attach replays resolved asks): dropping it would
+  read as a gap. Only unstamped repeats are deduped.
+- **The echo's attachment blocks win** for the row sharing its identity: durable prompts are
+  text-only until attachment refs land, so the store overlays the buffered echo's content on a read
+  row with the same id. Remove this with the attachment store.
+- Live user echoes carry no envelope `turnId` (they precede turn tracking); never bucket by it.
+
+`history-unavailable` read items become `ConversationItem`s of that kind under the current turn:
+the prompt-only fallback for a lost, compacted, or never-recorded transcript.

packages/client/core/CLAUDE.md

@@ -0,0 +1 @@
+@AGENTS.md

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

@@ -25,6 +25,19 @@ describe('EventBuffer', () => {
     expect(listener).toHaveBeenCalledOnce();
   });
 
+  it('retains a stamped repeat so the daemon sequence stays contiguous', () => {
+    const buffer = new EventBuffer();
+
+    buffer.ingest(SESSION_ID, RESOLUTION, { epoch: 1, seq: 1 });
+    buffer.ingest(SESSION_ID, RESOLUTION, { epoch: 1, seq: 2 });
+
+    // Each stamped frame is a distinct daemon position; dropping one would read as a gap.
+    expect(buffer.snapshot(SESSION_ID).map(({ position }) => position)).toEqual([
+      { epoch: 1, seq: 1 },
+      { epoch: 1, seq: 2 },
+    ]);
+  });
+
   it('drops the buffered suffix at a conversation rewind without dropping subscribers', () => {
     const buffer = new EventBuffer();
     const listener = vi.fn();