diff --git a/src/shared/agent-session-journal-item-key.ts b/src/shared/agent-session-journal-item-key.ts index 45931ddc4ed..9548bc4de63 100644 --- a/src/shared/agent-session-journal-item-key.ts +++ b/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 +} diff --git a/src/shared/agent-session-mutation-envelope.test.ts b/src/shared/agent-session-mutation-envelope.test.ts new file mode 100644 index 00000000000..406cc3788d7 --- /dev/null +++ b/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 = {}) { + 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' } + }) + }) +}) diff --git a/src/shared/agent-session-mutation-envelope.ts b/src/shared/agent-session-mutation-envelope.ts new file mode 100644 index 00000000000..40a3ec5fbf6 --- /dev/null +++ b/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 { + 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) + .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.' + } +} diff --git a/src/shared/agent-session-wire.ts b/src/shared/agent-session-wire.ts new file mode 100644 index 00000000000..fc6982036fe --- /dev/null +++ b/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 = + | { + 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 +} diff --git a/src/shared/protocol-version.ts b/src/shared/protocol-version.ts index c967c66822b..53aff48de44 100644 --- a/src/shared/protocol-version.ts +++ b/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