feat(agent-adapter): per-turn provider checkpoints and the forkAfterTurn capability

apps/daemon/src/__tests__/conversation-store.test.ts

@@ -209,7 +209,7 @@ describe('SQLite conversation store', () => {
     ).toEqual(prompt('p-1'));
   });
 
-  it('round-trips bindings and re-captures by (turn, history)', async () => {
+  it('round-trips bindings: replay rows re-capture and yield to a live one, which nothing overwrites', async () => {
     const { database } = await databaseWithSessions('s-1');
     const store = createConversationStore(database.client);
     await seedIntent(store);
@@ -220,18 +220,26 @@ describe('SQLite conversation store', () => {
       checkpoint: '{"uuid":"a"}',
       capturedFrom: 'live',
     });
+    // A cold read can land first; the live capture that follows replaces it.
+    await store.saveBinding({ ...live, checkpoint: '{"uuid":"early"}', capturedFrom: 'replay' });
     await store.saveBinding(live);
-    await store.saveBinding({ ...live, historyId: 'native-2', capturedFrom: 'replay' });
-    const recaptured = ProviderTurnBindingSchema.parse({
+    // Neither a later live capture nor a cold-read replay may move the first live cut.
+    await store.saveBinding(
+      ProviderTurnBindingSchema.parse({ ...live, runId: 'run-9', checkpoint: '{"uuid":"b"}' }),
+    );
+    await store.saveBinding({ ...live, checkpoint: '{"uuid":"c"}', capturedFrom: 'replay' });
+    const replay = ProviderTurnBindingSchema.parse({
       ...live,
-      runId: 'run-9',
-      checkpoint: '{"uuid":"b"}',
+      historyId: 'native-2',
+      capturedFrom: 'replay',
     });
+    await store.saveBinding(replay);
+    const recaptured = { ...replay, checkpoint: '{"uuid":"d"}' };
     await store.saveBinding(recaptured);
 
     expect(
       await createConversationStore(database.client).listBindings(TurnIdSchema.parse('t-prompted')),
-    ).toEqual([recaptured, { ...live, historyId: 'native-2', capturedFrom: 'replay' }]);
+    ).toEqual([live, recaptured]);
   });
 
   it('round-trips operations through every state', async () => {

apps/daemon/src/conversation-store.ts

@@ -16,7 +16,7 @@ import {
   PromptRecordSchema,
   ProviderTurnBindingSchema,
 } from '@linkcode/schema';
-import { and, asc, count, eq, inArray, isNotNull, isNull, notInArray } from 'drizzle-orm';
+import { and, asc, count, eq, inArray, isNotNull, isNull, ne, notInArray } from 'drizzle-orm';
 import type { DaemonDatabaseClient } from './db/database';
 import {
   conversationOperations,
@@ -83,6 +83,8 @@ export function createConversationStore(db: DaemonDatabaseClient): ConversationS
         .onConflictDoUpdate({
           target: [providerTurnBindings.turnId, providerTurnBindings.historyId],
           set: binding,
+          // The first live capture stands: a later live or replay write cannot move the cut.
+          setWhere: ne(providerTurnBindings.capturedFrom, 'live'),
         })
         .run();
       return Promise.resolve();

packages/foundation/schema/src/model/history.ts

@@ -10,7 +10,12 @@ export const AgentHistoryCapabilitiesSchema = z.object({
   read: z.boolean(),
   /** Adapter can resume a live session from a known provider-local history id. */
   resume: z.boolean(),
-  /** Adapter can fork provider history before a historical user prompt for replacement. */
+  /** Adapter can fork provider history right after a turn's captured checkpoint — the one fork
+   * primitive every provider actually has ("before prompt T" ≡ "after parent(T)"). Per-turn
+   * availability additionally depends on a captured, still-valid checkpoint. */
+  forkAfterTurn: z.boolean().optional(),
+  /** The legacy `history.branch` capability (≤v79 clients); stays true for a harness whose
+   * turn-level cut is still unverified while `forkAfterTurn` is false. Retired at the floor bump. */
   branch: z.boolean().optional(),
 });
 export type AgentHistoryCapabilities = z.infer<typeof AgentHistoryCapabilitiesSchema>;

packages/host/agent-adapter/AGENTS.md

@@ -44,6 +44,7 @@ Every new adapter MUST honor these (`base.ts`); downstream relies on them, they
 - **`teardown()`** (idempotent) sweeps liveness on cancel/stop/abnormal-end: resolves every pending permission ask `{outcome:'cancelled'}` and forces every non-terminal tool to `failed`. A cancelled turn never leaves a stuck tool or a hung permission.
 - **`streamDelta(id, fullText, kind)`** turns a provider's CUMULATIVE per-item text into incremental deltas keyed by item id. opencode reports cumulative and MUST use it; claude/pi/codex emit true incremental deltas and call `emitAssistantText`/`emitThought` directly (codex additionally keeps a per-item length ledger so `item/completed` can backstop deltas the stream dropped). Mixing the two double-renders or drops text.
 - **`freshSegment()`** opens fresh `messageId` AND `thoughtId` cursors; call it at turn start and after EVERY tool call (`buildConversation` buckets `agent-message-chunk` by `messageId`; `message-grouping.test.ts` guards it). A message's `messageId` must stay STABLE across all its deltas (the Pi adapter once minted a new id per delta and broke dedup).
+- **`emitCheckpoint(historyId, branchPoint, turn)`** mints the turn's provider fork checkpoint (an `onCheckpoint` subscriber, never an agent event; the engine persists it as the turn's `provider_turn_bindings` row and `branchHistory` later forks at it). Emit it BEFORE the turn's `stop`/`idle`, and only for a genuinely completed turn; the engine keeps the FIRST live binding per `(turn, history)`, so a second mint for the same turn is ignored. Branch points: claude = the last main-agent assistant frame's `uuid` (a chain-correct inclusive cut on the EXPECTATION — unverified pending CODE-632's live multi-block turn — that the SDK streams one frame per persisted row; NOT equal to the next user row's `parentUuid` when a Stop hook ran, since a `system/stop_hook_summary` row then sits between; both cut validly and nothing may equate them); codex = the completed `turn/completed` id (`thread/fork lastTurnId` is inclusive); pi = `sessionManager.getLeafId()`; opencode has no "after" cut — its `session.fork {messageID}` cuts BEFORE a message, so the first user `message.updated` FIRST SEEN inside each turn is minted as a `preceding` checkpoint (the engine binds it to the parent turn); every user id is recorded on sight (pre-seeded from `session.messages` on resume), so a mid-turn compaction (`CompactionPart` on a later user message), a re-emitted settled prompt, or an idle straggler re-emitted inside a later turn never mints. opencode advertises `branch` (the legacy cold-read fork) but not `forkAfterTurn` until `session.fork` cut inclusivity is verified on a live server. `branchHistory` must throw `HistoryCheckpointInvalidError` for a cut the provider no longer honours (row gone from the raw transcript, JSON-RPC refusal, vanished opencode message, missing pi entry) — the engine maps it to a typed `unsupported`; codex `history_mode: "paginated"` rollouts (CODE-645) are refused there AND mint no replay cursors, so they stay fork-dark end to end.
 - **`onCommand(name, args)` / `onShellCommand(command)`** (CODE-161) back the `command` / `shell-command` AgentInput variants; both default-reject (`` `${kind}: slash/shell commands are not supported` ``). `AGENT_INPUT_CAPABILITIES` is the complete per-kind source of truth, which Base emits as `capabilities-update` at start; draft composers use the same matrix before a live event stream exists. **`emitCommands(commands)`** advertises the slash-command catalog (`available-commands-update`, full-replace). A missing catalog means discovery is still unavailable and host validation owns an early typed command; an emitted empty catalog is authoritative, so a completed-but-failed discovery must publish `[]` instead of leaving validation fail-open. The engine caches and replays both capabilities and the latest catalog on `session.attach`, prevalidates command/shell inputs before echoing them, and broadcasts an `input_rejected` error when dispatch fails. Adapters must NOT re-emit the user's invocation.
 
 ## Slash commands & shell passthrough (CODE-161)

packages/host/agent-adapter/src/__tests__/base.test.ts

@@ -8,6 +8,9 @@ import type {
 } from '@linkcode/schema';
 import { describe, expect, it } from 'vitest';
 import { BaseAgentAdapter } from '../base';
+import type { HistoryCheckpoint } from '../history-branch';
+import { encodeHistoryBranchCursor } from '../history-branch';
+import { asHistoryId } from '../history-util';
 
 /** Minimal concrete adapter that exposes the protected emit/permission surface for testing. */
 class TestAdapter extends BaseAgentAdapter {
@@ -44,6 +47,9 @@ class TestAdapter extends BaseAgentAdapter {
   title(value: string): void {
     this.emitTitle(value);
   }
+  checkpoint(branchPoint: string, turn?: HistoryCheckpoint['turn']): void {
+    this.emitCheckpoint(asHistoryId('hist-1'), branchPoint, turn);
+  }
   askQuestion(signal?: AbortSignal): Promise<unknown> {
     return this.requestQuestion(
       { toolCallId: 't1' },
@@ -319,6 +325,57 @@ describe('BaseAgentAdapter command/shell defaults', () => {
   });
 });
 
+describe('BaseAgentAdapter fork checkpoints', () => {
+  it('hands checkpoints to onCheckpoint subscribers as branch cursors, never as agent events', () => {
+    const a = new TestAdapter();
+    const seen: HistoryCheckpoint[] = [];
+    a.onCheckpoint((checkpoint) => seen.push(checkpoint));
+    a.checkpoint('entry-9');
+    a.checkpoint('msg-next', 'preceding');
+
+    expect(seen).toEqual([
+      {
+        historyId: 'hist-1',
+        cursor: encodeHistoryBranchCursor('pi', asHistoryId('hist-1'), 'entry-9'),
+        turn: 'ending',
+      },
+      {
+        historyId: 'hist-1',
+        cursor: encodeHistoryBranchCursor('pi', asHistoryId('hist-1'), 'msg-next'),
+        turn: 'preceding',
+      },
+    ]);
+    expect(a.seen).toEqual([]);
+  });
+
+  it('drops checkpoint subscribers on stop', async () => {
+    const a = new TestAdapter();
+    const seen: HistoryCheckpoint[] = [];
+    a.onCheckpoint((checkpoint) => seen.push(checkpoint));
+    await a.stop();
+    a.checkpoint('entry-9');
+    expect(seen).toEqual([]);
+  });
+});
+
+describe('BaseAgentAdapter turn contract', () => {
+  // The engine's dispatch rescue commits a turn whose send() outlives the timer as long as the
+  // session is visibly `running`, so that status must only ever come from a turn-starting input.
+  it('emits no status for control inputs — running is reserved for genuine turn execution', async () => {
+    const a = new TestAdapter();
+    const controls = [
+      { type: 'set-mode', modeId: 'plan' },
+      { type: 'set-approval-policy', policyId: 'default' },
+      { type: 'set-model', model: 'test/model' },
+      { type: 'set-effort', effort: 'high' },
+      { type: 'permission-response', requestId: 'unknown', outcome: { outcome: 'cancelled' } },
+      { type: 'question-response', requestId: 'unknown', outcome: { outcome: 'cancelled' } },
+    ] as const;
+    await Promise.allSettled(controls.map((input) => a.send(input)));
+    expect(a.seen.filter((event) => event.type === 'status')).toEqual([]);
+  });
+});
+
 describe('BaseAgentAdapter initial effort', () => {
   it('validates and applies initial effort before starting the provider', async () => {
     const a = new EffortTestAdapter();

packages/host/agent-adapter/src/__tests__/claude-code-checkpoint.test.ts

@@ -0,0 +1,222 @@
+import type { SDKMessage } from '@anthropic-ai/claude-agent-sdk';
+import type { AgentEvent, StartOptions } from '@linkcode/schema';
+import { describe, expect, it, vi } from 'vitest';
+import type { HistoryCheckpoint } from '../history-branch';
+import { encodeHistoryBranchCursor, HistoryCheckpointInvalidError } from '../history-branch';
+import { asHistoryId } from '../history-util';
+import type { ClaudeTranscriptSupplement } from '../native/claude-code';
+import { buildClaudeTranscriptSupplement, ClaudeCodeAdapter } from '../native/claude-code';
+
+/**
+ * Fork checkpoints: the cut claude's `forkSession` needs is the uuid of the row the
+ * next user row hangs off — the turn's last main-agent assistant frame on a linear history — and
+ * a fork must re-verify that row still exists in the raw transcript before cutting.
+ */
+
+const SESSION = 'sid-fork';
+
+function forkedChild() {
+  return Promise.resolve({ sessionId: 'sid-child' });
+}
+
+class TestClaude extends ClaudeCodeAdapter {
+  forkSession = vi.fn(forkedChild);
+  supplementUuids: string[] = [];
+  started: StartOptions[] = [];
+
+  feed(value: object): void {
+    this.handleMessage(value as SDKMessage);
+  }
+
+  protected override loadSdk<T>(): Promise<T> {
+    return Promise.resolve({ forkSession: this.forkSession } as T);
+  }
+
+  protected override readTranscriptSupplement(): Promise<ClaudeTranscriptSupplement> {
+    return Promise.resolve({
+      records: new Map(),
+      droppedRows: [],
+      parentUuidByUuid: new Map(this.supplementUuids.map((uuid) => [uuid, null])),
+      toolUses: new Map(),
+      toolUseResults: new Map(),
+      toolUsePatches: new Map(),
+    });
+  }
+
+  protected override onStart(opts: StartOptions): Promise<void> {
+    this.started.push(opts);
+    return Promise.resolve();
+  }
+}
+
+function assistantFrame(
+  uuid: string,
+  parentToolUseId: string | null = null,
+  apiMessageId = `api-${uuid}`,
+): object {
+  return {
+    type: 'assistant',
+    session_id: SESSION,
+    uuid,
+    parent_tool_use_id: parentToolUseId,
+    message: {
+      id: apiMessageId,
+      model: 'claude-test',
+      content: [{ type: 'text', text: 'hi' }],
+    },
+  };
+}
+
+function resultFrame(subtype: 'success' | 'error_during_execution'): object {
+  return {
+    type: 'result',
+    subtype,
+    session_id: SESSION,
+    uuid: `result-${subtype}`,
+    stop_reason: 'end_turn',
+    usage: {},
+    total_cost_usd: 0,
+    ...(subtype !== 'success' && { errors: ['boom'] }),
+  };
+}
+
+function harness() {
+  const adapter = new TestClaude();
+  const events: AgentEvent[] = [];
+  const checkpoints: HistoryCheckpoint[] = [];
+  adapter.onEvent((event) => events.push(event));
+  adapter.onCheckpoint((checkpoint) => {
+    checkpoints.push(checkpoint);
+    events.push({ type: 'title-update', title: 'checkpoint-marker' });
+  });
+  return { adapter, events, checkpoints };
+}
+
+describe('ClaudeCodeAdapter live fork checkpoints', () => {
+  it('mints the last main-agent assistant uuid at a successful result, before the stop', () => {
+    const { adapter, events, checkpoints } = harness();
+
+    adapter.feed(assistantFrame('row-a'));
+    adapter.feed(assistantFrame('row-b'));
+    // A subagent frame is not a main-chain row: the next user row never hangs off it.
+    adapter.feed(assistantFrame('row-sub', 'toolu_agent'));
+    adapter.feed(resultFrame('success'));
+
+    expect(checkpoints).toEqual([
+      {
+        historyId: SESSION,
+        cursor: encodeHistoryBranchCursor('claude-code', asHistoryId(SESSION), 'row-b'),
+        turn: 'ending',
+      },
+    ]);
+    const marker = events.findIndex(
+      (event) => event.type === 'title-update' && event.title === 'checkpoint-marker',
+    );
+    const stop = events.findIndex((event) => event.type === 'stop');
+    expect(marker).toBeGreaterThanOrEqual(0);
+    expect(stop).toBeGreaterThan(marker);
+  });
+
+  it('mints the LAST frame of a multi-block API message — one frame per persisted row', () => {
+    const { adapter, checkpoints } = harness();
+
+    // The CLI persists one transcript row per content block, each with its own uuid, all sharing
+    // the API `message.id`; the next user row hangs off the last of them.
+    adapter.feed(assistantFrame('row-b1', null, 'api-multi'));
+    adapter.feed(assistantFrame('row-b2', null, 'api-multi'));
+    adapter.feed(resultFrame('success'));
+
+    expect(checkpoints.map((checkpoint) => JSON.parse(checkpoint.cursor).branchPoint)).toEqual([
+      'row-b2',
+    ]);
+  });
+
+  it('mints nothing for a failed result', () => {
+    const { adapter, checkpoints } = harness();
+
+    adapter.feed(assistantFrame('row-a'));
+    adapter.feed(resultFrame('error_during_execution'));
+
+    expect(checkpoints).toEqual([]);
+  });
+});
+
+describe('claude fork cuts across a Stop hook summary row', () => {
+  const start: StartOptions = { kind: 'claude-code', cwd: '/repo' };
+  const row = (value: object) => JSON.stringify(value);
+  const transcript = [
+    row({ type: 'user', uuid: 'u0', parentUuid: null, message: { role: 'user', content: 'q' } }),
+    row({ type: 'assistant', uuid: 'row-a', parentUuid: 'u0', message: { role: 'assistant' } }),
+    row({ type: 'system', subtype: 'stop_hook_summary', uuid: 'row-s', parentUuid: 'row-a' }),
+    row({
+      type: 'user',
+      uuid: 'u1',
+      parentUuid: 'row-s',
+      message: { role: 'user', content: 'q2' },
+    }),
+  ];
+
+  it('the cold-read cursor is the system row, the live checkpoint the assistant row — both cut', async () => {
+    const supplement = buildClaudeTranscriptSupplement(transcript);
+    expect(supplement.parentUuidByUuid.get('u1')).toBe('row-s');
+
+    const cuts = ['row-s', 'row-a'];
+    for (let i = 0, len = cuts.length; i < len; i++) {
+      const cut = cuts[i];
+      const adapter = new TestClaude();
+      adapter.supplementUuids = [...supplement.parentUuidByUuid.keys()];
+      // eslint-disable-next-line no-await-in-loop -- one fork per cut, sequential by construction
+      await adapter.branchHistory(
+        {
+          historyId: asHistoryId(SESSION),
+          cursor: encodeHistoryBranchCursor('claude-code', asHistoryId(SESSION), cut),
+        },
+        start,
+      );
+      expect(adapter.forkSession).toHaveBeenCalledWith(SESSION, {
+        upToMessageId: cut,
+        dir: '/repo',
+      });
+    }
+  });
+});
+
+describe('ClaudeCodeAdapter.branchHistory checkpoint validity', () => {
+  const start: StartOptions = { kind: 'claude-code', cwd: '/repo' };
+
+  it('forks through the checkpoint row when the transcript still has it', async () => {
+    const adapter = new TestClaude();
+    adapter.supplementUuids = ['row-a', 'row-b'];
+
+    await adapter.branchHistory(
+      {
+        historyId: asHistoryId(SESSION),
+        cursor: encodeHistoryBranchCursor('claude-code', asHistoryId(SESSION), 'row-b'),
+      },
+      start,
+    );
+
+    expect(adapter.forkSession).toHaveBeenCalledWith(SESSION, {
+      upToMessageId: 'row-b',
+      dir: '/repo',
+    });
+    expect(adapter.started).toEqual([start]);
+  });
+
+  it('refuses typed, without forking or starting, when the row is gone (rewritten or deleted transcript)', async () => {
+    const adapter = new TestClaude();
+    adapter.supplementUuids = ['row-a'];
+
+    await expect(
+      adapter.branchHistory(
+        {
+          historyId: asHistoryId(SESSION),
+          cursor: encodeHistoryBranchCursor('claude-code', asHistoryId(SESSION), 'row-b'),
+        },
+        start,
+      ),
+    ).rejects.toBeInstanceOf(HistoryCheckpointInvalidError);
+    expect(adapter.forkSession).not.toHaveBeenCalled();
+    expect(adapter.started).toEqual([]);
+  });
+});

packages/host/agent-adapter/src/__tests__/claude-code-compaction.test.ts

@@ -159,6 +159,9 @@ describe('buildClaudeTranscriptSupplement', () => {
       convRow('user', 'u1'),
     ]);
     expect(supplement.droppedRows.map((r) => r.uuid)).toEqual(['u0', 'a0']);
+    // A sidechain row can never be a fork cut (forkSession drops it); meta rows stay chain rows.
+    expect(supplement.parentUuidByUuid.has('side0')).toBe(false);
+    expect(supplement.parentUuidByUuid.has('meta0')).toBe(true);
     expect(supplement.droppedRows[0]).toMatchObject({
       type: 'user',
       session_id: 'sid-1',