diff --git a/packages/agent-harness/src/bridge.ts b/packages/agent-harness/src/bridge.ts
new file mode 100644
index 0000000000..614d7acdc2
--- /dev/null
+++ b/packages/agent-harness/src/bridge.ts
@@ -0,0 +1,28 @@
+import { z } from 'zod';
+import type { ExecutionIntent, JournalScope } from './journal';
+
+export const BridgeReadinessSchema = z.strictObject({
+  available: z.boolean(),
+  foreground: z.boolean(),
+  connectivity: z.enum(['confirmed', 'offline', 'unknown']),
+  unlock: z.enum(['ready', 'locked', 'unknown']),
+  gesture: z.enum(['not_required', 'required', 'satisfied']),
+});
+export type BridgeReadiness = z.infer<typeof BridgeReadinessSchema>;
+export function bridgeWaitReason(input: unknown) {
+  const readiness = BridgeReadinessSchema.parse(input);
+  if (!readiness.available) return 'unavailable';
+  if (!readiness.foreground) return 'background';
+  if (readiness.connectivity !== 'confirmed') return 'offline';
+  if (readiness.unlock !== 'ready') return 'locked';
+  if (readiness.gesture === 'required') return 'gesture';
+  return null;
+}
+export type ClientBridge = {
+  readiness: (scope: JournalScope, execution: ExecutionIntent) => BridgeReadiness;
+  // Recheck account, grant, and every readiness gate at the actual effect boundary. Return a
+  // ToolOutcome only for a known result; throw on uncertainty. Never retry an effect internally.
+  execute: (scope: JournalScope, execution: ExecutionIntent) => Promise<unknown>;
+  // Read evidence only. null means unknown, NOT permission to execute or transfer the grant.
+  reconcileReceipt: (scope: JournalScope, execution: ExecutionIntent) => Promise<unknown>;
+};
diff --git a/packages/agent-harness/src/client.test.ts b/packages/agent-harness/src/client.test.ts
new file mode 100644
index 0000000000..bc7d8411fb
--- /dev/null
+++ b/packages/agent-harness/src/client.test.ts
@@ -0,0 +1,970 @@
+import { createHash } from 'node:crypto';
+import { expect, it } from 'vitest';
+import { createHarnessClient, type ClientResult } from './client';
+import { canonicalizeValidatedInput, type Command } from './commands';
+import type { BridgeReadiness, ClientBridge } from './bridge';
+import {
+  CommandReplySchema,
+  JournalSnapshotSchema,
+  type CommandReply,
+  type ExecutionRequest,
+  type HarnessJournal,
+  type JournalScope,
+  type JournalSnapshot,
+} from './journal';
+import type { ToolOutcome } from './contracts';
+
+const id = (n: number) => `00000000-0000-4000-8000-${String(n).padStart(12, '0')}`;
+const scope: JournalScope = { ownerUserId: 'owner', clientId: id(1), storageGeneration: id(2) };
+const command: Extract<Command, { type: 'sendMessage' }> = {
+  type: 'sendMessage',
+  protocolVersion: 1,
+  clientId: scope.clientId,
+  commandId: id(3),
+  conversationId: id(4),
+  text: 'Keep this draft',
+  modelId: 'test/model',
+  permissionRevision: 0,
+};
+const digest = (input: string) => createHash('sha256').update(input).digest('hex');
+const request: ExecutionRequest = {
+  toolCall: {
+    id: id(5),
+    runId: id(6),
+    name: 'app.openScreen',
+    definitionVersion: '1',
+    arguments: { screen: 'preferences' },
+    context: { type: 'personal' },
+    effect: 'side_effect',
+    executionTarget: { kind: 'client', clientId: scope.clientId },
+    approval: null,
+    state: 'executing',
+    result: null,
+  },
+  grant: {
+    id: id(7),
+    conversationId: command.conversationId,
+    ownerUserId: scope.ownerUserId,
+    clientId: scope.clientId,
+    toolCallId: id(5),
+    context: { type: 'personal' },
+    definitionVersion: '1',
+    inputDigest: digest(canonicalizeValidatedInput({ screen: 'preferences' })),
+    generation: 1,
+    expiresAt: '2026-08-29T00:00:00.000Z',
+  },
+  completionCommandId: id(8),
+};
+const unrelatedRequest: ExecutionRequest = {
+  ...request,
+  completionCommandId: id(21),
+  toolCall: { ...request.toolCall, id: id(20), runId: id(24) },
+  grant: { ...request.grant, id: id(22), toolCallId: id(20), conversationId: id(23) },
+};
+const receipt: ToolOutcome = { status: 'succeeded', output: { screen: 'preferences' } };
+const ready: BridgeReadiness = {
+  available: true,
+  foreground: true,
+  connectivity: 'confirmed',
+  unlock: 'ready',
+  gesture: 'not_required',
+};
+type Boundary =
+  | 'read'
+  | 'intent'
+  | 'ack'
+  | 'delete'
+  | 'execution'
+  | 'execution_wait'
+  | 'receipt'
+  | 'send'
+  | 'effect'
+  | 'reconcile';
+type Side = 'before' | 'after';
+const sides: Side[] = ['before', 'after'];
+function fixture() {
+  const f = {
+    durable: {
+      scope,
+      revision: 0,
+      intents: [],
+      acknowledgments: [],
+      executions: [],
+    } as JournalSnapshot,
+    active: scope as JournalScope | null,
+    readiness: { ...ready },
+    time: Date.parse('2026-08-28T00:00:00.000Z'),
+    fault: undefined as { point: Boundary; side: Side } | undefined,
+    hook: (_point: Boundary, _side: Side) => {},
+    readOverride: undefined as (() => unknown) | undefined,
+    reply: undefined as ((input: Command) => unknown) | undefined,
+    evidence: (_execution: ExecutionRequest): unknown => null,
+    effectResult: receipt as unknown,
+    effects: [] as string[],
+    attempts: [] as Command[],
+    accepted: new Map<string, { canonical: string; reply: CommandReply }>(),
+    reported: new Map<string, ToolOutcome>(),
+    trace: [] as string[],
+  };
+  function edge(point: Boundary, side: Side) {
+    f.trace.push(`${point}:${side}`);
+    f.hook(point, side);
+    if (f.fault?.point === point && f.fault.side === side) {
+      f.fault = undefined;
+      throw new Error(`Killed ${side} ${point}`);
+    }
+  }
+  async function boundary<T>(point: Boundary, action: () => T): Promise<T> {
+    edge(point, 'before');
+    const value = action();
+    edge(point, 'after');
+    return value;
+  }
+  const journal: HarnessJournal = {
+    read: async () =>
+      boundary('read', () => (f.readOverride ? f.readOverride() : structuredClone(f.durable))),
+    compareAndSwap: async (bound, expected, next) => {
+      const point: Boundary =
+        next.executions.length > f.durable.executions.length
+          ? 'execution'
+          : next.executions.length < f.durable.executions.length
+            ? 'execution_wait'
+            : next.acknowledgments.length > f.durable.acknowledgments.length
+              ? 'ack'
+              : next.intents.length > f.durable.intents.length
+                ? 'intent'
+                : next.intents.length < f.durable.intents.length
+                  ? 'delete'
+                  : 'receipt';
+      return boundary(point, () => {
+        if (canonicalizeValidatedInput(bound) !== canonicalizeValidatedInput(f.durable.scope))
+          throw new Error('Scope changed');
+        if (expected !== f.durable.revision) return false;
+        f.durable = JournalSnapshotSchema.parse(structuredClone(next));
+        return true;
+      });
+    },
+  };
+  const bridge: ClientBridge = {
+    readiness: () => f.readiness,
+    execute: async (bound, execution) =>
+      boundary('effect', () => {
+        if (canonicalizeValidatedInput(f.active) !== canonicalizeValidatedInput(bound))
+          throw new Error('Account changed');
+        if (
+          !f.durable.executions.some(
+            item => item.grant.id === execution.grant.id && item.receipt === null
+          )
+        )
+          throw new Error('Effect without durable intent');
+        f.effects.push(execution.toolCall.id);
+        return f.effectResult;
+      }),
+    reconcileReceipt: async (_bound, execution) =>
+      boundary('reconcile', () => f.evidence(execution)),
+  };
+  const open = () =>
+    createHarnessClient({
+      scope,
+      currentScope: () => f.active,
+      journal,
+      bridge,
+      now: () => f.time,
+      digest,
+      transport: {
+        send: async (bound, input) =>
+          boundary('send', () => {
+            if (!f.durable.intents.some(item => item.command.commandId === input.commandId))
+              throw new Error('Send without durable intent');
+            if (
+              input.type === 'completeClientTool' &&
+              !f.durable.executions.some(
+                item => item.completionCommandId === input.commandId && item.receipt !== null
+              )
+            )
+              throw new Error('Completion without durable receipt');
+            f.attempts.push(structuredClone(input));
+            const key = `${bound.ownerUserId}:${bound.clientId}:${input.commandId}`;
+            const canonical = canonicalizeValidatedInput(input);
+            const prior = f.accepted.get(key);
+            if (prior) {
+              if (prior.canonical !== canonical)
+                return {
+                  status: 'rejected',
+                  commandId: input.commandId,
+                  error: { code: 'command_conflict', message: 'Conflict', retryable: false },
+                };
+              return prior.reply;
+            }
+            const reply = f.reply
+              ? f.reply(input)
+              : {
+                  status: 'accepted',
+                  commandId: input.commandId,
+                  result: { position: f.accepted.size + 1 },
+                };
+            const parsed = parseReply(reply);
+            if (parsed) {
+              f.accepted.set(key, { canonical, reply: parsed });
+              if (input.type === 'completeClientTool' && parsed.status === 'accepted')
+                f.reported.set(input.toolCallId, input.result);
+            }
+            return reply;
+          }),
+      },
+    });
+  return Object.assign(f, { open, journal });
+}
+// Invalid transport fixtures must reach the client unchanged, not fail inside the adapter.
+function parseReply(input: unknown) {
+  const parsed = CommandReplySchema.safeParse(input);
+  return parsed.success ? parsed.data : undefined;
+}
+const errorCode = (result: ClientResult) => ('error' in result ? result.error.code : undefined);
+
+it.each(
+  (['read', 'intent', 'send', 'ack', 'delete'] as const).flatMap(point =>
+    sides.map(side => ({ point, side }))
+  )
+)(
+  'relaunches across $side $point with one accepted command and its original result',
+  async ({ point, side }) => {
+    const f = fixture();
+    f.fault = { point, side };
+    const first = await f.open().submit(command);
+    expect(first.status).toBe(point === 'intent' ? 'unsent' : 'unknown');
+    const relaunched = f.open();
+    await relaunched.recover();
+    const result = await relaunched.submit(command);
+    expect(result).toMatchObject({ status: 'accepted', command, result: { position: 1 } });
+    expect(f.accepted.size).toBe(1);
+    expect(f.durable.intents).toEqual([]);
+    expect(f.durable.acknowledgments[0].intent.command).toEqual(command);
+    expect(
+      f.attempts.every(
+        input => canonicalizeValidatedInput(input) === canonicalizeValidatedInput(command)
+      )
+    ).toBe(true);
+    if (point === 'delete' || (point === 'ack' && side === 'after'))
+      expect(f.attempts).toHaveLength(1);
+  }
+);
+it('freezes the submitted input before queued work and ignores caller mutations', async () => {
+  const f = fixture();
+  const input = { ...command };
+  const sending = f.open().submit(input);
+  input.text = 'Changed after Send';
+  await sending;
+  expect(f.attempts).toEqual([command]);
+  expect(f.durable.acknowledgments[0].intent.command).toEqual(command);
+});
+it.each(['pending', 'acknowledged'] as const)(
+  'rejects changed input for a %s command ID',
+  async state => {
+    const f = fixture();
+    if (state === 'pending') f.fault = { point: 'send', side: 'after' };
+    await f.open().submit(command);
+    const result = await f.open().submit({ ...command, text: 'Different input' });
+    expect(errorCode(result)).toBe('command_conflict');
+    expect(f.accepted.size).toBe(1);
+    expect(f.attempts).toEqual([command]);
+  }
+);
+it('latches failed storage until a verified recovery read, without new commands or effects', async () => {
+  const f = fixture(),
+    client = f.open();
+  f.fault = { point: 'intent', side: 'before' };
+  expect(await client.submit(command)).toMatchObject({
+    status: 'unsent',
+    error: { code: 'storage_unavailable' },
+  });
+  expect(errorCode(await client.submit({ ...command, commandId: id(30) }))).toBe(
+    'storage_unavailable'
+  );
+  expect(errorCode(await client.dispatch(request))).toBe('storage_unavailable');
+  expect(f.accepted.size).toBe(0);
+  expect(f.effects).toEqual([]);
+  expect(await client.recover()).toEqual([]);
+  expect((await client.submit(command)).status).toBe('accepted');
+});
+it('retains a stale draft after relaunch and requires explicit review for a changed send', async () => {
+  const f = fixture();
+  f.reply = input => ({
+    status: 'rejected',
+    commandId: input.commandId,
+    error: { code: 'stale_revision', message: 'Review the new mode', retryable: true },
+  });
+  await f.open().submit(command);
+  f.reply = undefined;
+  const client = f.open();
+  expect(await client.recover()).toMatchObject([
+    { status: 'rejected', command, error: { code: 'stale_revision' } },
+  ]);
+  const replacement = { ...command, commandId: id(30), permissionRevision: 1 };
+  expect(await client.submit(replacement)).toMatchObject({
+    status: 'rejected',
+    command,
+    error: { code: 'stale_revision' },
+  });
+  expect(f.accepted.size).toBe(1);
+  expect((await client.submit(replacement, { reviewedCommandId: command.commandId })).status).toBe(
+    'accepted'
+  );
+  expect(f.attempts).toEqual([command, replacement]);
+  expect(f.durable.acknowledgments[0].intent.command).toEqual(command);
+  expect(await f.open().recover()).toEqual([]);
+});
+it.each(['access_revoked', 'retired', 'unsupported_protocol'] as const)(
+  'preserves a non-retryable %s rejection without replay',
+  async code => {
+    const f = fixture();
+    f.reply = input => ({
+      status: 'rejected',
+      commandId: input.commandId,
+      error: { code, message: 'Blocked', retryable: false },
+    });
+    expect(await f.open().submit(command)).toMatchObject({
+      status: 'rejected',
+      error: { code, retryable: false },
+    });
+    await f.open().recover();
+    await f.open().submit(command);
+    expect(f.attempts).toHaveLength(1);
+  }
+);
+it.each(['access_revoked', 'retired'] as const)(
+  'retains a %s completion without blocking a distinct authorized action',
+  async code => {
+    const f = fixture();
+    f.reply = input => ({
+      status: 'rejected',
+      commandId: input.commandId,
+      error: { code, message: 'Blocked', retryable: false },
+    });
+    const rejection = {
+      status: 'rejected',
+      command: { commandId: request.completionCommandId },
+      error: { code, retryable: false },
+    };
+    expect(await f.open().dispatch(request)).toMatchObject(rejection);
+    const retained = structuredClone(f.durable.executions[0]);
+    f.reply = undefined;
+    const client = f.open();
+    expect(await client.recover()).toMatchObject([rejection]);
+    const completion = {
+      status: 'accepted',
+      command: {
+        commandId: unrelatedRequest.completionCommandId,
+        conversationId: unrelatedRequest.grant.conversationId,
+      },
+    };
+    expect(await client.dispatch(unrelatedRequest)).toMatchObject(completion);
+    expect(await client.dispatch(unrelatedRequest)).toMatchObject(completion);
+    expect(await f.open().dispatch(request)).toMatchObject(rejection);
+    expect(await f.open().recover()).toMatchObject([rejection, completion]);
+    expect(f.effects).toEqual([request.toolCall.id, unrelatedRequest.toolCall.id]);
+    expect([...f.reported]).toEqual([[unrelatedRequest.toolCall.id, receipt]]);
+    expect(f.attempts.map(input => input.commandId)).toEqual([
+      request.completionCommandId,
+      unrelatedRequest.completionCommandId,
+    ]);
+    expect(f.durable.executions[0]).toEqual(retained);
+  }
+);
+it.each(['access_revoked', 'retired'] as const)(
+  'does not bypass a retryable %s completion for an unrelated action',
+  async code => {
+    const f = fixture();
+    f.reply = input => ({
+      status: 'rejected',
+      commandId: input.commandId,
+      error: { code, message: 'Retry', retryable: true },
+    });
+    await f.open().dispatch(request);
+    expect(await f.open().dispatch(unrelatedRequest)).toMatchObject({
+      status: 'rejected',
+      command: { commandId: request.completionCommandId },
+      error: { code, retryable: true },
+    });
+    expect(f.effects).toEqual([request.toolCall.id]);
+    expect(f.reported.size).toBe(0);
+  }
+);
+it.each(sides)(
+  'keeps a crash %s the effect unknown across readiness loss and restoration',
+  async side => {
+    const f = fixture();
+    f.fault = { point: 'effect', side };
+    await f.open().dispatch(request);
+    const client = f.open();
+    const unknown = { status: 'unknown', error: { code: 'outcome_unknown' } };
+    f.readiness = { ...ready, foreground: false };
+    expect(await client.dispatch(request)).toMatchObject(unknown);
+    f.readiness = ready;
+    expect(await client.dispatch(request)).toMatchObject(unknown);
+    expect(await client.dispatch(unrelatedRequest)).toMatchObject(unknown);
+    expect(f.effects).toEqual(side === 'before' ? [] : [request.toolCall.id]);
+    expect(f.reported.size).toBe(0);
+    expect(f.durable.executions).toHaveLength(1);
+  }
+);
+it.each([null, {}, { status: 'accepted', commandId: id(99), result: {} }])(
+  'retains the intent after an invalid acknowledgment: %j',
+  async reply => {
+    const f = fixture();
+    f.reply = () => reply;
+    expect(errorCode(await f.open().submit(command))).toBe('invalid_output');
+    expect(f.durable.intents).toHaveLength(1);
+    expect(f.durable.acknowledgments).toEqual([]);
+  }
+);
+it.each(
+  (['read', 'execution', 'effect', 'receipt', 'intent', 'send', 'ack', 'delete'] as const).flatMap(
+    point => sides.map(side => ({ point, side }))
+  )
+)('never repeats an effect after a crash $side $point', async ({ point, side }) => {
+  const f = fixture();
+  f.fault = { point, side };
+  await f.open().dispatch(request);
+  const hadIntent = f.durable.executions.length > 0;
+  const effects = f.effects.length;
+  const client = f.open();
+  await client.recover();
+  await client.dispatch(request);
+  expect(f.effects).toHaveLength(hadIntent ? effects : 1);
+  expect(f.effects.length).toBeLessThanOrEqual(1);
+  const execution = f.durable.executions[0];
+  if (execution.receipt !== null) {
+    expect(f.reported.get(request.toolCall.id)).toEqual(receipt);
+    expect(f.accepted.size).toBe(1);
+  } else {
+    expect(f.reported.size).toBe(0);
+    expect(await client.dispatch(request)).toMatchObject({
+      status: 'unknown',
+      error: { code: 'outcome_unknown' },
+    });
+  }
+});
+it('rejects completion reports without a committed receipt', async () => {
+  const f = fixture();
+  const result = await f.open().submit({
+    type: 'completeClientTool',
+    protocolVersion: 1,
+    clientId: scope.clientId,
+    commandId: request.completionCommandId,
+    conversationId: request.grant.conversationId,
+    toolCallId: request.toolCall.id,
+    grantId: request.grant.id,
+    generation: request.grant.generation,
+    result: receipt,
+  });
+  expect(errorCode(result)).toBe('invalid_input');
+  expect(f.accepted.size).toBe(0);
+});
+it.each(sides)(
+  'reconciles evidence without executing again after failure %s the evidence read',
+  async side => {
+    const f = fixture();
+    f.fault = { point: 'effect', side: 'after' };
+    await f.open().dispatch(request);
+    f.evidence = () => receipt;
+    f.fault = { point: 'reconcile', side };
+    await f.open().recover();
+    await f.open().recover();
+    expect(f.effects).toEqual([request.toolCall.id]);
+    expect(f.reported.get(request.toolCall.id)).toEqual(receipt);
+  }
+);
+it.each(['recover', 'dispatch'] as const)(
+  'reports stored receipts before evidence queries during %s',
+  async mode => {
+    const f = fixture();
+    const second = {
+      ...request,
+      completionCommandId: id(21),
+      toolCall: { ...request.toolCall, id: id(20) },
+      grant: { ...request.grant, id: id(22), toolCallId: id(20) },
+    };
+    f.durable.executions = [
+      { ...request, receipt: null },
+      { ...second, receipt },
+    ];
+    f.evidence = () => (f.reported.has(second.toolCall.id) ? receipt : null);
+    if (mode === 'recover') await f.open().recover();
+    else await f.open().dispatch(request);
+    expect([...f.reported.keys()]).toEqual([second.toolCall.id, request.toolCall.id]);
+    expect(f.effects).toEqual([]);
+  }
+);
+it('does not transfer an uncertain execution to a replacement grant', async () => {
+  const f = fixture();
+  f.fault = { point: 'effect', side: 'after' };
+  await f.open().dispatch(request);
+  const result = await f
+    .open()
+    .dispatch({ ...request, grant: { ...request.grant, id: id(50), generation: 2 } });
+  expect(errorCode(result)).toBe('outcome_unknown');
+  expect(f.effects).toEqual([request.toolCall.id]);
+  expect(f.durable.executions[0].grant).toEqual(request.grant);
+});
+it.each([
+  [{ available: false }, 'unavailable'],
+  [{ foreground: false }, 'background'],
+  [{ connectivity: 'offline' }, 'offline'],
+  [{ connectivity: 'unknown' }, 'offline'],
+  [{ unlock: 'locked' }, 'locked'],
+  [{ unlock: 'unknown' }, 'locked'],
+  [{ gesture: 'required' }, 'gesture'],
+] as const)('waits for bridge readiness %j without admitting an effect', async (change, reason) => {
+  const f = fixture();
+  f.readiness = { ...ready, ...change };
+  expect(await f.open().dispatch(request)).toEqual({ status: 'waiting', reason });
+  expect(f.effects).toEqual([]);
+  expect(f.durable.executions).toEqual([]);
+  f.readiness = ready;
+  expect((await f.open().dispatch(request)).status).toBe('accepted');
+  expect(f.effects).toEqual([request.toolCall.id]);
+});
+it.each(
+  (
+    [
+      [{ available: false }, 'unavailable'],
+      [{ foreground: false }, 'background'],
+      [{ connectivity: 'offline' }, 'offline'],
+      [{ connectivity: 'unknown' }, 'offline'],
+      [{ unlock: 'locked' }, 'locked'],
+      [{ unlock: 'unknown' }, 'locked'],
+      [{ gesture: 'required' }, 'gesture'],
+    ] as const
+  ).flatMap(([change, reason]) =>
+    (['same host', 'relaunch'] as const).map(resume => ({ change, reason, resume }))
+  )
+)(
+  'resumes a $reason wait after the intent commit on $resume without repeating effects',
+  async ({ change, reason, resume }) => {
+    const f = fixture(),
+      client = f.open();
+    f.hook = (point, side) => {
+      if (point === 'execution' && side === 'after') f.readiness = { ...ready, ...change };
+    };
+    expect(await client.dispatch(request)).toEqual({ status: 'waiting', reason });
+    expect(f.effects).toEqual([]);
+    expect(f.reported.size).toBe(0);
+    f.hook = () => {};
+    f.readiness = ready;
+    const resumed = resume === 'same host' ? client : f.open();
+    await resumed.recover();
+    expect(f.effects).toEqual([]);
+    expect(await resumed.dispatch(request)).toMatchObject({
+      status: 'accepted',
+      command: { commandId: request.completionCommandId },
+    });
+    await resumed.dispatch(request);
+    await f.open().recover();
+    expect(f.effects).toEqual([request.toolCall.id]);
+    expect(f.reported.get(request.toolCall.id)).toEqual(receipt);
+    expect(f.attempts).toHaveLength(1);
+  }
+);
+it('refuses a grant that expires during the execution intent commit', async () => {
+  const f = fixture();
+  f.hook = (point, side) => {
+    if (point === 'execution' && side === 'after') f.time = Date.parse(request.grant.expiresAt);
+  };
+  expect(await f.open().dispatch(request)).toMatchObject({
+    status: 'unsent',
+    error: { code: 'access_revoked' },
+  });
+  f.hook = () => {};
+  await f.open().recover();
+  expect(f.effects).toEqual([]);
+  expect(f.reported.size).toBe(0);
+});
+it.each(sides)('requires a durable wait release after a crash %s its commit', async side => {
+  const f = fixture(),
+    client = f.open();
+  f.hook = (point, edge) => {
+    if (point === 'execution' && edge === 'after') f.readiness = { ...ready, foreground: false };
+  };
+  f.fault = { point: 'execution_wait', side };
+  expect(await client.dispatch(request)).toMatchObject({
+    status: 'unsent',
+    error: { code: 'storage_unavailable' },
+  });
+  expect(f.effects).toEqual([]);
+  expect(f.reported.size).toBe(0);
+  f.hook = () => {};
+  f.readiness = ready;
+  expect(errorCode(await client.dispatch(request))).toBe('storage_unavailable');
+  const relaunched = f.open();
+  await relaunched.recover();
+  const result = await relaunched.dispatch(request);
+  if (side === 'before') {
+    expect(result).toMatchObject({ status: 'unknown', error: { code: 'outcome_unknown' } });
+    expect(f.effects).toEqual([]);
+    expect(f.reported.size).toBe(0);
+  } else {
+    expect(result.status).toBe('accepted');
+    expect(f.effects).toEqual([request.toolCall.id]);
+    expect(f.reported.get(request.toolCall.id)).toEqual(receipt);
+    await relaunched.dispatch(request);
+    expect(f.effects).toHaveLength(1);
+    expect(f.attempts).toHaveLength(1);
+  }
+});
+it('retains the execution fence when a concurrent commit prevents the wait release', async () => {
+  const f = fixture();
+  f.hook = (point, side) => {
+    if (point === 'execution' && side === 'after') f.readiness = { ...ready, foreground: false };
+    if (point === 'execution_wait' && side === 'before') f.durable.revision++;
+  };
+  expect(await f.open().dispatch(request)).toMatchObject({
+    status: 'unknown',
+    error: { code: 'storage_unavailable' },
+  });
+  f.hook = () => {};
+  f.readiness = ready;
+  await f.open().recover();
+  expect(await f.open().dispatch(request)).toMatchObject({
+    status: 'unknown',
+    error: { code: 'outcome_unknown' },
+  });
+  expect(f.effects).toEqual([]);
+  expect(f.reported.size).toBe(0);
+  expect(f.durable.executions).toHaveLength(1);
+});
+it('serializes competing hosts with an atomic execution fence', async () => {
+  const f = fixture();
+  const results = await Promise.all([f.open().dispatch(request), f.open().dispatch(request)]);
+  expect(results).toContainEqual(
+    expect.objectContaining({
+      status: 'unknown',
+      error: expect.objectContaining({ code: 'storage_unavailable' }),
+    })
+  );
+  await f.open().recover();
+  expect(f.effects).toEqual([request.toolCall.id]);
+  expect(f.reported.get(request.toolCall.id)).toEqual(receipt);
+});
+it.each(
+  (
+    [
+      'read',
+      'intent',
+      'send',
+      'ack',
+      'delete',
+      'execution',
+      'execution_wait',
+      'effect',
+      'receipt',
+      'reconcile',
+    ] as const
+  ).flatMap(point => sides.map(side => ({ point, side })))
+)(
+  'stops an old account at $side $point without exposing its result to the next account',
+  async ({ point, side }) => {
+    const f = fixture();
+    if (point === 'reconcile') f.durable.executions = [{ ...request, receipt: null }];
+    const client = f.open();
+    f.hook = (at, edge) => {
+      if (point === 'execution_wait' && at === 'execution' && edge === 'after')
+        f.readiness = { ...ready, foreground: false };
+      if (at === point && edge === side) f.active = { ...scope, ownerUserId: 'another-owner' };
+    };
+    const result =
+      point === 'reconcile'
+        ? await client.recover()
+        : ['execution', 'execution_wait', 'effect', 'receipt'].includes(point)
+          ? await client.dispatch(request)
+          : await client.submit(command);
+    expect(result).toMatchObject({
+      status: ['intent', 'execution', 'execution_wait'].includes(point) ? 'unsent' : 'unknown',
+      error: { code: 'access_revoked' },
+    });
+    const accepted = f.accepted.size,
+      effects = f.effects.length;
+    expect(errorCode(await client.submit({ ...command, commandId: id(80) }))).toBe(
+      'access_revoked'
+    );
+    expect(errorCode(await client.dispatch(request))).toBe('access_revoked');
+    expect(f.accepted.size).toBe(accepted);
+    expect(f.effects).toHaveLength(effects);
+    expect(f.durable.scope).toEqual(scope);
+  }
+);
+it.each(['clientId', 'storageGeneration'] as const)(
+  'invalidates the host when %s changes',
+  async field => {
+    const f = fixture(),
+      client = f.open();
+    f.active = { ...scope, [field]: id(90) };
+    expect(errorCode(await client.submit(command))).toBe('access_revoked');
+    f.active = scope;
+    expect(errorCode(await client.dispatch(request))).toBe('access_revoked');
+    expect(f.effects).toEqual([]);
+  }
+);
+it('ignores a disposed host and restores the committed command on a new host', async () => {
+  const f = fixture(),
+    client = f.open();
+  f.hook = (point, side) => {
+    if (point === 'send' && side === 'after') client.dispose();
+  };
+  expect(errorCode(await client.submit(command))).toBe('access_revoked');
+  f.hook = () => {};
+  await f.open().recover();
+  expect(f.accepted.size).toBe(1);
+  expect(f.durable.intents).toEqual([]);
+});
+const corruptionCases: [string, (state: JournalSnapshot) => unknown][] = [
+  ['missing storage', () => undefined],
+  ['null storage', () => null],
+  ['missing collections', state => ({ scope: state.scope, revision: state.revision })],
+  ['extra fields', state => ({ ...state, extra: true })],
+  ['wrong account', state => ({ ...state, scope: { ...scope, ownerUserId: 'other' } })],
+  ['wrong client', state => ({ ...state, scope: { ...scope, clientId: id(40) } })],
+  ['lost generation', state => ({ ...state, scope: { ...scope, storageGeneration: id(40) } })],
+  ['invalid revision', state => ({ ...state, revision: -1 })],
+  [
+    'changed canonical input',
+    state => ({ ...state, intents: [{ ...state.intents[0], canonicalInput: '{}' }] }),
+  ],
+  ['duplicate command', state => ({ ...state, intents: [...state.intents, ...state.intents] })],
+  [
+    'duplicate acknowledgment',
+    state => ({ ...state, acknowledgments: [...state.acknowledgments, ...state.acknowledgments] }),
+  ],
+  [
+    'unmatched acknowledgment',
+    state => ({
+      ...state,
+      acknowledgments: [
+        {
+          ...state.acknowledgments[0],
+          reply: { status: 'accepted', commandId: id(40), result: null },
+        },
+      ],
+    }),
+  ],
+  [
+    'duplicate execution',
+    state => ({ ...state, executions: [...state.executions, ...state.executions] }),
+  ],
+  [
+    'foreign execution',
+    state => ({
+      ...state,
+      executions: [{ ...state.executions[0], grant: { ...request.grant, ownerUserId: 'other' } }],
+    }),
+  ],
+  [
+    'changed execution target',
+    state => ({
+      ...state,
+      executions: [
+        {
+          ...state.executions[0],
+          toolCall: { ...request.toolCall, executionTarget: { kind: 'client', clientId: id(40) } },
+        },
+      ],
+    }),
+  ],
+  [
+    'invalid receipt',
+    state => ({
+      ...state,
+      executions: [{ ...state.executions[0], receipt: { status: 'succeeded' } }],
+    }),
+  ],
+];
+it.each(corruptionCases)(
+  'blocks %s instead of replacing it with empty storage',
+  async (_name, corrupt) => {
+    const f = fixture();
+    const intent = { command, canonicalInput: canonicalizeValidatedInput(command) };
+    const stored: JournalSnapshot = {
+      scope,
+      revision: 4,
+      intents: [intent],
+      acknowledgments: [
+        { intent, reply: { status: 'accepted', commandId: command.commandId, result: {} } },
+      ],
+      executions: [{ ...request, receipt }],
+    };
+    f.readOverride = () => corrupt(stored);
+    expect(await f.open().submit({ ...command, commandId: id(60) })).toMatchObject({
+      status: 'unknown',
+      error: { code: 'storage_unavailable' },
+    });
+    expect(f.accepted.size).toBe(0);
+    expect(f.durable.revision).toBe(0);
+    expect(f.effects).toEqual([]);
+  }
+);
+
+it('keeps an executed effect unknown when its completion intent cannot commit', async () => {
+  const f = fixture();
+  f.fault = { point: 'intent', side: 'before' };
+  expect(await f.open().dispatch(request)).toMatchObject({
+    status: 'unknown',
+    error: { code: 'storage_unavailable' },
+  });
+  expect(f.effects).toEqual([request.toolCall.id]);
+  expect(f.reported.size).toBe(0);
+  await f.open().recover();
+  expect(f.reported.get(request.toolCall.id)).toEqual(receipt);
+  expect(f.effects).toEqual([request.toolCall.id]);
+});
+it.each([
+  { status: 'denied' },
+  { status: 'cancelled' },
+  { status: 'failed', error: { code: 'unavailable_tool', message: 'Retry', retryable: true } },
+  { status: 'failed', error: { code: 'access_revoked', message: 'Blocked', retryable: false } },
+  { status: 'outcome_unknown', reason: 'Provider reply lost' },
+] satisfies ToolOutcome[])(
+  'preserves a %j receipt without turning it into success or repeating the effect',
+  async result => {
+    const f = fixture();
+    f.effectResult = result;
+    expect((await f.open().dispatch(request)).status).toBe('accepted');
+    await f.open().recover();
+    expect(f.reported.get(request.toolCall.id)).toEqual(result);
+    expect(f.effects).toEqual([request.toolCall.id]);
+  }
+);
+it.each([undefined, { status: 'succeeded' }, { status: 'succeeded', output: {}, extra: true }])(
+  'leaves an invalid executor receipt unknown: %j',
+  async result => {
+    const f = fixture();
+    f.effectResult = result;
+    expect(errorCode(await f.open().dispatch(request))).toBe('invalid_output');
+    await f.open().recover();
+    expect(f.effects).toEqual([request.toolCall.id]);
+    expect(f.reported.size).toBe(0);
+  }
+);
+it('preserves the first committed receipt when a concurrent reconciliation finishes', async () => {
+  const f = fixture();
+  f.durable.executions = [{ ...request, receipt: null }];
+  f.evidence = () => ({ status: 'cancelled' });
+  f.hook = (point, side) => {
+    if (point === 'reconcile' && side === 'after') {
+      f.durable.executions = [{ ...request, receipt }];
+      f.durable.revision++;
+    }
+  };
+  await f.open().recover();
+  expect(f.reported.get(request.toolCall.id)).toEqual(receipt);
+  expect(f.durable.executions[0].receipt).toEqual(receipt);
+  expect(f.effects).toEqual([]);
+});
+it('rejects a journal rollback instead of admitting new work into an empty snapshot', async () => {
+  const f = fixture(),
+    client = f.open();
+  const empty = structuredClone(f.durable);
+  await client.submit(command);
+  f.readOverride = () => empty;
+  expect(errorCode(await client.submit({ ...command, commandId: id(60) }))).toBe(
+    'storage_unavailable'
+  );
+  expect(f.accepted.size).toBe(1);
+  expect(f.durable.acknowledgments[0].intent.command).toEqual(command);
+});
+it.each([
+  [null, 'invalid_input'],
+  [{ ...request, grant: { ...request.grant, inputDigest: 'changed' } }, 'invalid_input'],
+  [{ ...request, grant: { ...request.grant, ownerUserId: 'other' } }, 'access_revoked'],
+  [
+    {
+      ...request,
+      toolCall: { ...request.toolCall, executionTarget: { kind: 'client', clientId: id(60) } },
+      grant: { ...request.grant, clientId: id(60) },
+    },
+    'access_revoked',
+  ],
+  [{ ...request, grant: { ...request.grant, definitionVersion: '2' } }, 'invalid_input'],
+  [
+    {
+      ...request,
+      grant: { ...request.grant, context: { type: 'organization', organizationId: id(60) } },
+    },
+    'invalid_input',
+  ],
+  [{ ...request, grant: { ...request.grant, toolCallId: id(60) } }, 'invalid_input'],
+  [
+    { ...request, toolCall: { ...request.toolCall, state: 'settled', result: receipt } },
+    'invalid_input',
+  ],
+] as const)('refuses invalid or foreign execution authority %#', async (input, code) => {
+  const f = fixture();
+  expect(errorCode(await f.open().dispatch(input))).toBe(code);
+  expect(f.effects).toEqual([]);
+  expect(f.durable.executions).toEqual([]);
+});
+it('does not reuse a message command ID for an execution completion', async () => {
+  const f = fixture();
+  await f.open().submit(command);
+  expect(
+    errorCode(await f.open().dispatch({ ...request, completionCommandId: command.commandId }))
+  ).toBe('invalid_input');
+  expect(f.effects).toEqual([]);
+  expect(f.accepted.size).toBe(1);
+});
+it.each([
+  { available: true },
+  { ...ready, unlock: undefined },
+  { ...ready, connectivity: true },
+] as const)('fails closed for missing or malformed readiness: %j', async readiness => {
+  const f = fixture();
+  f.readiness = readiness as BridgeReadiness;
+  await f.open().dispatch(request);
+  expect(f.effects).toEqual([]);
+  expect(f.durable.executions).toEqual([]);
+});
+it.each(
+  (['dispatch', 'recover'] as const).flatMap(mode =>
+    (mode === 'dispatch' ? [2, 3, 4] : [1, 2, 3, 4, 5]).flatMap(ordinal =>
+      sides.map(side => ({ mode, ordinal, side }))
+    )
+  )
+)(
+  'survives $side journal read $ordinal during $mode without repeating effects',
+  async ({ mode, ordinal, side }) => {
+    const f = fixture();
+    if (mode === 'recover') {
+      f.durable.executions = [{ ...request, receipt: null }];
+      f.evidence = () => receipt;
+    }
+    let reads = 0,
+      killed = false;
+    f.hook = (point, edge) => {
+      if (point === 'read' && edge === side && ++reads === ordinal) {
+        killed = true;
+        throw new Error('Killed during journal read');
+      }
+    };
+    const first = mode === 'dispatch' ? await f.open().dispatch(request) : await f.open().recover();
+    expect(first).toMatchObject({ status: 'unknown', error: { code: 'storage_unavailable' } });
+    expect(killed).toBe(true);
+    const hadIntent = f.durable.executions.length > 0,
+      effects = f.effects.length;
+    f.hook = () => {};
+    const client = f.open();
+    await client.recover();
+    if (mode === 'dispatch') await client.dispatch(request);
+    expect(f.effects).toHaveLength(hadIntent ? effects : 1);
+    if (f.durable.executions[0].receipt !== null)
+      expect(f.reported.get(request.toolCall.id)).toEqual(receipt);
+  }
+);
+it.each([Number.NaN, Date.parse(request.grant.expiresAt)])(
+  'refuses dispatch at an invalid or expired time: %s',
+  async time => {
+    const f = fixture();
+    f.time = time;
+    expect(errorCode(await f.open().dispatch(request))).toBe('access_revoked');
+    expect(f.effects).toEqual([]);
+    expect(f.durable.executions).toEqual([]);
+  }
+);
diff --git a/packages/agent-harness/src/client.ts b/packages/agent-harness/src/client.ts
new file mode 100644
index 0000000000..462f33ab05
--- /dev/null
+++ b/packages/agent-harness/src/client.ts
@@ -0,0 +1,371 @@
+import type { z } from 'zod';
+import { canonicalizeValidatedInput, CommandSchema, type Command } from './commands';
+import { ErrorSchema, ToolOutcomeSchema } from './contracts';
+import { bridgeWaitReason, type ClientBridge } from './bridge';
+import {
+  CommandReplySchema,
+  ExecutionIntentSchema,
+  ExecutionRequestSchema,
+  JournalScopeSchema,
+  JournalSnapshotSchema,
+  completionCommand,
+  executionKey,
+  type CommandIntent,
+  type CommandReply,
+  type ExecutionIntent,
+  type HarnessJournal,
+  type JournalScope,
+  type JournalSnapshot,
+} from './journal';
+
+type Failure = z.infer<typeof ErrorSchema>;
+export type ClientProblem = { status: 'unsent' | 'unknown'; error: Failure };
+export type ClientResult =
+  | {
+      status: 'accepted';
+      command: Command;
+      result: Extract<CommandReply, { status: 'accepted' }>['result'];
+    }
+  | { status: 'rejected'; command: Command; error: Failure }
+  | { status: 'waiting'; reason: NonNullable<ReturnType<typeof bridgeWaitReason>> }
+  | ClientProblem;
+export type CommandTransport = {
+  // Resolve only a committed command result. Reject ambiguous transport failures. The adapter must
+  // bind authentication to this exact scope, not whichever account is active after an await.
+  send: (scope: JournalScope, command: Command) => Promise<unknown>;
+};
+export type ClientOptions = {
+  scope: JournalScope;
+  currentScope: () => JournalScope | null;
+  journal: HarnessJournal;
+  transport: CommandTransport;
+  bridge: ClientBridge;
+  now: () => number;
+  digest: (canonicalArguments: string) => string | Promise<string>;
+};
+const problem = (code: Failure['code'], message: string, retryable = false): Failure => ({
+  code,
+  message,
+  retryable,
+});
+const same = (left: unknown, right: unknown) =>
+  canonicalizeValidatedInput(left) === canonicalizeValidatedInput(right);
+const rejected = (command: Command, code: Failure['code']): ClientResult => ({
+  status: 'rejected',
+  command,
+  error: problem(code, code),
+});
+const acknowledged = (intent: CommandIntent, reply: CommandReply): ClientResult =>
+  reply.status === 'accepted'
+    ? { status: 'accepted', command: intent.command, result: reply.result }
+    : { status: 'rejected', command: intent.command, error: reply.error };
+const blocksUnrelatedDispatch = (result: ClientResult) =>
+  result.status !== 'accepted' &&
+  !(
+    result.status === 'rejected' &&
+    !result.error.retryable &&
+    (result.error.code === 'access_revoked' || result.error.code === 'retired')
+  );
+
+export function createHarnessClient(options: ClientOptions) {
+  const scope = JournalScopeSchema.parse(options.scope);
+  let blocked: Failure | undefined;
+  let disposed = false,
+    revision = -1,
+    uncertain = false;
+  let outcome: ClientProblem['status'] = 'unknown';
+  let queue = Promise.resolve();
+  function guard() {
+    if (disposed || !same(options.currentScope(), scope)) {
+      disposed = true;
+      throw problem('access_revoked', 'The journal scope is no longer active.');
+    }
+  }
+  async function storage<T>(operation: () => Promise<T>): Promise<T> {
+    guard();
+    let result: T;
+    try {
+      result = await operation();
+    } catch {
+      blocked = problem('storage_unavailable', 'Durable storage is unavailable.', true);
+      throw blocked;
+    }
+    guard();
+    return result;
+  }
+  async function load() {
+    return storage(async () => {
+      const state = JournalSnapshotSchema.parse(await options.journal.read(scope));
+      if (!same(state.scope, scope) || state.revision < revision)
+        throw new Error('Journal scope or revision changed');
+      revision = state.revision;
+      return state;
+    });
+  }
+  async function commit(before: JournalSnapshot, next: JournalSnapshot) {
+    const state = await storage(async () => {
+      const checked = JournalSnapshotSchema.parse({ ...next, revision: before.revision + 1 });
+      if ((await options.journal.compareAndSwap(scope, before.revision, checked)) !== true) {
+        outcome = 'unknown';
+        throw new Error('Concurrent journal change');
+      }
+      return checked;
+    });
+    revision = state.revision;
+    return state;
+  }
+  function run<T>(operation: () => Promise<T>, recovering = false): Promise<T | ClientProblem> {
+    const task = queue.then(async () => {
+      outcome = 'unknown';
+      uncertain = recovering;
+      try {
+        guard();
+        if (blocked && !recovering) throw blocked;
+        return await operation();
+      } catch (error) {
+        let failure = error;
+        try {
+          guard();
+        } catch (scopeError) {
+          failure = scopeError;
+        }
+        const parsed = ErrorSchema.safeParse(failure);
+        return {
+          status: outcome,
+          error: parsed.success
+            ? parsed.data
+            : problem('outcome_unknown', 'The outcome requires reconciliation.', true),
+        };
+      }
+    });
+    queue = task.then(() => {});
+    return task;
+  }
+  async function send(command: Command, reviewOf?: string): Promise<ClientResult> {
+    if (command.clientId !== scope.clientId) return rejected(command, 'access_revoked');
+    let state = await load();
+    const canonicalInput = canonicalizeValidatedInput(command);
+    const ack = state.acknowledgments.find(item => item.reply.commandId === command.commandId);
+    const existing = state.intents.find(item => item.command.commandId === command.commandId);
+    if (
+      (ack && ack.intent.canonicalInput !== canonicalInput) ||
+      (existing && existing.canonicalInput !== canonicalInput)
+    )
+      return rejected(command, 'command_conflict');
+    if (ack) {
+      if (existing)
+        await commit(state, { ...state, intents: state.intents.filter(item => item !== existing) });
+      return acknowledged(ack.intent, ack.reply);
+    }
+    if (!existing && command.type === 'sendMessage') {
+      const records = [...state.intents, ...state.acknowledgments.map(item => item.intent)];
+      const stale = state.acknowledgments.find(
+        item =>
+          item.reply.status === 'rejected' &&
+          item.reply.error.code === 'stale_revision' &&
+          item.intent.command.type === 'sendMessage' &&
+          item.intent.command.conversationId === command.conversationId &&
+          !records.some(record => record.reviewOf === item.reply.commandId)
+      );
+      if (stale && reviewOf !== stale.reply.commandId)
+        return acknowledged(stale.intent, stale.reply);
+      if (reviewOf && !stale) return rejected(command, 'invalid_input');
+    }
+    if (
+      command.type === 'completeClientTool' &&
+      !state.executions.some(
+        execution => execution.receipt !== null && same(completionCommand(execution), command)
+      )
+    )
+      return rejected(command, 'invalid_input');
+    if (reviewOf && command.type !== 'sendMessage') return rejected(command, 'invalid_input');
+    const intent = existing ?? { command, canonicalInput, ...(reviewOf ? { reviewOf } : {}) };
+    if (!existing) {
+      outcome = uncertain ? 'unknown' : 'unsent';
+      state = await commit(state, { ...state, intents: [...state.intents, intent] });
+    }
+    guard();
+    uncertain = true;
+    outcome = 'unknown';
+    const raw = await options.transport.send(scope, CommandSchema.parse(intent.command));
+    guard();
+    const parsed = CommandReplySchema.safeParse(raw);
+    if (!parsed.success || parsed.data.commandId !== command.commandId)
+      throw problem('invalid_output', 'Invalid command acknowledgment.');
+    const reply = parsed.data;
+    state = await commit(state, {
+      ...state,
+      acknowledgments: [...state.acknowledgments, { intent, reply }],
+    });
+    await commit(state, {
+      ...state,
+      intents: state.intents.filter(item => item.command.commandId !== command.commandId),
+    });
+    return acknowledged(intent, reply);
+  }
+  async function saveReceipt(execution: ExecutionIntent, raw: unknown) {
+    const receipt = ToolOutcomeSchema.safeParse(raw);
+    if (!receipt.success) throw problem('invalid_output', 'Invalid execution receipt.');
+    const state = await load();
+    const stored = state.executions.find(item => executionKey(item) === executionKey(execution));
+    if (!stored || !same({ ...stored, receipt: null }, { ...execution, receipt: null })) {
+      blocked = problem('storage_unavailable', 'The execution journal changed.', true);
+      throw blocked;
+    }
+    // Another host can reconcile while an executor is returning. Never replace its committed result.
+    if (stored.receipt !== null) return stored;
+    const settled = { ...stored, receipt: receipt.data };
+    await commit(state, {
+      ...state,
+      executions: state.executions.map(item => (item === stored ? settled : item)),
+    });
+    return settled;
+  }
+  async function reconcile(execution: ExecutionIntent): Promise<ClientResult> {
+    uncertain = true;
+    outcome = 'unknown';
+    if (execution.receipt === null) {
+      guard();
+      const raw = await options.bridge.reconcileReceipt(
+        scope,
+        ExecutionIntentSchema.parse(execution)
+      );
+      guard();
+      if (raw === null)
+        return {
+          status: 'unknown',
+          error: problem('outcome_unknown', 'Execution has no confirmed receipt.'),
+        };
+      execution = await saveReceipt(execution, raw);
+    }
+    return send(completionCommand(execution));
+  }
+  async function reconcileExecutions(state: JournalSnapshot) {
+    const results: ClientResult[] = [];
+    // Existing receipts precede evidence queries; neither path calls execute.
+    for (const execution of [
+      ...state.executions.filter(item => item.receipt !== null),
+      ...state.executions.filter(item => item.receipt === null),
+    ])
+      results.push(await reconcile(execution));
+    return results;
+  }
+  return {
+    submit(input: unknown, review?: { reviewedCommandId: string }) {
+      const parsed = CommandSchema.safeParse(input);
+      const reviewedCommandId = review?.reviewedCommandId;
+      return run(() => {
+        if (!parsed.success) throw problem('invalid_input', 'Invalid command.');
+        return send(parsed.data, reviewedCommandId);
+      });
+    },
+    recover() {
+      return run(async () => {
+        const state = await load();
+        blocked = undefined;
+        const results = await reconcileExecutions(state);
+        for (const intent of (await load()).intents)
+          results.push(await send(intent.command, intent.reviewOf));
+        const restored = await load();
+        const reviewed = new Set(
+          [...restored.intents, ...restored.acknowledgments.map(item => item.intent)].map(
+            item => item.reviewOf
+          )
+        );
+        for (const ack of restored.acknowledgments)
+          if (
+            ack.reply.status === 'rejected' &&
+            ack.reply.error.code === 'stale_revision' &&
+            !reviewed.has(ack.reply.commandId) &&
+            !results.some(
+              result => 'command' in result && result.command.commandId === ack.reply.commandId
+            )
+          )
+            results.push(acknowledged(ack.intent, ack.reply));
+        return results;
+      }, true);
+    },
+    dispatch(input: unknown) {
+      const parsed = ExecutionRequestSchema.safeParse(input);
+      return run(async (): Promise<ClientResult> => {
+        if (!parsed.success) throw problem('invalid_input', 'Invalid execution request.');
+        const validated = ExecutionIntentSchema.safeParse({ ...parsed.data, receipt: null });
+        if (!validated.success || validated.data.toolCall.state === 'settled')
+          throw problem('invalid_input', 'Invalid execution intent.');
+        const execution = validated.data;
+        if (
+          execution.grant.ownerUserId !== scope.ownerUserId ||
+          execution.grant.clientId !== scope.clientId
+        )
+          throw problem('access_revoked', 'Execution belongs to another scope.');
+        let state = await load();
+        const existing = state.executions.find(
+          item => executionKey(item) === executionKey(execution)
+        );
+        if (existing) {
+          if (
+            !same(existing.grant, execution.grant) ||
+            !same(existing.toolCall, execution.toolCall)
+          )
+            throw problem(
+              'outcome_unknown',
+              'An existing execution cannot change grants or input.'
+            );
+          for (const stored of state.executions.filter(
+            item => item !== existing && item.receipt !== null
+          )) {
+            const result = await reconcile(stored);
+            if (blocksUnrelatedDispatch(result)) return result;
+          }
+          return reconcile(existing);
+        }
+        const pending = (await reconcileExecutions(state)).find(blocksUnrelatedDispatch);
+        if (pending) return pending;
+        state = await load();
+        if (
+          [...state.intents, ...state.acknowledgments.map(item => item.intent)].some(
+            item => item.command.commandId === execution.completionCommandId
+          ) ||
+          state.executions.some(item => item.completionCommandId === execution.completionCommandId)
+        )
+          throw problem('invalid_input', 'The completion command ID is already reserved.');
+        if (
+          execution.grant.inputDigest !==
+          (await options.digest(canonicalizeValidatedInput(execution.toolCall.arguments)))
+        )
+          throw problem('invalid_input', 'Execution input does not match the grant.');
+        guard();
+        const readiness = () => {
+          guard();
+          const now = options.now();
+          if (!Number.isFinite(now) || Date.parse(execution.grant.expiresAt) <= now)
+            throw problem('access_revoked', 'Execution grant expired or clock unavailable.');
+          return bridgeWaitReason(options.bridge.readiness(scope, execution));
+        };
+        const wait = readiness();
+        if (wait) return { status: 'waiting', reason: wait };
+        outcome = uncertain ? 'unknown' : 'unsent';
+        const committed = await commit(state, {
+          ...state,
+          executions: [...state.executions, execution],
+        });
+        const changed = readiness();
+        if (changed) {
+          // Only this live dispatch proves execute was never called. Release its fence atomically;
+          // a crash before the release leaves the intent uncertain, never permission to replay.
+          await commit(committed, { ...committed, executions: state.executions });
+          return { status: 'waiting', reason: changed };
+        }
+        uncertain = true;
+        outcome = 'unknown';
+        const raw = await options.bridge.execute(scope, ExecutionIntentSchema.parse(execution));
+        guard();
+        return reconcile(await saveReceipt(execution, raw));
+      });
+    },
+    dispose() {
+      disposed = true;
+    },
+  };
+}
+export type HarnessClient = ReturnType<typeof createHarnessClient>;
diff --git a/packages/agent-harness/src/journal.ts b/packages/agent-harness/src/journal.ts
new file mode 100644
index 0000000000..0cc0c4aba4
--- /dev/null
+++ b/packages/agent-harness/src/journal.ts
@@ -0,0 +1,125 @@
+import { z } from 'zod';
+import { canonicalizeValidatedInput, CommandSchema, type Command } from './commands';
+import { ErrorSchema, ExecutionGrantSchema, ToolCallSchema, ToolOutcomeSchema } from './contracts';
+import { AGENT_HARNESS_PROTOCOL_VERSION } from './version';
+
+export const JournalScopeSchema = z
+  .strictObject({ ownerUserId: z.string().min(1), clientId: z.uuid(), storageGeneration: z.uuid() })
+  .readonly();
+export type JournalScope = z.infer<typeof JournalScopeSchema>;
+export const CommandReplySchema = z.discriminatedUnion('status', [
+  z.strictObject({ status: z.literal('accepted'), commandId: z.uuid(), result: z.json() }),
+  z.strictObject({ status: z.literal('rejected'), commandId: z.uuid(), error: ErrorSchema }),
+]);
+export type CommandReply = z.infer<typeof CommandReplySchema>;
+export const CommandIntentSchema = z
+  .strictObject({
+    command: CommandSchema,
+    canonicalInput: z.string(),
+    reviewOf: z.uuid().optional(),
+  })
+  .refine(intent => canonicalizeValidatedInput(intent.command) === intent.canonicalInput);
+export type CommandIntent = z.infer<typeof CommandIntentSchema>;
+const AcknowledgmentSchema = z
+  .strictObject({ intent: CommandIntentSchema, reply: CommandReplySchema })
+  .refine(ack => ack.intent.command.commandId === ack.reply.commandId);
+export const ExecutionRequestSchema = z.strictObject({
+  toolCall: ToolCallSchema,
+  grant: ExecutionGrantSchema,
+  completionCommandId: z.uuid(),
+});
+export const ExecutionIntentSchema = ExecutionRequestSchema.extend({
+  receipt: ToolOutcomeSchema.nullable(),
+}).refine(
+  ({ toolCall, grant }) =>
+    toolCall.executionTarget.kind === 'client' &&
+    toolCall.executionTarget.clientId === grant.clientId &&
+    toolCall.id === grant.toolCallId &&
+    toolCall.definitionVersion === grant.definitionVersion &&
+    canonicalizeValidatedInput(toolCall.context) === canonicalizeValidatedInput(grant.context)
+);
+export type ExecutionIntent = z.infer<typeof ExecutionIntentSchema>;
+export type ExecutionRequest = z.infer<typeof ExecutionRequestSchema>;
+export function completionCommand(execution: ExecutionIntent): Command {
+  if (!execution.receipt) throw new Error('A committed receipt is required');
+  return CommandSchema.parse({
+    type: 'completeClientTool',
+    protocolVersion: AGENT_HARNESS_PROTOCOL_VERSION,
+    commandId: execution.completionCommandId,
+    clientId: execution.grant.clientId,
+    conversationId: execution.grant.conversationId,
+    toolCallId: execution.grant.toolCallId,
+    grantId: execution.grant.id,
+    generation: execution.grant.generation,
+    result: execution.receipt,
+  });
+}
+export const executionKey = (execution: ExecutionRequest) =>
+  `${execution.grant.conversationId}:${execution.toolCall.id}`;
+const unique = (values: string[]) => new Set(values).size === values.length;
+export const JournalSnapshotSchema = z
+  .strictObject({
+    scope: JournalScopeSchema,
+    revision: z.int().nonnegative(),
+    intents: z.array(CommandIntentSchema),
+    acknowledgments: z.array(AcknowledgmentSchema),
+    executions: z.array(ExecutionIntentSchema),
+  })
+  .refine(state => {
+    const records = [...state.intents, ...state.acknowledgments.map(ack => ack.intent)];
+    return (
+      unique(state.intents.map(intent => intent.command.commandId)) &&
+      unique(state.acknowledgments.map(ack => ack.reply.commandId)) &&
+      unique(state.executions.map(executionKey)) &&
+      unique(state.executions.map(execution => execution.completionCommandId)) &&
+      state.executions.every(
+        execution =>
+          execution.grant.ownerUserId === state.scope.ownerUserId &&
+          execution.grant.clientId === state.scope.clientId
+      ) &&
+      records.every(intent => {
+        const command = intent.command;
+        const ack = state.acknowledgments.find(item => item.reply.commandId === command.commandId);
+        const reviewed = state.acknowledgments.find(
+          item => item.reply.commandId === intent.reviewOf
+        );
+        return (
+          command.clientId === state.scope.clientId &&
+          !state.executions.some(
+            execution =>
+              execution.completionCommandId === command.commandId &&
+              command.type !== 'completeClientTool'
+          ) &&
+          (!ack || canonicalizeValidatedInput(ack.intent) === canonicalizeValidatedInput(intent)) &&
+          (!intent.reviewOf ||
+            (reviewed?.reply.status === 'rejected' &&
+              reviewed.reply.error.code === 'stale_revision' &&
+              reviewed.intent.command.type === 'sendMessage' &&
+              command.type === 'sendMessage' &&
+              command.commandId !== intent.reviewOf &&
+              command.conversationId === reviewed.intent.command.conversationId)) &&
+          (command.type !== 'completeClientTool' ||
+            state.executions.some(
+              execution =>
+                execution.receipt !== null &&
+                canonicalizeValidatedInput(completionCommand(execution)) === intent.canonicalInput
+            ))
+        );
+      })
+    );
+  });
+export type JournalSnapshot = z.infer<typeof JournalSnapshotSchema>;
+
+// Initialize a generation only for a fresh clientId during registration. Storage loss must suspend the old
+// registration, never reuse its grants. Missing/corrupt storage is an error, never an empty journal.
+// Reads must return one atomic, durable, exactly scoped snapshot.
+export type HarnessJournal = {
+  read: (scope: JournalScope) => Promise<unknown>;
+  // Compare scope AND revision in one strictly durable transaction. Resolve true only after commit;
+  // false means no write. Reject uncertain commits. Never replace, reset, or fall back to memory.
+  compareAndSwap: (
+    scope: JournalScope,
+    expectedRevision: number,
+    next: JournalSnapshot
+  ) => Promise<boolean>;
+};