Define the structured agent-session wire contract

Adds the request/refusal shapes every `agentSession.*` method speaks, the
four-field idempotency envelope and its lease admission, the item-key inverse an
upsert needs, and the `agent-session.structured.v1` capability. The protocol
version stays at 3: the surface is additive, and bumping it would strand every
paired device on the current release rather than degrade one feature.

src/shared/agent-session-journal-item-key.ts

@@ -49,3 +49,35 @@ export function agentJournalItemKey(identity: AgentJournalItemIdentity): string
 export function agentJournalSubmissionKey(clientMessageId: string): string {
   return agentJournalItemKey({ provider: 'orca', clientMessageId })
 }
+
+/**
+ * Inverse of {@link agentJournalItemKey}. Clients hold item KEYS, but an upsert
+ * needs the identity behind one — answering an approval re-appends the same
+ * item at the next revision. Every component is percent-encoded, and
+ * `encodeURIComponent` escapes the delimiter, so the split is unambiguous.
+ */
+export function parseAgentJournalItemKey(key: string): AgentJournalItemIdentity | null {
+  const parts = key.split(KEY_DELIMITER).map((part) => decodeURIComponent(part))
+  const [provider, ...rest] = parts
+  if (provider === 'codex' && rest.length === 3) {
+    const ordinal = Number(rest[2])
+    return Number.isSafeInteger(ordinal) && ordinal >= 0
+      ? { provider, threadId: rest[0] as string, turnId: rest[1] as string, ordinal }
+      : null
+  }
+  if (provider === 'claude' && rest.length === 2) {
+    return { provider, sessionId: rest[0] as string, uuid: rest[1] as string }
+  }
+  if (provider === 'orca' && rest.length === 1) {
+    return { provider, clientMessageId: rest[0] as string }
+  }
+  if (provider === 'legacy' && rest.length === 3) {
+    return {
+      provider,
+      agent: rest[0] as string,
+      sessionId: rest[1] as string,
+      recordId: rest[2] as string
+    }
+  }
+  return null
+}

src/shared/agent-session-mutation-envelope.test.ts

@@ -0,0 +1,170 @@
+import { describe, expect, it } from 'vitest'
+import type { AgentSessionOperationDecision } from './agent-session-operation-ledger'
+import type { AgentSessionLease } from './agent-session-record'
+import {
+  admitAgentSessionMutation,
+  agentSessionFingerprintConflict,
+  computeAgentSessionPayloadFingerprint
+} from './agent-session-mutation-envelope'
+import type { AgentSessionMutationEnvelope } from './agent-session-wire'
+
+const LEASE: AgentSessionLease = {
+  sessionId: 'session-1',
+  runtimeKind: 'native',
+  provenHandleLinkId: 'link-1',
+  deathEvidence: null,
+  runtimeFence: 4,
+  ownerProcess: { hostId: 'host-1', pid: 42, processStartTimeMs: null, spawnToken: 'token-1' },
+  reservedSpawnToken: null,
+  leaseDeadlineAt: 10_000,
+  lastRenewedAt: 9_000,
+  handoffOperationId: null,
+  journalCheckpoint: null,
+  claimKeyId: 'key-1',
+  claimStatus: 'live',
+  unreconciled: false,
+  handoffStage: null
+}
+
+function envelope(overrides: Partial<AgentSessionMutationEnvelope> = {}) {
+  return {
+    sessionId: 'session-1',
+    clientOperationId: 'op-1',
+    expectedRuntimeFence: 4,
+    payloadFingerprint: 'f'.repeat(64),
+    ...overrides
+  }
+}
+
+function row(fingerprint: string) {
+  return {
+    callerKey: 'caller-1',
+    operationId: 'op-1',
+    fingerprint,
+    operationTimestamp: 1_000,
+    recordedAt: 1_000,
+    expiresAt: 100_000,
+    outcome: { status: 'pending' as const }
+  }
+}
+
+const ADMIT = (fingerprint: string): AgentSessionOperationDecision => ({
+  decision: 'admit',
+  row: row(fingerprint)
+})
+
+describe('computeAgentSessionPayloadFingerprint', () => {
+  it('is stable across key order at every depth', () => {
+    const a = computeAgentSessionPayloadFingerprint({
+      method: 'agentSession.send',
+      sessionId: 'session-1',
+      fields: { body: { kind: 'message', blocks: [{ type: 'text', text: 'hi' }] }, extra: 1 }
+    })
+    const b = computeAgentSessionPayloadFingerprint({
+      method: 'agentSession.send',
+      sessionId: 'session-1',
+      fields: { extra: 1, body: { blocks: [{ text: 'hi', type: 'text' }], kind: 'message' } }
+    })
+    expect(a).toBe(b)
+    expect(a).toMatch(/^[0-9a-f]{64}$/)
+  })
+
+  it('separates one payload from another and one method from another', () => {
+    const send = computeAgentSessionPayloadFingerprint({
+      method: 'agentSession.send',
+      sessionId: 'session-1',
+      fields: { text: 'hi' }
+    })
+    expect(
+      computeAgentSessionPayloadFingerprint({
+        method: 'agentSession.send',
+        sessionId: 'session-1',
+        fields: { text: 'hi there' }
+      })
+    ).not.toBe(send)
+    expect(
+      computeAgentSessionPayloadFingerprint({
+        method: 'agentSession.cancel',
+        sessionId: 'session-1',
+        fields: { text: 'hi' }
+      })
+    ).not.toBe(send)
+  })
+})
+
+describe('agentSessionFingerprintConflict', () => {
+  it('refuses a payload that does not match what the client declared', () => {
+    const conflict = agentSessionFingerprintConflict(envelope(), 'a'.repeat(64))
+    expect(conflict?.code).toBe('agent_session_operation_conflict')
+  })
+
+  it('passes a matching declaration', () => {
+    expect(agentSessionFingerprintConflict(envelope(), 'f'.repeat(64))).toBeNull()
+  })
+})
+
+describe('admitAgentSessionMutation', () => {
+  const base = { envelope: envelope(), hostFingerprint: 'f'.repeat(64), lease: LEASE }
+
+  it('admits a first-time operation under a live lease at the expected fence', () => {
+    expect(admitAgentSessionMutation({ ...base, ledger: ADMIT('f'.repeat(64)) }).decision).toBe(
+      'admit'
+    )
+  })
+
+  it('replays a recorded operation without re-checking the fence', () => {
+    const admission = admitAgentSessionMutation({
+      ...base,
+      envelope: envelope({ expectedRuntimeFence: 1 }),
+      ledger: { decision: 'replay', row: row('f'.repeat(64)) }
+    })
+    expect(admission.decision).toBe('replay')
+  })
+
+  it('refuses a stale fence and hands back the current one', () => {
+    const admission = admitAgentSessionMutation({
+      ...base,
+      envelope: envelope({ expectedRuntimeFence: 3 }),
+      ledger: ADMIT('f'.repeat(64))
+    })
+    expect(admission).toMatchObject({
+      decision: 'refused',
+      refusal: { code: 'agent_session_checkpoint_stale', currentFence: 4 }
+    })
+  })
+
+  it('refuses a writer while the lease is unreconciled', () => {
+    const admission = admitAgentSessionMutation({
+      ...base,
+      lease: { ...LEASE, unreconciled: true },
+      ledger: ADMIT('f'.repeat(64))
+    })
+    expect(admission).toMatchObject({
+      decision: 'refused',
+      refusal: { code: 'execution_owner_reconciling' }
+    })
+  })
+
+  it('refuses a writer mid-handoff', () => {
+    const admission = admitAgentSessionMutation({
+      ...base,
+      lease: { ...LEASE, handoffStage: 'new-owner-proving' },
+      ledger: ADMIT('f'.repeat(64))
+    })
+    expect(admission).toMatchObject({
+      decision: 'refused',
+      refusal: { code: 'agent_session_conflict' }
+    })
+  })
+
+  it('surfaces a ledger refusal verbatim', () => {
+    const admission = admitAgentSessionMutation({
+      ...base,
+      ledger: { decision: 'refused', code: 'agent_session_operation_expired' }
+    })
+    expect(admission).toMatchObject({
+      decision: 'refused',
+      refusal: { code: 'agent_session_operation_expired' }
+    })
+  })
+})

src/shared/agent-session-mutation-envelope.ts

@@ -0,0 +1,151 @@
+// Admission for one mutating `agentSession.*` call.
+//
+// The rules themselves live in the durable ledger and the lease adjudicator;
+// this is only the fixed order they are applied in, plus the payload
+// fingerprint both peers derive from the same request fields. Nothing here
+// re-derives who may write — that answer comes from
+// `agentSessionLeaseAdmitsWriter` alone.
+
+import { createHash } from 'node:crypto'
+import type {
+  AgentSessionOperationDecision,
+  AgentSessionOperationRow
+} from './agent-session-operation-ledger'
+import {
+  agentSessionLeaseAdmitsWriter,
+  isAgentSessionFenceCurrent
+} from './agent-session-lease-adjudication'
+import type { AgentSessionLease } from './agent-session-record'
+import type { AgentSessionMutationEnvelope, AgentSessionWireRefusal } from './agent-session-wire'
+
+/**
+ * Stable digest over the fields that define what this call DOES. Keys are
+ * emitted in sorted order at every depth so two peers serializing the same
+ * request in different property order agree, and an undefined field is dropped
+ * rather than hashed as present-but-empty.
+ */
+export function computeAgentSessionPayloadFingerprint(input: {
+  method: string
+  sessionId: string
+  fields: Record<string, unknown>
+}): string {
+  const canonical = canonicalize({
+    method: input.method,
+    sessionId: input.sessionId,
+    fields: input.fields
+  })
+  return createHash('sha256').update(canonical).digest('hex')
+}
+
+function canonicalize(value: unknown): string {
+  if (value === null || typeof value !== 'object') {
+    return JSON.stringify(value ?? null)
+  }
+  if (Array.isArray(value)) {
+    return `[${value.map(canonicalize).join(',')}]`
+  }
+  const entries = Object.entries(value as Record<string, unknown>)
+    .filter(([, entry]) => entry !== undefined)
+    .sort(([left], [right]) => (left < right ? -1 : left > right ? 1 : 0))
+  return `{${entries.map(([key, entry]) => `${JSON.stringify(key)}:${canonicalize(entry)}`).join(',')}}`
+}
+
+/**
+ * A retry whose payload changed is a different call wearing the same id.
+ * Checked BEFORE the ledger is consulted, so a refused call never leaves an
+ * admitted row that a later honest retry would replay as already-done.
+ */
+export function agentSessionFingerprintConflict(
+  envelope: AgentSessionMutationEnvelope,
+  hostFingerprint: string
+): AgentSessionWireRefusal | null {
+  return envelope.payloadFingerprint === hostFingerprint
+    ? null
+    : {
+        code: 'agent_session_operation_conflict',
+        message:
+          'The payload does not match the fingerprint the client declared for this operation.'
+      }
+}
+
+export type AgentSessionMutationAdmission =
+  | { decision: 'admit'; row: AgentSessionOperationRow }
+  /** The recorded outcome answers this call; do not run the effect again. */
+  | { decision: 'replay'; row: AgentSessionOperationRow }
+  | { decision: 'refused'; refusal: AgentSessionWireRefusal }
+
+/**
+ * Fixed order: fingerprint agreement, then the ledger (so a retry replays
+ * before anything else can refuse it), then the lease, then the fence. Putting
+ * the ledger ahead of the fence is deliberate — a retry that crossed an owner
+ * change must still return its recorded answer instead of a stale-checkpoint
+ * refusal the client would then resend as a second effect.
+ */
+export function admitAgentSessionMutation(input: {
+  envelope: AgentSessionMutationEnvelope
+  /** Fingerprint the host computed from the request it actually received. */
+  hostFingerprint: string
+  /** Decision from the durable ledger, evaluated under `hostFingerprint`. */
+  ledger: AgentSessionOperationDecision
+  lease: AgentSessionLease
+}): AgentSessionMutationAdmission {
+  const { envelope, lease, ledger } = input
+  const mismatch = agentSessionFingerprintConflict(envelope, input.hostFingerprint)
+  if (mismatch) {
+    return { decision: 'refused', refusal: mismatch }
+  }
+  if (ledger.decision === 'refused') {
+    return {
+      decision: 'refused',
+      refusal: {
+        code: ledger.code,
+        message: `Operation ${envelope.clientOperationId} was refused: ${ledger.code}.`
+      }
+    }
+  }
+  if (ledger.decision === 'replay') {
+    return { decision: 'replay', row: ledger.row }
+  }
+  const leaseRefusal = refuseUnlessWriterAdmitted(lease)
+  if (leaseRefusal) {
+    return { decision: 'refused', refusal: leaseRefusal }
+  }
+  if (
+    envelope.expectedRuntimeFence === null ||
+    !isAgentSessionFenceCurrent(lease, envelope.expectedRuntimeFence)
+  ) {
+    return {
+      decision: 'refused',
+      refusal: {
+        code: 'agent_session_checkpoint_stale',
+        message: `Expected runtime fence ${envelope.expectedRuntimeFence ?? 'none'}; the session is at ${lease.runtimeFence}.`,
+        currentFence: lease.runtimeFence
+      }
+    }
+  }
+  return { decision: 'admit', row: ledger.row }
+}
+
+/** Why the single admission oracle said no, mapped to what the client can do
+ *  about it. The predicate itself is never re-implemented here. */
+function refuseUnlessWriterAdmitted(lease: AgentSessionLease): AgentSessionWireRefusal | null {
+  if (agentSessionLeaseAdmitsWriter(lease)) {
+    return null
+  }
+  if (lease.unreconciled) {
+    return {
+      code: 'execution_owner_reconciling',
+      message: 'This host has not yet adjudicated the session lease.'
+    }
+  }
+  if (lease.handoffStage !== null) {
+    return {
+      code: 'agent_session_conflict',
+      message: `The session is mid-handoff (${lease.handoffStage}).`
+    }
+  }
+  return {
+    code: 'agent_session_ownership_unknown',
+    message: 'The session has no live owner to accept writes.'
+  }
+}

src/shared/agent-session-wire.ts

@@ -0,0 +1,166 @@
+// ─── Structured agent-session wire contract ─────────────────────────────────
+// The shapes `agentSession.*` accepts and publishes. Phase 2 builds provider
+// adapters and the mobile client against exactly these types, so everything
+// here must be plain JSON and every new field must be optional to old readers
+// (docs/reference/remote-wire-compatibility.md).
+
+import type {
+  AgentJournalCursor,
+  AgentJournalRenderItem,
+  AgentJournalResetReason,
+  AgentJournalResolution,
+  AgentJournalSnapshot,
+  AgentJournalSubmission
+} from './agent-session-journal-types'
+
+/** Backward paging is the client's normal read; 40 matches the page size the
+ *  mobile list renders without a visible fill-in. */
+export const AGENT_SESSION_HISTORY_DEFAULT_LIMIT = 40
+export const AGENT_SESSION_HISTORY_MAX_LIMIT = 200
+
+export const AGENT_SESSION_HISTORY_DIRECTIONS = ['tail', 'before', 'after'] as const
+/** `tail` is the newest page, `before` pages backward, `after` catches a live
+ *  reader up. Only `after` needs replayable rows; the other two read the
+ *  reduced timeline and so survive compaction. */
+export type AgentSessionHistoryDirection = (typeof AGENT_SESSION_HISTORY_DIRECTIONS)[number]
+
+export type AgentSessionHistoryRequest = {
+  sessionId: string
+  direction: AgentSessionHistoryDirection
+  /** Required for `before` and `after`; ignored for `tail`. */
+  cursor?: AgentJournalCursor
+  limit?: number
+}
+
+export type AgentSessionHistoryPage = {
+  sessionId: string
+  epoch: string
+  direction: AgentSessionHistoryDirection
+  items: AgentJournalRenderItem[]
+  /** Submissions overlapping this page, so an unconfirmed bubble renders with
+   *  its dispatch state instead of as a plain message. */
+  submissions: AgentJournalSubmission[]
+  /** Page edges. `nextCursor` is what the client sends back for the same
+   *  direction; it equals the request cursor when the page is empty. */
+  window: {
+    oldest: AgentJournalCursor | null
+    newest: AgentJournalCursor | null
+    nextCursor: AgentJournalCursor
+  }
+  hasOlder: boolean
+  hasNewer: boolean
+}
+
+export type AgentSessionHistoryResult =
+  | { ok: true; page: AgentSessionHistoryPage }
+  /** Every reset forces a clean reload; the snapshot is inlined so the client
+   *  never has to make a second call to recover. */
+  | { ok: false; reset: AgentJournalResetReason; snapshot: AgentJournalSnapshot }
+
+/** Cursor-qualified incremental publication. Items and submissions carry their
+ *  CURRENT reduced state rather than a delta, so applying a batch twice
+ *  converges instead of double-appending. */
+export type AgentSessionJournalBatch = {
+  cursor: AgentJournalCursor
+  items: AgentJournalRenderItem[]
+  removedItemIds: string[]
+  submissions: AgentJournalSubmission[]
+}
+
+export type AgentSessionSubscribeEvent =
+  | { type: 'snapshot'; sessionId: string; snapshot: AgentJournalSnapshot; fence: number }
+  | { type: 'batch'; sessionId: string; batch: AgentSessionJournalBatch }
+  | {
+      type: 'reset'
+      sessionId: string
+      reset: AgentJournalResetReason
+      snapshot: AgentJournalSnapshot
+    }
+  | { type: 'end' }
+
+// ─── Mutation envelope ──────────────────────────────────────────────────────
+
+/**
+ * The four fields every mutating call carries. Same operation id and same
+ * fingerprint replays the recorded outcome; a different fingerprint under one
+ * operation id is a conflict, never a second effect.
+ */
+export type AgentSessionMutationEnvelope = {
+  sessionId: string
+  clientOperationId: string
+  /** Null only on a create for a session that does not exist yet. */
+  expectedRuntimeFence: number | null
+  /** Client-declared; the host recomputes it and compares. */
+  payloadFingerprint: string
+}
+
+export const AGENT_SESSION_WIRE_REFUSAL_CODES = [
+  'structured_agent_session_unsupported',
+  'agent_session_checkpoint_stale',
+  'agent_session_conflict',
+  'agent_session_ownership_unknown',
+  'agent_session_operation_conflict',
+  'agent_session_operation_expired',
+  'agent_session_operation_capacity',
+  'agent_session_operation_invalid',
+  'agent_session_item_revision_stale',
+  'agent_session_already_resolved',
+  'agent_session_identity_required',
+  'agent_session_journal_unreadable',
+  'execution_owner_reconciling'
+] as const
+export type AgentSessionWireRefusalCode = (typeof AGENT_SESSION_WIRE_REFUSAL_CODES)[number]
+
+export type AgentSessionWireRefusal = {
+  code: AgentSessionWireRefusalCode
+  message: string
+  /** On a stale fence, so the client can retry without another round trip. */
+  currentFence?: number
+  /** On a lost compare-and-set: the winning answer and who gave it. */
+  resolution?: AgentJournalResolution
+  /** On a lost compare-and-set: the revision the host actually holds. */
+  currentRevision?: number
+}
+
+export type AgentSessionMutationResult<TValue> =
+  | {
+      ok: true
+      /** True when the recorded outcome was returned instead of a new effect. */
+      replayed: boolean
+      fence: number
+      cursor: AgentJournalCursor
+      value: TValue
+    }
+  | { ok: false; refusal: AgentSessionWireRefusal }
+
+// ─── Per-method payloads ────────────────────────────────────────────────────
+
+export type AgentSessionAttachResult = {
+  sessionId: string
+  fence: number
+  snapshot: AgentJournalSnapshot
+  /** Submissions the crash boundary settled as `unknown` while attaching. */
+  unconfirmedClientMessageIds: string[]
+}
+
+export type AgentSessionSendResult = {
+  clientMessageId: string
+  submission: AgentJournalSubmission
+}
+
+export type AgentSessionCancelResult = {
+  /** The turn the client named, echoed so a late reply can be matched. */
+  turnId: string
+  cancelled: boolean
+}
+
+export type AgentSessionPromptResult = {
+  itemId: string
+  revision: number
+  resolution: AgentJournalResolution
+}
+
+export type AgentSessionOptionResult = {
+  key: string
+  value: string
+}

src/shared/protocol-version.ts

@@ -79,6 +79,11 @@ export const AGENT_SESSION_HOST_AUTHORITY_RUNTIME_CAPABILITY =
   'agent-session.host-authority.v1' as const
 export const AGENT_SESSION_OMP_RESUME_PATH_RUNTIME_CAPABILITY =
   'agent-session.omp-resume-path.v1' as const
+// Why: structured sessions are journal-backed, not PTY-backed, so a client that
+// cannot read them must not see them at all — it would render an agent tab it
+// can neither display nor drive. The host also refuses every agentSession.*
+// method from a connection that does not advertise this.
+export const STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY = 'agent-session.structured.v1' as const
 // Why: older runtimes strip mutation owner fields, so clients must fence writes before RPC.
 export const FILE_MUTATION_OWNERSHIP_RUNTIME_CAPABILITY = 'files.mutation-ownership.v1' as const
 export const FILE_MUTATION_OWNERSHIP_UPDATE_REQUIRED_MESSAGE =
@@ -114,6 +119,7 @@ export const RUNTIME_CAPABILITIES = [
   REMOTE_SERVER_UPDATE_CAPABILITY,
   AGENT_SESSION_HOST_AUTHORITY_RUNTIME_CAPABILITY,
   AGENT_SESSION_OMP_RESUME_PATH_RUNTIME_CAPABILITY,
+  STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY,
   FILE_MUTATION_OWNERSHIP_RUNTIME_CAPABILITY,
   ACCOUNT_IMPORT_RUNTIME_CAPABILITY,
   CODEX_RESET_CREDIT_RUNTIME_CAPABILITY