diff --git a/src/main/runtime/agent-session-claim-key-retention.ts b/src/main/runtime/agent-session-claim-key-retention.ts new file mode 100644 index 00000000000..80d13aed2b6 --- /dev/null +++ b/src/main/runtime/agent-session-claim-key-retention.ts @@ -0,0 +1,29 @@ +// Retired execution-claim keys. Split from the store on the same rule its ledger admission is: +// the state transition lives here, the transaction stays in the store. + +import type { AgentSessionStoreState } from './agent-session-record-store-file' + +/** Retired claim keys stay verifiable this long so a rotation cannot strand a running agent. */ +export const AGENT_SESSION_CLAIM_KEY_RETENTION_MS = 30 * 24 * 60 * 60 * 1000 + +export function isAgentSessionClaimKeyVerifiable( + state: AgentSessionStoreState, + keyId: string, + now: number +): boolean { + const retired = state.retiredClaimKeys.find((entry) => entry.keyId === keyId) + return !retired || now - retired.retiredAt <= AGENT_SESSION_CLAIM_KEY_RETENTION_MS +} + +export function retireAgentSessionClaimKey( + state: AgentSessionStoreState, + keyId: string, + now: number +): void { + if (!state.retiredClaimKeys.some((entry) => entry.keyId === keyId)) { + state.retiredClaimKeys.push({ keyId, retiredAt: now }) + } + state.retiredClaimKeys = state.retiredClaimKeys.filter( + (entry) => now - entry.retiredAt <= AGENT_SESSION_CLAIM_KEY_RETENTION_MS + ) +} diff --git a/src/main/runtime/agent-session-record-store.test.ts b/src/main/runtime/agent-session-record-store.test.ts index 4b325fa0ff3..208eb7431b4 100644 --- a/src/main/runtime/agent-session-record-store.test.ts +++ b/src/main/runtime/agent-session-record-store.test.ts @@ -10,10 +10,8 @@ import type { } from '../../shared/agent-session-record' import type { AgentSessionProviderHandleLink } from '../../shared/agent-session-provider-handle' import { setStoredAgentSessionHandoffStage } from './agent-session-handoff-record-transitions' -import { - AGENT_SESSION_CLAIM_KEY_RETENTION_MS, - AgentSessionRecordStore -} from './agent-session-record-store' +import { AgentSessionRecordStore } from './agent-session-record-store' +import { AGENT_SESSION_CLAIM_KEY_RETENTION_MS } from './agent-session-claim-key-retention' import { agentSessionStorePath, AGENT_SESSION_STORE_FILE_NAME diff --git a/src/main/runtime/agent-session-record-store.ts b/src/main/runtime/agent-session-record-store.ts index e3655b273fc..a12eafa370e 100644 --- a/src/main/runtime/agent-session-record-store.ts +++ b/src/main/runtime/agent-session-record-store.ts @@ -4,7 +4,9 @@ import { setAgentSessionRecordConversationName } from './agent-session-record-co /** Durable single-writer session records and their operation ledger. */ import { + claimAgentSessionOperation, settleAgentSessionOperation, + type AgentSessionOperationClaim, type AgentSessionOperationDecision, type AgentSessionOperationOutcome, type AgentSessionOperationRow @@ -64,6 +66,10 @@ import { type AgentSessionStoreState } from './agent-session-record-store-file' import { loadProtectedAgentSessionStore } from './agent-session-record-store-security' +import { + isAgentSessionClaimKeyVerifiable, + retireAgentSessionClaimKey +} from './agent-session-claim-key-retention' import { AgentSessionStoreTransactionQueue, markAgentSessionStoreLeasesUnreconciled @@ -71,8 +77,6 @@ import { export const AGENT_SESSION_LEASE_TTL_MS = 30_000, AGENT_SESSION_LEASE_RENEW_INTERVAL_MS = 10_000 -/** Retired claim keys stay verifiable this long so a rotation cannot strand a running agent. */ -export const AGENT_SESSION_CLAIM_KEY_RETENTION_MS = 30 * 24 * 60 * 60 * 1000 export class AgentSessionRecordStore { private constructor(private readonly transactions: AgentSessionStoreTransactionQueue) {} @@ -159,10 +163,8 @@ export class AgentSessionRecordStore { listOperationRows = (): AgentSessionOperationRow[] => [...this.state.operations.values()] - isClaimKeyVerifiable(keyId: string, now: number): boolean { - const retired = this.state.retiredClaimKeys.find((entry) => entry.keyId === keyId) - return !retired || now - retired.retiredAt <= AGENT_SESSION_CLAIM_KEY_RETENTION_MS - } + isClaimKeyVerifiable = (keyId: string, now: number): boolean => + isAgentSessionClaimKeyVerifiable(this.state, keyId, now) /** Spawn tokens observed on the host with no matching lease. Stop them; never adopt them. */ listOrphanSpawnTokens(observedTokens: readonly string[]): string[] { @@ -273,30 +275,39 @@ export class AgentSessionRecordStore { } /** Admits one non-reservation mutation through the durable ledger. */ - async admitOperation( - args: AgentSessionOperationAdmission - ): Promise { - return this.transact(() => { + admitOperation = (args: AgentSessionOperationAdmission): Promise => + this.transact(() => { const admitted = admitAgentSessionOperationRow(this.state.operations, args) this.state.operations = admitted.rows return admitted.decision }) - } /** Send ids stay global after a caller reconnects under a different identity. */ - async admitGlobalOperation( + admitGlobalOperation = ( args: AgentSessionOperationAdmission - ): Promise { - return this.transact(() => { + ): Promise => + this.transact(() => { const admitted = admitAgentSessionGlobalOperationRow(this.state.operations, args) this.state.operations = admitted.rows return admitted.decision }) - } admitMutationOperation = (args: AgentSessionMutationOperationAdmission) => this.transact(() => admitAgentSessionMutationOperation(this.state, args)) + /** Durable compare-and-swap for the right to run an admitted operation's effect: two replays both + * read `pending`, and only a conditional swap tells the one that may run from the one that must + * replay. */ + claimOperation = (args: { + callerKey: string + operationId: string + }): Promise => + this.transact(() => { + const claimed = claimAgentSessionOperation(this.state.operations, args) + this.state.operations = claimed.rows + return claimed.claim + }) + async recordOperationOutcome(args: { callerKey?: string operationId: string @@ -321,14 +332,7 @@ export class AgentSessionRecordStore { this.mutate(args.sessionId, (record) => replaceAgentSessionRecordOptions(record, args)) async retireClaimKey(keyId: string, now: number): Promise { - await this.transact(() => { - if (!this.state.retiredClaimKeys.some((entry) => entry.keyId === keyId)) { - this.state.retiredClaimKeys.push({ keyId, retiredAt: now }) - } - this.state.retiredClaimKeys = this.state.retiredClaimKeys.filter( - (entry) => now - entry.retiredAt <= AGENT_SESSION_CLAIM_KEY_RETENTION_MS - ) - }) + await this.transact(() => retireAgentSessionClaimKey(this.state, keyId, now)) } private async mutate( diff --git a/src/main/runtime/rpc/methods/agent-launch-replay.test.ts b/src/main/runtime/rpc/methods/agent-launch-replay.test.ts new file mode 100644 index 00000000000..6787bcfcf10 --- /dev/null +++ b/src/main/runtime/rpc/methods/agent-launch-replay.test.ts @@ -0,0 +1,603 @@ +/** + * Replay safety for `agent.launch`, against the real durable ledger. + * + * The property under test is narrow and total: one execution per operation, a recorded answer for + * every replay, a truthful refusal when the outcome is unknown. Each guard here has an ablation + * beside it, because a replay test that never watched the unguarded code duplicate is a test of the + * harness rather than of the guard. + */ + +import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { AGENT_LAUNCH_RUNTIME_CAPABILITY } from '../../../../shared/protocol-version' +import { + computeAgentLaunchFingerprint, + deriveAgentLaunchChildOperationId, + type AgentLaunchFingerprintInput +} from '../../../../shared/agent-launch-operation' +import { + agentSessionOperationKey, + claimAgentSessionOperation, + isAgentSessionOperationRow, + settleAgentSessionOperation, + type AgentSessionOperationRow +} from '../../../../shared/agent-session-operation-ledger' +import { AgentSessionRecordStore } from '../../agent-session-record-store' +import { agentSessionStorePath } from '../../agent-session-record-store-file' +import { setStructuredAgentSessionHost } from '../../../native-chat/agent-session-wire/structured-agent-session-registry' +import type { StructuredAgentSessionHost } from '../../../native-chat/agent-session-wire/structured-agent-session-host' +import type { RpcContext } from '../core' +import type { OrcaRuntimeService } from '../../orca-runtime' +import { RpcDispatcher } from '../dispatcher' +import { + methodNamed, + rpcContext, + runtimeStub, + type AgentLaunchRuntimeStub +} from './agent-launch.test-fixture' + +type StructuredCreateReply = + | { ok: true; value: { sessionId: string } } + | { ok: false; refusal: { code: string; message: string } } + +/** Records the operation id the launch handed its inner attach, so the child-id rule is observable + * rather than inferred. */ +const attachOperationIds: string[] = [] +const attachCallerKeys: string[] = [] + +const createStructuredSession = vi.fn( + async (args: { + caller: { callerKey: string } + envelope: { clientOperationId: string } + }): Promise => { + attachOperationIds.push(args.envelope.clientOperationId) + attachCallerKeys.push(args.caller.callerKey) + return { ok: true, value: { sessionId: 'sess-1' } } + } +) + +vi.mock('./structured-agent-session-create', () => ({ + createStructuredAgentSessionForWorktree: (args: { + caller: { callerKey: string } + envelope: { clientOperationId: string } + }) => createStructuredSession(args) +})) + +const { AGENT_LAUNCH_METHODS } = await import('./agent-launch') + +const AGENT_LAUNCH = methodNamed(AGENT_LAUNCH_METHODS, 'agent.launch') + +// Real wall-clock, because the handler admits against `Date.now()`: the ledger refuses an id dated +// far from now in either direction, so a frozen fixture timestamp would only ever test that. +const NOW = Date.now() +const OPERATION_ID = `${NOW}-000000000000000000000000000000aa` + +const PAIRED_CLIENT: Partial = { + clientKind: 'mobile', + pairedDeviceId: 'device-1', + clientId: 'credential-a', + clientCapabilities: [AGENT_LAUNCH_RUNTIME_CAPABILITY] +} +const ROTATED_CREDENTIAL_CLIENT: Partial = { + ...PAIRED_CLIENT, + clientId: 'credential-b', + clientCapabilities: [AGENT_LAUNCH_RUNTIME_CAPABILITY] +} + +let directory: string +let store: AgentSessionRecordStore + +type LaunchParams = AgentLaunchFingerprintInput & { operationId?: string } + +function createLaunch(overrides: Partial = {}): LaunchParams { + return { + agent: 'claude', + target: { kind: 'create-worktree', create: { repo: 'id:repo-1', name: 'task' } }, + ...overrides + } +} + +async function launch( + params: LaunchParams, + runtime: AgentLaunchRuntimeStub, + context: Partial = PAIRED_CLIENT +) { + const parsed = AGENT_LAUNCH.params.safeParse(params) + if (!parsed.success) { + throw new Error(parsed.error.issues[0]?.message ?? 'invalid') + } + return AGENT_LAUNCH.handler(parsed.data, rpcContext(runtime, context)) +} + +function rowFor(operationId: string): AgentSessionOperationRow | undefined { + return store.listOperationRows().find((row) => row.operationId === operationId) +} + +beforeEach(async () => { + attachOperationIds.length = 0 + attachCallerKeys.length = 0 + createStructuredSession.mockClear() + directory = await mkdtemp(join(tmpdir(), 'orca-agent-launch-replay-')) + store = await AgentSessionRecordStore.open({ directory, hostId: 'local' }) + // The launch reaches the ledger through the installed host; nothing else on the host is used, + // because the structured create below it is mocked out. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: `deps.store` is the only member `agent.launch` reads, and a member it omits throws on call. + setStructuredAgentSessionHost({ deps: { store } } as unknown as StructuredAgentSessionHost) +}) + +afterEach(async () => { + setStructuredAgentSessionHost(null) + await rm(directory, { recursive: true, force: true }) +}) + +describe('exactly one execution per launch operation', () => { + it('joins an identical live retry through settlement and conflicts on changed intent', async () => { + const runtime = runtimeStub() + let markStarted: (() => void) | undefined + const started = new Promise((resolve) => { + markStarted = resolve + }) + let releaseEffect: (() => void) | undefined + const effectGate = new Promise((resolve) => { + releaseEffect = resolve + }) + runtime.createManagedWorktree.mockImplementationOnce(async () => { + markStarted?.() + await effectGate + return { worktree: { id: 'wt-new' }, startupTerminal: undefined } + }) + const params = createLaunch({ operationId: OPERATION_ID }) + const first = launch(params, runtime) + await started + const joined = launch(params, runtime) + await expect( + launch(createLaunch({ operationId: OPERATION_ID, agent: 'codex' }), runtime) + ).rejects.toThrow('agent_session_operation_conflict') + releaseEffect?.() + const [firstResult, joinedResult] = await Promise.all([first, joined]) + + expect(runtime.createManagedWorktree).toHaveBeenCalledTimes(1) + expect(createStructuredSession).toHaveBeenCalledTimes(1) + expect(joinedResult).toEqual(firstResult) + }) + + it('the atomic claim admits exactly one winner where the blind settle admitted two', async () => { + await store.admitOperation({ + callerKey: 'device-1', + operationId: OPERATION_ID, + fingerprint: 'fp-1', + now: NOW + }) + + const claims = await Promise.all([ + store.claimOperation({ callerKey: 'device-1', operationId: OPERATION_ID }), + store.claimOperation({ callerKey: 'device-1', operationId: OPERATION_ID }) + ]) + expect(claims.filter((claim) => claim.claim === 'won')).toHaveLength(1) + expect(claims.filter((claim) => claim.claim === 'lost')).toHaveLength(1) + }) +}) + +describe('stable replay identity', () => { + it('replays across a bearer-credential change under the paired device subject', async () => { + const params = createLaunch({ operationId: OPERATION_ID }) + const firstRuntime = runtimeStub() + const first = await launch(params, firstRuntime, PAIRED_CLIENT) + const replayRuntime = runtimeStub() + + await expect(launch(params, replayRuntime, ROTATED_CREDENTIAL_CLIENT)).resolves.toEqual(first) + expect(firstRuntime.createManagedWorktree).toHaveBeenCalledTimes(1) + expect(replayRuntime.createManagedWorktree).not.toHaveBeenCalled() + expect(attachCallerKeys).toEqual(['device-1']) + }) + + it('refuses remote replay safety without a stable paired-device subject', async () => { + const runtime = runtimeStub() + + await expect( + launch(createLaunch({ operationId: OPERATION_ID }), runtime, { + clientKind: 'runtime', + clientId: 'rotating-credential', + clientCapabilities: [AGENT_LAUNCH_RUNTIME_CAPABILITY] + }) + ).rejects.toThrow('agent_session_identity_required') + expect(runtime.ensureStructuredAgentSessionHost).not.toHaveBeenCalled() + expect(runtime.createManagedWorktree).not.toHaveBeenCalled() + expect(store.listOperationRows()).toHaveLength(0) + }) +}) + +describe('a replay answers from the record', () => { + it('returns the whole recorded result rather than recomputing it', async () => { + const runtime = runtimeStub({ createWarning: 'Could not copy untracked files.' }) + const params = createLaunch({ + operationId: OPERATION_ID, + prompt: { text: 'go', delivery: 'draft' } + }) + const first = await launch(params, runtime) + + // The settings that produced the receipt move underneath the replay. A recomputed answer would + // now say the user prefers a terminal; the recorded one still says what actually ran. + const movedSettings = runtimeStub({ + settings: { + experimentalNativeChat: false, + experimentalStructuredNativeChat: false, + openAgentTabsInChatByDefault: false + } + }) + const replayed = await launch(params, movedSettings, PAIRED_CLIENT) + + expect(replayed).toEqual(first) + expect(replayed.receipt.preferred).toBe('structured') + expect(replayed.warning).toBe('Could not copy untracked files.') + expect(replayed.prompt).toEqual({ delivery: 'draft', outcome: 'not-delivered' }) + expect(movedSettings.createManagedWorktree).not.toHaveBeenCalled() + }) + + it('survives a host restart, because the record is on disk', async () => { + const runtime = runtimeStub() + const params = createLaunch({ operationId: OPERATION_ID }) + const first = await launch(params, runtime) + + const reopened = await AgentSessionRecordStore.open({ directory, hostId: 'local' }) + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: see the setup above. + setStructuredAgentSessionHost({ + deps: { store: reopened } + } as unknown as StructuredAgentSessionHost) + const afterRestart = runtimeStub() + + expect(await launch(params, afterRestart)).toEqual(first) + expect(afterRestart.createManagedWorktree).not.toHaveBeenCalled() + }) + + it('refuses a retry that changed what it asks for, and creates nothing', async () => { + const runtime = runtimeStub() + await launch(createLaunch({ operationId: OPERATION_ID }), runtime) + + const conflicting = runtimeStub() + await expect( + launch( + createLaunch({ + operationId: OPERATION_ID, + target: { kind: 'create-worktree', create: { repo: 'id:repo-1', name: 'other' } } + }), + conflicting + ) + ).rejects.toThrow('agent_session_operation_conflict') + expect(conflicting.createManagedWorktree).not.toHaveBeenCalled() + expect(conflicting.createTerminal).not.toHaveBeenCalled() + }) +}) + +describe('an uncertain launch stays uncertain', () => { + it('refuses an operation left at unknown, and never falls back to launching again', async () => { + const params = createLaunch({ operationId: OPERATION_ID }) + await store.admitOperation({ + callerKey: 'device-1', + operationId: OPERATION_ID, + // The host's own digest, so the retry passes the fingerprint check and is refused for the + // reason under test rather than for disagreeing about what it asked for. + fingerprint: computeAgentLaunchFingerprint(params), + now: NOW + }) + await store.claimOperation({ callerKey: 'device-1', operationId: OPERATION_ID }) + + const runtime = runtimeStub() + await expect(launch(params, runtime)).rejects.toThrow('agent_session_operation_unknown') + expect(runtime.createManagedWorktree).not.toHaveBeenCalled() + expect(runtime.createTerminal).not.toHaveBeenCalled() + expect(createStructuredSession).not.toHaveBeenCalled() + }) + + it('leaves the row at unknown when the launch itself throws past the claim', async () => { + const runtime = runtimeStub() + runtime.createManagedWorktree.mockRejectedValueOnce(new Error('worktree_create_failed')) + + await expect(launch(createLaunch({ operationId: OPERATION_ID }), runtime)).rejects.toThrow( + 'worktree_create_failed' + ) + expect(rowFor(OPERATION_ID)?.outcome.status).toBe('unknown') + }) + + it('preserves the unknown refusal code through RPC dispatch', async () => { + const params = createLaunch({ operationId: OPERATION_ID }) + await store.admitOperation({ + callerKey: 'trusted-local:runtime', + operationId: OPERATION_ID, + fingerprint: computeAgentLaunchFingerprint(params), + now: NOW + }) + await store.claimOperation({ + callerKey: 'trusted-local:runtime', + operationId: OPERATION_ID + }) + const runtime = { ...runtimeStub(), getRuntimeId: () => 'runtime-1' } + const dispatcher = new RpcDispatcher({ + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the fixture implements every runtime method reached by agent.launch and dispatcher metadata. + runtime: runtime as unknown as OrcaRuntimeService, + methods: AGENT_LAUNCH_METHODS + }) + + const response = await dispatcher.dispatch({ + id: 'request-1', + authToken: 'token', + method: 'agent.launch', + params + }) + + expect(response).toMatchObject({ + ok: false, + error: { + code: 'agent_session_operation_unknown', + message: 'agent_session_operation_unknown' + } + }) + expect(runtime.createManagedWorktree).not.toHaveBeenCalled() + }) + + it('records a failure that happened before anything could be created', async () => { + const runtime = runtimeStub() + runtime.showManagedTerminalWorkspace.mockRejectedValueOnce(new Error('worktree_not_found')) + + await expect( + launch( + createLaunch({ operationId: OPERATION_ID, target: { kind: 'existing', worktree: 'gone' } }), + runtime + ) + ).rejects.toThrow('worktree_not_found') + expect(rowFor(OPERATION_ID)?.outcome).toMatchObject({ + status: 'failed', + code: 'worktree_not_found' + }) + }) +}) + +describe('settlement is monotone', () => { + it('does not let a late unknown clobber a recorded success', () => { + const succeeded: AgentSessionOperationRow = { + callerKey: 'device-1', + operationId: OPERATION_ID, + fingerprint: 'fp-1', + operationTimestamp: NOW, + recordedAt: NOW, + expiresAt: NOW + 1, + outcome: { status: 'succeeded', sessionId: 'sess-1' } + } + const rows = new Map([[agentSessionOperationKey('device-1', OPERATION_ID), succeeded]]) + + const settled = settleAgentSessionOperation(rows, { + callerKey: 'device-1', + operationId: OPERATION_ID, + outcome: { status: 'unknown' } + }) + + expect([...settled.values()][0].outcome).toEqual({ status: 'succeeded', sessionId: 'sess-1' }) + }) + + it('still lets a claim take a pending row, which is the one state it may take', () => { + const pending: AgentSessionOperationRow = { + callerKey: 'device-1', + operationId: OPERATION_ID, + fingerprint: 'fp-1', + operationTimestamp: NOW, + recordedAt: NOW, + expiresAt: NOW + 1, + outcome: { status: 'pending' } + } + const rows = new Map([[agentSessionOperationKey('device-1', OPERATION_ID), pending]]) + + const claimed = claimAgentSessionOperation(rows, { + callerKey: 'device-1', + operationId: OPERATION_ID + }) + + expect(claimed.claim.claim).toBe('won') + expect([...claimed.rows.values()][0].outcome).toEqual({ status: 'unknown' }) + }) +}) + +describe('the recorded row stays readable by a build that predates it', () => { + it('writes a launch success as a succeeded row with a string sessionId', async () => { + // A terminal launch: the surface has a handle and no session id, which is the case that would + // tempt a new outcome status or an optional field. + await launch( + createLaunch({ + agent: 'codex', + operationId: OPERATION_ID, + target: { kind: 'existing', worktree: 'id:wt-7' } + }), + runtimeStub({ createSupport: { supported: false, reason: 'agent' } }) + ) + + const file: { operations: Record }> } = JSON.parse( + await readFile(agentSessionStorePath(directory), 'utf-8') + ) + const outcome = Object.values(file.operations)[0].outcome + + // The ratchet, and the reason this is not a new status arm or an optional `sessionId`: a build + // without `launch` validates a row by these two fields, one row it rejects returns null for the + // whole file, and the schema version cannot be bumped to excuse it — a store is unreadable to + // any build whose version is higher than the file's. A downgrade must skip what it cannot + // understand, not lose every lease. + expect(outcome.status).toBe('succeeded') + expect(typeof outcome.sessionId).toBe('string') + expect(outcome.launch).toMatchObject({ outcome: { kind: 'terminal' } }) + }) +}) + +describe('an unreadable launch payload costs one replay, never the store', () => { + /** The whole file, primary and backup: `loadAgentSessionStore` falls through to the backup, and + * the backup is a copy of the validated primary, so both carry the same payload in real life. */ + async function rewriteRecordedLaunch(payload: unknown): Promise { + const path = agentSessionStorePath(directory) + const file: { operations: Record }> } = JSON.parse( + await readFile(path, 'utf-8') + ) + const row = Object.values(file.operations)[0] + row.outcome.launch = payload + const written = JSON.stringify(file) + await writeFile(path, written) + await writeFile(`${path}.bak`, written) + } + + it('still admits the row, because one rejected row makes the whole file unparseable', () => { + // The ratchet. `isAgentLaunchResult` mirrors a result type by hand, so a field tightened there + // would reject rows this same build wrote — and a primary and backup that both fail to parse + // raise `agent_session_store_corrupt`, taking every lease in the profile with them. + expect( + isAgentSessionOperationRow({ + callerKey: 'device-1', + operationId: OPERATION_ID, + fingerprint: 'fp-1', + operationTimestamp: NOW, + recordedAt: NOW, + expiresAt: NOW + 1, + outcome: { status: 'succeeded', sessionId: 'sess-1', launch: { not: 'a launch result' } } + }) + ).toBe(true) + }) + + it('reopens the store and refuses only the operation whose payload it cannot read', async () => { + const runtime = runtimeStub() + const params = createLaunch({ operationId: OPERATION_ID }) + await launch(params, runtime) + await rewriteRecordedLaunch({ outcome: { kind: 'structured' }, worktreeId: 'wt-1' }) + + const reopened = await AgentSessionRecordStore.open({ directory, hostId: 'local' }) + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: see the setup above. + setStructuredAgentSessionHost({ + deps: { store: reopened } + } as unknown as StructuredAgentSessionHost) + + expect(reopened.listOperationRows()).toHaveLength(1) + const retry = runtimeStub() + await expect(launch(params, retry)).rejects.toThrow('agent_session_operation_unknown') + expect(retry.createManagedWorktree).not.toHaveBeenCalled() + }) +}) + +describe('a recorded failure replays as the failure it was', () => { + it('answers with the code the launch actually raised, not the ledger vocabulary', async () => { + const runtime = runtimeStub() + runtime.showManagedTerminalWorkspace.mockRejectedValue(new Error('worktree_not_found')) + const params = createLaunch({ + operationId: OPERATION_ID, + target: { kind: 'existing', worktree: 'gone' } + }) + await expect(launch(params, runtime)).rejects.toThrow('worktree_not_found') + + // `worktree_not_found` is not in AGENT_SESSION_WIRE_REFUSAL_CODES. Narrowing the recorded code + // through that closed list answers `agent_session_operation_invalid` — the ledger's "your id is + // malformed" signal, which tells a client to mint a fresh id when the truthful answer is that + // this launch definitively did not run. + const replayed = runtimeStub() + replayed.showManagedTerminalWorkspace.mockRejectedValue(new Error('worktree_not_found')) + await expect(launch(params, replayed)).rejects.toThrow('worktree_not_found') + expect(replayed.showManagedTerminalWorkspace).not.toHaveBeenCalled() + }) + + it('bounds the code it persists, because a code is an identifier and a message is not', async () => { + const runtime = runtimeStub() + runtime.showManagedTerminalWorkspace.mockRejectedValue( + new Error(`ENOENT: no such file or directory, stat '${'/very/long/path'.repeat(400)}'`) + ) + + await expect( + launch( + createLaunch({ operationId: OPERATION_ID, target: { kind: 'existing', worktree: 'gone' } }), + runtime + ) + ).rejects.toThrow('ENOENT') + + const outcome = rowFor(OPERATION_ID)?.outcome + if (outcome?.status !== 'failed') { + throw new Error('the pre-execution failure must record a failed row') + } + expect(outcome.code.length).toBeLessThanOrEqual(128) + }) +}) + +describe('a client that names no operation keeps today behaviour', () => { + it('runs the launch and writes no ledger row at all', async () => { + const runtime = runtimeStub() + await launch(createLaunch(), runtime) + + expect(runtime.createManagedWorktree).toHaveBeenCalledTimes(1) + expect(store.listOperationRows()).toHaveLength(0) + }) + + it('still dedupes a repeated create through the in-memory mutation-id cache', async () => { + const runtime = runtimeStub() + const params = createLaunch({ + target: { + kind: 'create-worktree', + create: { repo: 'id:repo-1', name: 'task', clientMutationId: 'launch-1' } + } + }) + + const [first, second] = await Promise.all([launch(params, runtime), launch(params, runtime)]) + + expect(first).toEqual(second) + expect(runtime.createManagedWorktree).toHaveBeenCalledTimes(1) + expect(store.listOperationRows()).toHaveLength(0) + }) +}) + +describe('the inner attach reserves under its own id', () => { + it('does not conflict with its own launch when both share a caller key', async () => { + const runtime = runtimeStub() + const params = createLaunch({ + operationId: OPERATION_ID, + target: { kind: 'existing', worktree: 'id:wt-7' } + }) + + const result = await launch(params, runtime, PAIRED_CLIENT) + + expect(result.outcome).toEqual({ + kind: 'structured', + sessionId: 'sess-1', + handle: expect.any(String) + }) + expect(attachOperationIds).toHaveLength(1) + expect(attachOperationIds[0]).not.toBe(OPERATION_ID) + expect(attachOperationIds[0]).toBe(deriveAgentLaunchChildOperationId(OPERATION_ID)) + expect(attachCallerKeys).toEqual(['device-1']) + }) + + it('what forwarding the launch id unchanged would do: the attach refuses a conflict', async () => { + const callerKey = 'client-9' + await store.admitOperation({ + callerKey, + operationId: OPERATION_ID, + fingerprint: 'launch-fingerprint', + now: NOW + }) + + // What the attach does with whatever id it is handed: reserve in the same ledger, under the + // same caller, with its own attach fingerprint. + const forwarded = await store.admitOperation({ + callerKey, + operationId: OPERATION_ID, + fingerprint: 'attach-fingerprint', + now: NOW + }) + expect(forwarded).toEqual({ + decision: 'refused', + code: 'agent_session_operation_conflict' + }) + + const derived = deriveAgentLaunchChildOperationId(OPERATION_ID) + if (derived === null) { + throw new Error('the launch id must derive a child id') + } + const child = await store.admitOperation({ + callerKey, + operationId: derived, + fingerprint: 'attach-fingerprint', + now: NOW + }) + expect(child.decision).toBe('admit') + }) +}) diff --git a/src/main/runtime/rpc/methods/agent-launch-replay.ts b/src/main/runtime/rpc/methods/agent-launch-replay.ts new file mode 100644 index 00000000000..223d4a56c9e --- /dev/null +++ b/src/main/runtime/rpc/methods/agent-launch-replay.ts @@ -0,0 +1,205 @@ +/** + * Durable admission for `agent.launch`. + * + * The contract this enforces is three sentences: an operation runs at most once, a replay returns + * the recorded answer, and an operation whose outcome is unknown is refused. Everything else a lost + * launch might want — finding the workspace a dead attempt left behind, adopting a half-created + * session, finishing an interrupted publication — is recovery, and none of it is here. Recovery + * makes a stranded user whole; this makes a retry harmless, and the two are bought separately. + * + * The order is the inverse of what the handler did before. Admission comes first, ahead of + * resolving the caller's worktree selector, because a selector resolution is a live precondition + * and a replay must not be able to fail on one: an operation that already ran has an answer, and + * re-deciding it against today's world is how a recorded success becomes a fresh refusal the client + * then retries as a second effect. `admitAgentSessionMutation` puts the ledger ahead of the lease + * and the fence for that same reason. + */ + +import { deriveAgentLaunchChildOperationId } from '../../../../shared/agent-launch-operation' +import { isAgentLaunchResult, type AgentLaunchResult } from '../../../../shared/agent-launch-intent' +import type { + AgentSessionOperationOutcome, + AgentSessionOperationRefusalCode +} from '../../../../shared/agent-session-operation-ledger' +import { resolveAgentSessionReplayOutcome } from '../../../native-chat/agent-session-wire/structured-agent-session-replay-outcome' +import { getStructuredAgentSessionHost } from '../../../native-chat/agent-session-wire/structured-agent-session-registry' +import type { AgentSessionRecordStore } from '../../agent-session-record-store' +import type { RpcContext } from '../core' +import type { AgentLaunchParams } from './agent-launch-schemas' + +/** Remote replay needs the paired-device subject because its bearer credential can rotate. */ +export function agentLaunchOperationCallerKey( + context: Pick +): string { + if (context.clientKind === undefined) { + return 'trusted-local:runtime' + } + const pairedDeviceId = context.pairedDeviceId?.trim() + if (!pairedDeviceId) { + throw new Error('agent_session_identity_required') + } + return pairedDeviceId +} + +/** + * The store is owned by the structured session host, so reaching it installs that host — already + * true of any structured launch, which installs it to attach. The change is that a terminal-bound + * launch now opens the record store too, when and only when its caller asked for replay safety. + */ +async function requireLaunchOperationStore(context: RpcContext): Promise { + await context.runtime.ensureStructuredAgentSessionHost() + const host = getStructuredAgentSessionHost() + if (!host) { + throw new Error('structured_agent_session_unsupported') + } + return host.deps.store +} + +/** + * `agent.launch` raises its refusals as the thrown code, the way the method's own guards do, so a + * refusal here carries whatever code the operation recorded rather than the closed `agentSession.*` + * envelope. An `AgentSessionWireRefusal` still fits, which is how the shared replay resolver's + * answers pass through unchanged. + */ +export type AgentLaunchRefusal = { code: string; message: string } + +export type AgentLaunchAdmission = + /** This caller owns the operation. It alone runs the effect, and it must settle the row. */ + | { + decision: 'execute' + settle: (result: AgentLaunchResult) => Promise + fail: (code: string) => Promise + /** Distinct from the launch id: the inner attach reserves in this same ledger. */ + attachOperationId: string + callerKey: string + } + /** Already run under this id; hand back what it produced rather than producing it again. */ + | { decision: 'replay'; result: AgentLaunchResult } + | { decision: 'refuse'; refusal: AgentLaunchRefusal } + +/** A recorded row read back as an answer. `pending` is the one state with no answer yet — nobody + * has claimed it — so it reports `rerun`, and the caller goes on to try the claim. */ +function answerFromRecordedRow( + operationId: string, + outcome: AgentSessionOperationOutcome +): AgentLaunchAdmission | null { + if (outcome.status === 'failed') { + // Replayed verbatim rather than narrowed to the `agentSession.*` vocabulary. A launch fails + // with its own codes — `worktree_not_found` and the reuse-terminal guards — none of which is on + // that closed list, so narrowing would answer every one of them with + // `agent_session_operation_invalid`: the ledger's "your id is malformed" signal. The original + // code says this launch definitively failed; the same id replays that answer, while a deliberate + // new attempt must use a fresh id. + return { + decision: 'refuse', + refusal: { + code: outcome.code, + message: + outcome.message ?? `Launch operation ${operationId} already failed: ${outcome.code}.` + } + } + } + const replay = resolveAgentSessionReplayOutcome({ + operationId, + outcome, + // Narrowed here rather than in the row validator: a launch payload this build cannot read must + // cost this one replay, not the whole store. `isAgentSessionOperationRow` says why. + reconstruct: () => + outcome.status === 'succeeded' && isAgentLaunchResult(outcome.launch) ? outcome.launch : null + }) + if (replay.decision === 'rerun') { + return null + } + return replay.decision === 'replay' + ? { decision: 'replay', result: replay.value } + : { decision: 'refuse', refusal: replay.refusal } +} + +/** + * Admit, then claim. + * + * Two steps because they answer different questions — "is this id known and consistent?" and "may + * *I* run it?" — and the second cannot be folded into the first. Admission hands two concurrent + * replays the same `pending` row; only a conditional swap can tell the one that may run from the + * one that must replay. + */ +export async function admitAgentLaunchOperation( + context: RpcContext, + params: AgentLaunchParams & { operationId: string }, + fingerprint: string, + now: number = Date.now() +): Promise { + const operationId = params.operationId + const attachOperationId = deriveAgentLaunchChildOperationId(operationId) + if (!attachOperationId) { + return refusal(operationId, 'agent_session_operation_invalid', 'is not a durable operation id') + } + const store = await requireLaunchOperationStore(context) + const callerKey = agentLaunchOperationCallerKey(context) + const admitted = await store.admitOperation({ + callerKey, + operationId, + fingerprint, + now + }) + if (admitted.decision === 'refused') { + return refusal(operationId, admitted.code, `was refused: ${admitted.code}`) + } + if (admitted.decision === 'replay') { + const answer = answerFromRecordedRow(operationId, admitted.row.outcome) + if (answer) { + return answer + } + } + const claim = await store.claimOperation({ callerKey, operationId }) + if (claim.claim === 'lost') { + // The handler joins same-process retries before admission. Reaching a claimed row here means + // this runtime did not start it, so treating it as restart uncertainty is the safe answer. + return ( + answerFromRecordedRow(operationId, claim.row.outcome) ?? + refusal(operationId, 'agent_session_operation_unknown', 'is claimed but unsettled') + ) + } + if (claim.claim === 'absent') { + // Admitted a moment ago and gone already: the row cannot be re-admitted without reopening the + // duplicate-spawn window it exists to close, so this stays uncertain. + return refusal( + operationId, + 'agent_session_operation_unknown', + 'was pruned between admission and its claim; its outcome is unknown' + ) + } + return { + decision: 'execute', + attachOperationId, + callerKey, + settle: (result) => + store.recordOperationOutcome({ + callerKey, + operationId, + outcome: { + status: 'succeeded', + // A terminal surface has a handle, not a session id; `launch` carries whichever it is. + sessionId: result.outcome.kind === 'structured' ? result.outcome.sessionId : '', + launch: result + } + }), + fail: (code) => + store.recordOperationOutcome({ + callerKey, + operationId, + outcome: { status: 'failed', code } + }) + } +} + +function refusal( + operationId: string, + code: AgentSessionOperationRefusalCode | 'agent_session_operation_unknown', + detail: string +): AgentLaunchAdmission { + return { + decision: 'refuse', + refusal: { code, message: `Launch operation ${operationId} ${detail}.` } + } +} diff --git a/src/main/runtime/rpc/methods/agent-launch-surfaces.ts b/src/main/runtime/rpc/methods/agent-launch-surfaces.ts index 8d60944b261..ad5cb7c4c86 100644 --- a/src/main/runtime/rpc/methods/agent-launch-surfaces.ts +++ b/src/main/runtime/rpc/methods/agent-launch-surfaces.ts @@ -22,7 +22,12 @@ import type { RpcContext } from '../core' import { structuredCallerFor } from './structured-agent-session-gate' import { createStructuredAgentSessionForWorktree } from './structured-agent-session-create' -export function agentLaunchSurfaceFactory(context: RpcContext): AgentLaunchSurfaceFactory { +/** Replay-safe launches keep the nested attach in the same stable caller namespace as the launch. */ +export function agentLaunchSurfaceFactory( + context: RpcContext, + attachOperationId?: string, + operationCallerKey?: string +): AgentLaunchSurfaceFactory { return { createStructuredSession: async ({ worktreeId, agent, options }) => { const sessionId = randomUUID() @@ -33,10 +38,13 @@ export function agentLaunchSurfaceFactory(context: RpcContext): AgentLaunchSurfa await context.runtime.ensureStructuredAgentSessionHost() return requireInstalledHost() }, - caller: structuredCallerFor(context), + caller: operationCallerKey + ? { callerKey: operationCallerKey } + : structuredCallerFor(context), envelope: { sessionId, - clientOperationId: createStructuredAgentSessionOperationId(randomUUID), + clientOperationId: + attachOperationId ?? createStructuredAgentSessionOperationId(randomUUID), expectedRuntimeFence: null, // Overwritten by `prepare` with the host's own attach fingerprint. The create-intent // conflict check it would otherwise feed guards a replayed client operation id, and this diff --git a/src/main/runtime/rpc/methods/agent-launch.test-fixture.ts b/src/main/runtime/rpc/methods/agent-launch.test-fixture.ts new file mode 100644 index 00000000000..7272b5bb7af --- /dev/null +++ b/src/main/runtime/rpc/methods/agent-launch.test-fixture.ts @@ -0,0 +1,109 @@ +/** + * The runtime surface `agent.launch` reaches, and nothing else. + * + * Shared by the RPC-boundary tests and the replay-safety tests so both drive the same host: a stub + * that diverges between them would let one file prove something the other's launch never does. + */ + +import { vi } from 'vitest' +import { AGENT_LAUNCH_RUNTIME_CAPABILITY } from '../../../../shared/protocol-version' +import type { RpcContext } from '../core' + +export const STRUCTURED_PREFERENCE = { + experimentalNativeChat: true, + experimentalStructuredNativeChat: true, + openAgentTabsInChatByDefault: true +} + +export type AgentLaunchRuntimeStubOptions = { + settings?: Record + createSupport?: { supported: boolean; reason?: 'agent' | 'remote' | 'wsl' } + setupReceipt?: { + startupPolicy: 'start-immediately' | 'wait-for-setup' + state: 'running' | 'skipped' | 'not_configured' | 'spawn_failed' + terminalHandle?: string + } + /** What `createManagedWorktree` reports when the workspace exists but is incomplete. */ + createWarning?: string + /** What `createTerminal` reports when the surface itself came up degraded. */ + terminalWarning?: string +} + +export function runtimeStub(options: AgentLaunchRuntimeStubOptions = {}) { + const worktreeCreateResults = new Map>() + const waitForSetupTerminalCompletion = vi.fn( + async (_handle: string, _signal?: AbortSignal): Promise<{ exitCode: number | null }> => ({ + exitCode: 0 + }) + ) + return { + getClientSettings: vi.fn(() => options.settings ?? STRUCTURED_PREFERENCE), + getStructuredAgentSessionCreateSupport: vi.fn( + async () => options.createSupport ?? { supported: true } + ), + dedupeWorktreeCreate: vi.fn( + (repo: string, key: string | undefined, run: () => Promise) => { + if (!key) { + return run() + } + const compositeKey = `${repo}\0${key}` + const existing = worktreeCreateResults.get(compositeKey) + if (existing) { + return existing + } + const result = run() + worktreeCreateResults.set(compositeKey, result) + void result.catch(() => worktreeCreateResults.delete(compositeKey)) + return result + } + ), + showRepo: vi.fn(async () => ({ id: 'repo-1' })), + createManagedWorktree: vi.fn(async (args: Record) => ({ + worktree: { id: 'wt-new' }, + startupTerminal: args.startupAgent ? { handle: 'term_agent_first' } : undefined, + ...(options.setupReceipt ? { setupReceipt: options.setupReceipt } : {}), + ...(options.createWarning ? { warning: options.createWarning } : {}) + })), + createTerminal: vi.fn(async () => ({ + handle: 'term_1', + ...(options.terminalWarning ? { warning: options.terminalWarning } : {}) + })), + showTerminal: vi.fn(async (handle: string) => ({ handle, worktreeId: 'wt-7' })), + isTerminalRunningAgent: vi.fn(async () => true), + showManagedTerminalWorkspace: vi.fn(async (selector: string) => ({ + id: selector.replace(/^id:/, '') + })), + ensureStructuredAgentSessionHost: vi.fn(async () => {}), + waitForSetupTerminalCompletion + } +} + +export type AgentLaunchRuntimeStub = ReturnType + +export function methodNamed( + methods: readonly TMethod[], + name: TName +): Extract { + const found = methods.find( + (entry): entry is Extract => entry.name === name + ) + if (!found) { + throw new Error(`missing method ${name}`) + } + return found +} + +// The one call the stub cannot satisfy structurally; every method it does implement is asserted. +export function rpcContext( + runtime: AgentLaunchRuntimeStub, + context: Partial +): RpcContext { + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the stub implements only the runtime surface these methods reach, so a method it omits throws on call rather than reading a wrong value. + return { runtime, ...context } as unknown as RpcContext +} + +export const CAPABLE_CLIENT: Partial = { + clientKind: 'mobile', + pairedDeviceId: 'device-1', + clientCapabilities: [AGENT_LAUNCH_RUNTIME_CAPABILITY] +} diff --git a/src/main/runtime/rpc/methods/agent-launch.test.ts b/src/main/runtime/rpc/methods/agent-launch.test.ts index 11caa1d3fa1..6806726f0d2 100644 --- a/src/main/runtime/rpc/methods/agent-launch.test.ts +++ b/src/main/runtime/rpc/methods/agent-launch.test.ts @@ -11,6 +11,13 @@ import { beforeEach, describe, expect, it, vi } from 'vitest' import { AGENT_LAUNCH_RUNTIME_CAPABILITY } from '../../../../shared/protocol-version' import type { RpcContext } from '../core' +import { + CAPABLE_CLIENT, + methodNamed, + rpcContext, + runtimeStub, + type AgentLaunchRuntimeStub as RuntimeStub +} from './agent-launch.test-fixture' /** The real `createStructuredAgentSessionForWorktree` answers ok-or-refusal. The stub used to * declare only the ok arm, which made the refusal-downgrade path unmodellable. */ @@ -33,102 +40,12 @@ vi.mock('./structured-agent-session-create', () => ({ const { AGENT_LAUNCH_METHODS } = await import('./agent-launch') const { WORKTREE_METHODS } = await import('./worktree') -const STRUCTURED_PREFERENCE = { - experimentalNativeChat: true, - experimentalStructuredNativeChat: true, - openAgentTabsInChatByDefault: true -} - -function runtimeStub( - options: { - settings?: Record - createSupport?: { supported: boolean; reason?: 'agent' | 'remote' | 'wsl' } - setupReceipt?: { - startupPolicy: 'start-immediately' | 'wait-for-setup' - state: 'running' | 'skipped' | 'not_configured' | 'spawn_failed' - terminalHandle?: string - } - /** What `createManagedWorktree` reports when the workspace exists but is incomplete. */ - createWarning?: string - /** What `createTerminal` reports when the surface itself came up degraded. */ - terminalWarning?: string - } = {} -) { - const worktreeCreateResults = new Map>() - const waitForSetupTerminalCompletion = vi.fn( - async (_handle: string, _signal?: AbortSignal): Promise<{ exitCode: number | null }> => ({ - exitCode: 0 - }) - ) - return { - getClientSettings: vi.fn(() => options.settings ?? STRUCTURED_PREFERENCE), - getStructuredAgentSessionCreateSupport: vi.fn( - async () => options.createSupport ?? { supported: true } - ), - dedupeWorktreeCreate: vi.fn( - (repo: string, key: string | undefined, run: () => Promise) => { - if (!key) { - return run() - } - const compositeKey = `${repo}\0${key}` - const existing = worktreeCreateResults.get(compositeKey) - if (existing) { - return existing - } - const result = run() - worktreeCreateResults.set(compositeKey, result) - void result.catch(() => worktreeCreateResults.delete(compositeKey)) - return result - } - ), - showRepo: vi.fn(async () => ({ id: 'repo-1' })), - createManagedWorktree: vi.fn(async (args: Record) => ({ - worktree: { id: 'wt-new' }, - startupTerminal: args.startupAgent ? { handle: 'term_agent_first' } : undefined, - ...(options.setupReceipt ? { setupReceipt: options.setupReceipt } : {}), - ...(options.createWarning ? { warning: options.createWarning } : {}) - })), - createTerminal: vi.fn(async () => ({ - handle: 'term_1', - ...(options.terminalWarning ? { warning: options.terminalWarning } : {}) - })), - showTerminal: vi.fn(async (handle: string) => ({ handle, worktreeId: 'wt-7' })), - isTerminalRunningAgent: vi.fn(async () => true), - showManagedTerminalWorkspace: vi.fn(async (selector: string) => ({ - id: selector.replace(/^id:/, '') - })), - ensureStructuredAgentSessionHost: vi.fn(async () => {}), - waitForSetupTerminalCompletion - } -} - -type RuntimeStub = ReturnType - -function methodNamed( - methods: readonly TMethod[], - name: TName -): Extract { - const found = methods.find( - (entry): entry is Extract => entry.name === name - ) - if (!found) { - throw new Error(`missing method ${name}`) - } - return found -} - const AGENT_LAUNCH = methodNamed(AGENT_LAUNCH_METHODS, 'agent.launch') function parseLaunch(params: unknown) { return AGENT_LAUNCH.params.safeParse(params) } -// The one call the stub cannot satisfy structurally; every method it does implement is asserted. -function rpcContext(runtime: RuntimeStub, context: Partial): RpcContext { - // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the stub implements only the runtime surface these methods reach, so a method it omits throws on call rather than reading a wrong value. - return { runtime, ...context } as unknown as RpcContext -} - function createArgs(runtime: RuntimeStub): Record { const [args] = runtime.createManagedWorktree.mock.calls[0] ?? [] if (!args) { @@ -137,12 +54,6 @@ function createArgs(runtime: RuntimeStub): Record { return args } -const CAPABLE_CLIENT: Partial = { - clientKind: 'mobile', - pairedDeviceId: 'device-1', - clientCapabilities: [AGENT_LAUNCH_RUNTIME_CAPABILITY] -} - async function launch( params: unknown, runtime: RuntimeStub, diff --git a/src/main/runtime/rpc/methods/agent-launch.ts b/src/main/runtime/rpc/methods/agent-launch.ts index 49a8f9e3951..3af6bffa801 100644 --- a/src/main/runtime/rpc/methods/agent-launch.ts +++ b/src/main/runtime/rpc/methods/agent-launch.ts @@ -11,13 +11,26 @@ * * A caller therefore never asks for a mode, and must read `outcome.kind` rather than assume one: * the receipt always says which surface ran and why, so a downgrade is never silent. + * + * A launch is also the one call whose retry is most expensive to get wrong — a lost reply means the + * caller cannot tell "never ran" from "ran, answer lost" — so a caller may name the operation with + * `operationId` and get exactly one execution, a recorded answer on every replay, and a refusal + * when the outcome is genuinely unknown. That guarantee is safety, not recovery: it makes a retry + * harmless, and does nothing to reunite a caller with a surface a dead attempt left behind. */ import { AGENT_LAUNCH_RUNTIME_CAPABILITY } from '../../../../shared/protocol-version' -import type { AgentLaunchIntent, AgentLaunchTarget } from '../../../../shared/agent-launch-intent' +import { computeAgentLaunchFingerprint } from '../../../../shared/agent-launch-operation' +import type { + AgentLaunchIntent, + AgentLaunchResult, + AgentLaunchTarget +} from '../../../../shared/agent-launch-intent' +import { agentSessionOperationKey } from '../../../../shared/agent-session-operation-ledger' import { executeAgentLaunch } from '../../../agent-launch/agent-launch-executor' import type { OrcaRuntimeService } from '../../orca-runtime' import { defineMethod, type RpcContext } from '../core' +import { admitAgentLaunchOperation, agentLaunchOperationCallerKey } from './agent-launch-replay' import { AgentLaunch, type AgentLaunchParams } from './agent-launch-schemas' import { agentLaunchSurfaceFactory } from './agent-launch-surfaces' import { agentLaunchWorkspaceFactory } from './agent-launch-worktree-creation' @@ -86,33 +99,179 @@ async function validateReusedTerminal( } } +/** + * The half before anything is created: resolve the caller's selector, then check a reused terminal. + * A throw from here proves no surface was built, which is what lets the ledger record a launch that + * failed in it as `failed` rather than `unknown`. + */ +async function resolveUnlaunchedIntent( + params: AgentLaunchParams, + runtime: OrcaRuntimeService +): Promise { + const intent = await agentLaunchIntent(params, runtime) + await validateReusedTerminal(intent, runtime) + return intent +} + +function runAgentLaunch( + intent: AgentLaunchIntent, + context: RpcContext, + attachOperationId?: string, + operationCallerKey?: string +): Promise { + return executeAgentLaunch({ + runtime: context.runtime, + intent, + surfaces: agentLaunchSurfaceFactory(context, attachOperationId, operationCallerKey), + workspaces: agentLaunchWorkspaceFactory(context, intent.agent) + }) +} + +/** + * The pre-ledger path, unchanged and kept for every caller that names no operation. + * + * `dedupeWorktreeCreate` is an in-memory 60-second window over the create half of a launch, keyed + * on repo plus mutation id with no caller partition, and it dies with the process. That was the + * only idempotency `agent.launch` ever had, and an existing-workspace launch never got even that. + * It is deliberately NOT a second correctness authority now: once a caller supplies `operationId`, + * durable admission encloses the whole operation and this cache is bypassed entirely, so there is + * one place that decides whether a launch runs. + */ +function runLegacyAgentLaunch( + params: AgentLaunchParams, + context: RpcContext +): Promise { + const execute = async () => + runAgentLaunch(await resolveUnlaunchedIntent(params, context.runtime), context) + if (params.target.kind === 'create-worktree' && params.target.create.clientMutationId) { + return context.runtime.dedupeWorktreeCreate( + params.target.create.repo, + `agent.launch:${params.target.create.clientMutationId}`, + execute + ) + } + return execute() +} + +function settleQuietly(settlement: Promise): Promise { + return settlement.catch((error: unknown) => { + console.warn('[agent-launch] the launch settled, its operation row did not', error) + }) +} + +/** Long enough for every code this path raises, with room for one a later guard adds. */ +const LAUNCH_FAILURE_CODE_MAX_LENGTH = 128 + +/** + * This path raises its refusals as the thrown code, the way the method's own guards do — and the + * recorded code is what a replay answers with, so it is worth keeping. + * + * Bounded because a code is an identifier but `error.message` is free text: an errno sentence + * carrying an absolute path arrives here as one, and it would be written into a ledger file that is + * re-serialized whole on every subsequent operation. Bounded on the way IN only. A length check in + * `isAgentSessionOperationRow` would reject rows this same build wrote, and one rejected row costs + * the entire store. + */ +function agentLaunchFailureCode(error: unknown): string { + const code = error instanceof Error ? error.message : '' + return code.length > 0 ? code.slice(0, LAUNCH_FAILURE_CODE_MAX_LENGTH) : 'agent_launch_failed' +} + +type ActiveAgentLaunch = { + fingerprint: string + promise: Promise +} + +const activeAgentLaunchesByRuntime = new WeakMap< + OrcaRuntimeService, + Map +>() + +function activeAgentLaunchesFor(runtime: OrcaRuntimeService): Map { + const existing = activeAgentLaunchesByRuntime.get(runtime) + if (existing) { + return existing + } + const active = new Map() + activeAgentLaunchesByRuntime.set(runtime, active) + return active +} + +async function executeReplaySafeAgentLaunch( + params: AgentLaunchParams & { operationId: string }, + context: RpcContext, + fingerprint: string +): Promise { + const admission = await admitAgentLaunchOperation(context, params, fingerprint) + if (admission.decision === 'refuse') { + throw new Error(admission.refusal.code) + } + if (admission.decision === 'replay') { + return admission.result + } + let intent: AgentLaunchIntent + try { + intent = await resolveUnlaunchedIntent(params, context.runtime) + } catch (error) { + await settleQuietly(admission.fail(agentLaunchFailureCode(error))) + throw error + } + // Any later failure may follow a created surface, so the claimed row must stay `unknown`. + const result = await runAgentLaunch( + intent, + context, + admission.attachOperationId, + admission.callerKey + ) + // Settlement is bookkeeping; failure leaves the truthful `unknown` refusal for later retries. + await settleQuietly(admission.settle(result)) + return result +} + +function runReplaySafeAgentLaunch( + params: AgentLaunchParams & { operationId: string }, + context: RpcContext +): Promise { + const callerKey = agentLaunchOperationCallerKey(context) + const key = agentSessionOperationKey(callerKey, params.operationId) + const fingerprint = computeAgentLaunchFingerprint(params) + const activeAgentLaunches = activeAgentLaunchesFor(context.runtime) + const active = activeAgentLaunches.get(key) + if (active) { + if (active.fingerprint !== fingerprint) { + return Promise.reject(new Error('agent_session_operation_conflict')) + } + return active.promise + } + + let promise: Promise + promise = executeReplaySafeAgentLaunch(params, context, fingerprint).finally(() => { + if (activeAgentLaunches.get(key)?.promise === promise) { + activeAgentLaunches.delete(key) + } + }) + activeAgentLaunches.set(key, { fingerprint, promise }) + return promise +} + export const AGENT_LAUNCH_METHODS = [ defineMethod({ name: 'agent.launch', params: AgentLaunch, - handler: async (params, context) => { + handler: async (params, context): Promise => { if (!supportsAgentLaunch(context)) { throw new Error('agent_launch_unsupported') } - const intent = await agentLaunchIntent(params, context.runtime) - await validateReusedTerminal(intent, context.runtime) - const execute = () => - executeAgentLaunch({ - runtime: context.runtime, - intent, - surfaces: agentLaunchSurfaceFactory(context), - workspaces: agentLaunchWorkspaceFactory(context, intent.agent) - }) - // Preserve the existing bounded create guard. Complete launch replay needs durable operation - // identity, caller scope and a host-computed payload fingerprint; this cache has none of them. - if (params.target.kind === 'create-worktree' && params.target.create.clientMutationId) { - return context.runtime.dedupeWorktreeCreate( - params.target.create.repo, - `agent.launch:${params.target.create.clientMutationId}`, - execute - ) + if (!params.operationId) { + return runLegacyAgentLaunch(params, context) } - return execute() + return runReplaySafeAgentLaunch( + { + ...params, + operationId: params.operationId + }, + context + ) } }) ] diff --git a/src/shared/agent-launch-intent.ts b/src/shared/agent-launch-intent.ts index 702add77acd..5350ea7684c 100644 --- a/src/shared/agent-launch-intent.ts +++ b/src/shared/agent-launch-intent.ts @@ -134,6 +134,77 @@ export type AgentLaunchModeReceipt = { detail: string } +/** + * Narrows a launch result read back from durable storage. + * + * Lives beside the type rather than in the store so the two cannot drift: a field added above and + * not checked here is a field a replay can hand back unvalidated. Every optional field is checked + * when present and ignored when absent, so a row written by an older host still reads. + */ +export function isAgentLaunchResult(value: unknown): value is AgentLaunchResult { + if (typeof value !== 'object' || value === null) { + return false + } + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: narrowing an unknown for field-by-field validation; every field read below is checked before use. + const result = value as Partial + return ( + isAgentLaunchOutcome(result.outcome) && + typeof result.worktreeId === 'string' && + isAgentLaunchModeReceipt(result.receipt) && + (result.warning === undefined || typeof result.warning === 'string') && + (result.prompt === undefined || isAgentLaunchPromptReceipt(result.prompt)) + ) +} + +function isAgentLaunchPromptReceipt(value: unknown): value is AgentLaunchPromptReceipt { + if (typeof value !== 'object' || value === null) { + return false + } + if (!('delivery' in value) || !isAgentLaunchPromptDelivery(value.delivery)) { + return false + } + if (!('outcome' in value)) { + return false + } + return value.outcome === 'journaled' + ? 'messageId' in value && typeof value.messageId === 'string' + : value.outcome === 'handed-to-terminal' || value.outcome === 'not-delivered' +} + +function isAgentLaunchOutcome(value: unknown): value is AgentLaunchOutcome { + if (typeof value !== 'object' || value === null) { + return false + } + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the assertion claims only that the keys may be present and unknown, which is true of any object. + const outcome = value as { kind?: unknown; handle?: unknown; sessionId?: unknown } + if (typeof outcome.handle !== 'string' || outcome.handle.length === 0) { + return false + } + return outcome.kind === 'terminal' + ? true + : outcome.kind === 'structured' && + typeof outcome.sessionId === 'string' && + outcome.sessionId.length > 0 +} + +function isAgentLaunchModeReceipt(value: unknown): value is AgentLaunchModeReceipt { + if (typeof value !== 'object' || value === null) { + return false + } + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: narrowing an unknown for field-by-field validation; every field read below is checked before use. + const receipt = value as Partial + return ( + (receipt.mode === 'structured' || receipt.mode === 'terminal') && + (receipt.preferred === 'structured' || receipt.preferred === 'terminal') && + typeof receipt.reason === 'string' && + typeof receipt.detail === 'string' + ) +} + +function isAgentLaunchPromptDelivery(value: unknown): value is AgentLaunchPromptDelivery { + return value === 'submit' || value === 'draft' +} + export function agentLaunchTargetIsCreate( target: AgentLaunchTarget ): target is Extract { diff --git a/src/shared/agent-launch-operation.ts b/src/shared/agent-launch-operation.ts new file mode 100644 index 00000000000..607808508f4 --- /dev/null +++ b/src/shared/agent-launch-operation.ts @@ -0,0 +1,76 @@ +/** + * Replay safety for `agent.launch`. + * + * A launch creates a workspace, an agent, or both. When its reply is lost, the caller cannot tell + * "the host never saw it" from "the host ran it and the answer went missing" — and mobile retries a + * lost create by design (`worktree-create-retry.ts`), so the retry is the ordinary case, not the + * edge. Retrying without an operation id is how one tap becomes two agents in two workspaces. + * + * The fix is the ledger `agentSession.*` already runs on: the caller names the operation once, the + * host records that name durably before doing anything, and every later arrival of that name gets + * the recorded answer rather than a second agent. What this module adds is the launch-shaped parts + * of that contract — the digest over what a launch DOES, and the child id its inner attach reserves + * under. + * + * This is safety, not recovery. Nothing here probes for a surface a previous attempt may have left + * behind, adopts one, or finishes an interrupted publication; an operation whose outcome is unknown + * stays unknown and is refused. + */ + +import { canonicalAgentSessionDigest } from './agent-session-mutation-envelope' +import { parseAgentSessionOperationTimestamp } from './agent-session-host-authority' + +/** + * The launch fields that decide what the call does. The operation id itself is excluded — it names + * the operation, it is not part of it — and so is every mutable host setting: the route a launch + * takes depends on the user's default surface, and folding that in would make an honest retry + * conflict merely because the setting moved between the two attempts. + */ +export type AgentLaunchFingerprintInput = { + agent: string + target: + | { kind: 'existing'; worktree: string } + | { kind: 'create-worktree'; create: Readonly> } + prompt?: { text: string; delivery: string } + sessionOptions?: Readonly> + reuseTerminal?: { handle: string } +} + +/** Host-computed, never accepted from the caller: a digest a client supplies is a digest a buggy + * client can make agree with anything. */ +export function computeAgentLaunchFingerprint(input: AgentLaunchFingerprintInput): string { + return canonicalAgentSessionDigest({ + method: 'agent.launch', + agent: input.agent, + target: input.target, + prompt: input.prompt, + sessionOptions: input.sessionOptions, + reuseTerminal: input.reuseTerminal + }) +} + +const OPERATION_ID_ENTROPY_LENGTH = 32 + +/** + * The id the launch's inner `agentSession.attach` reserves under. + * + * The ledger keys a row on `(callerKey, operationId)` with no method in it, and a structured launch + * reserves in that same ledger under the same caller. Forwarding the launch's own id would make the + * attach meet the launch's row, disagree with its fingerprint, and refuse a conflict before + * anything was created. Derived rather than random so the child of a given launch is always the + * same id — a launch is claimed once, and a claim that could name a different child each time would + * be ownership in name only. + * + * The launch's timestamp is kept so the child ages out on the same schedule as its parent. + */ +export function deriveAgentLaunchChildOperationId(operationId: string): string | null { + const timestamp = parseAgentSessionOperationTimestamp(operationId) + if (timestamp === null) { + return null + } + const entropy = canonicalAgentSessionDigest({ + child: 'agent.launch:attach', + operationId + }).slice(0, OPERATION_ID_ENTROPY_LENGTH) + return `${timestamp}-${entropy}` +} diff --git a/src/shared/agent-session-host-authority.ts b/src/shared/agent-session-host-authority.ts index e994c9f5a0e..7139858024d 100644 --- a/src/shared/agent-session-host-authority.ts +++ b/src/shared/agent-session-host-authority.ts @@ -23,6 +23,7 @@ export const AGENT_SESSION_RPC_ERROR_CODES = [ 'agent_session_operation_conflict', 'agent_session_operation_expired', 'agent_session_operation_capacity', + 'agent_session_operation_unknown', 'agent_session_legacy_required', 'execution_owner_reconciling', 'execution_owner_unavailable' diff --git a/src/shared/agent-session-mutation-envelope.ts b/src/shared/agent-session-mutation-envelope.ts index aebd2618522..bc7f5761db7 100644 --- a/src/shared/agent-session-mutation-envelope.ts +++ b/src/shared/agent-session-mutation-envelope.ts @@ -29,12 +29,17 @@ export function computeAgentSessionPayloadFingerprint(input: { sessionId: string fields: Record }): string { - const canonical = canonicalize({ + return canonicalAgentSessionDigest({ method: input.method, sessionId: input.sessionId, fields: input.fields }) - return createHash('sha256').update(canonical).digest('hex') +} + +/** The same digest for an operation that has no session to name — a launch decides which surface it + * gets, so it has no session id until after it runs. */ +export function canonicalAgentSessionDigest(value: Record): string { + return createHash('sha256').update(canonicalize(value)).digest('hex') } function canonicalize(value: unknown): string { diff --git a/src/shared/agent-session-operation-ledger.ts b/src/shared/agent-session-operation-ledger.ts index 1c63ff9d34f..e3d39fdee34 100644 --- a/src/shared/agent-session-operation-ledger.ts +++ b/src/shared/agent-session-operation-ledger.ts @@ -30,9 +30,30 @@ export type AgentSessionOperationOutcome = | { status: 'pending' } | { status: 'succeeded' + /** + * Empty exactly when `launch` recorded a terminal surface — a PTY has a handle, not a session + * id. Kept a required string rather than made optional because a build that predates `launch` + * rejects a `succeeded` row without one, and a single rejected row invalidates the whole + * store on load (`agent-session-record-store-file.ts`). A downgrade must skip what it cannot + * read, not lose every lease in the file. + */ sessionId: string conversationCommand?: AgentSessionConversationCommandResult rewind?: AgentSessionRewindResult + /** + * The full `agent.launch` answer. Recorded whole rather than rebuilt, because the preferred + * mode and the reason a launch downgraded away from it cannot be recomputed once the user's + * settings move: a replay must return what ran, not what would run now. + * + * Typed `unknown`, and deliberately NOT checked by `isAgentSessionOperationRow`, for the same + * reason `sessionId` above stays required: a row this file rejects makes the whole store + * unparseable, and a primary and backup that both fail to parse raise + * `agent_session_store_corrupt` rather than degrading. `isAgentLaunchResult` is a + * hand-maintained mirror of a result type later work will edit, so a field tightened there + * would reject rows this same build wrote and take every lease in the file with them. It is + * narrowed where the value is read instead, where a payload we cannot read costs one replay. + */ + launch?: unknown } | { status: 'failed'; code: string; message?: string; rewindReason?: AgentSessionRewindReason } /** The effect may or may not have happened; replay this answer instead of spawning again. */ @@ -81,13 +102,69 @@ export function settleAgentSessionOperation( return new Map( [...rows].map(([key, row]) => [ key, - (targetKey ? key === targetKey : row.operationId === args.operationId) + (targetKey ? key === targetKey : row.operationId === args.operationId) && + !supersedesSettledOutcome(row.outcome, args.outcome) ? { ...row, outcome: args.outcome } : row ]) ) } +/** + * Settlement is monotone in one direction only: once an operation is known to have succeeded or + * failed, a later `unknown` must not take that certainty away. A crash handler, a restart + * reconciler and the operation's own settle can all reach the same row, and the slowest of them is + * not the best informed — an `unknown` landing after a recorded success would turn a replayable + * answer into a permanent refusal for work that demonstrably completed. + */ +function supersedesSettledOutcome( + current: AgentSessionOperationOutcome, + next: AgentSessionOperationOutcome +): boolean { + return ( + next.status === 'unknown' && (current.status === 'succeeded' || current.status === 'failed') + ) +} + +/** Who owns the right to run this operation's effect. */ +export type AgentSessionOperationClaim = + /** This caller moved the row from `pending`; it alone may run the effect. */ + | { claim: 'won'; row: AgentSessionOperationRow } + /** Someone else already took it. The row says what to answer with. */ + | { claim: 'lost'; row: AgentSessionOperationRow } + /** Pruned or never admitted. */ + | { claim: 'absent' } + +/** + * Take exclusive ownership of an admitted operation, atomically. + * + * Admission alone does not decide who runs: two callers replaying one id both read `pending`, and + * two unconditional writes of `unknown` are not a compare-and-swap — both would see their own write + * land and both would execute. The swap has to be conditional on the state it read, in one step, + * and it has to report which caller won. `pending` is the only state that can be claimed. + * + * The row moves to `unknown` rather than staying `pending` on purpose: from the instant the effect + * may start, the truthful durable answer is "this may have happened", and a host that dies mid-run + * leaves exactly that behind. + */ +export function claimAgentSessionOperation( + rows: ReadonlyMap, + args: { callerKey: string; operationId: string } +): { rows: Map; claim: AgentSessionOperationClaim } { + const key = agentSessionOperationKey(args.callerKey, args.operationId) + const existing = rows.get(key) + if (!existing) { + return { rows: new Map(rows), claim: { claim: 'absent' } } + } + if (existing.outcome.status !== 'pending') { + return { rows: new Map(rows), claim: { claim: 'lost', row: existing } } + } + const claimed: AgentSessionOperationRow = { ...existing, outcome: { status: 'unknown' } } + const next = new Map(rows) + next.set(key, claimed) + return { rows: next, claim: { claim: 'won', row: claimed } } +} + /** * Retention floor. The tombstone must outlive the window in which its id could still be admitted * as new, plus the accepted future skew — otherwise a retry arriving in the gap becomes a second @@ -190,6 +267,7 @@ export function isAgentSessionOperationRow(value: unknown): value is AgentSessio typeof outcome === 'object' && outcome !== null && ((outcome.status === 'pending' && true) || + // `launch` is intentionally absent from this check; see the field's own note above. (outcome.status === 'succeeded' && typeof outcome.sessionId === 'string' && (outcome.rewind === undefined || isAgentSessionRewindResult(outcome.rewind)) && diff --git a/src/shared/protocol-version.ts b/src/shared/protocol-version.ts index 11b834f00da..789cdc23582 100644 --- a/src/shared/protocol-version.ts +++ b/src/shared/protocol-version.ts @@ -253,6 +253,19 @@ export const NOTIFICATIONS_REMOTE_PUSH_RUNTIME_CAPABILITY = 'notifications.remot // v2 makes prompt delivery an outcome union and top-level warnings the only supported shape. export const AGENT_LAUNCH_RUNTIME_CAPABILITY = 'agent.launch.v2' as const +/** + * The host admits `agent.launch` through the durable operation ledger, so a caller that names its + * launch with `operationId` gets exactly one execution and a recorded answer on every retry. + * + * This one is negotiated host-to-client, unlike `agent.launch.v1`, because of how RPC params + * degrade: an older host strips `operationId` as an unknown key and runs the launch anyway, with no + * error. A client that retried on the strength of having sent an id would get a second agent and + * never learn why. So `operationId` is optional on the wire — shipped mobile sends none and keeps + * today's behaviour verbatim — and a client may only treat a retry as safe once the host has + * advertised this. + */ +export const AGENT_LAUNCH_REPLAY_RUNTIME_CAPABILITY = 'agent.launch.replay.v1' as const + // Generic native clients include the CLI and must not claim Electron-only page // placement support. export const NATIVE_REMOTE_RUNTIME_CLIENT_CAPABILITIES = [ @@ -359,7 +372,8 @@ export const RUNTIME_CAPABILITIES = [ AUTOMATION_OWNER_FENCING_RUNTIME_CAPABILITY, AUTOMATION_CREATE_IDEMPOTENCY_RUNTIME_CAPABILITY, NOTIFICATIONS_REMOTE_PUSH_RUNTIME_CAPABILITY, - AGENT_LAUNCH_RUNTIME_CAPABILITY + AGENT_LAUNCH_RUNTIME_CAPABILITY, + AGENT_LAUNCH_REPLAY_RUNTIME_CAPABILITY ] as const export type RuntimeCapability = (typeof RUNTIME_CAPABILITIES)[number] | (string & {}) diff --git a/src/shared/rpc-contract/agent-launch-params.ts b/src/shared/rpc-contract/agent-launch-params.ts index 3bf6ef7a6f8..987b4f6e3f6 100644 --- a/src/shared/rpc-contract/agent-launch-params.ts +++ b/src/shared/rpc-contract/agent-launch-params.ts @@ -12,6 +12,7 @@ */ import { z } from 'zod' +import { parseAgentSessionOperationTimestamp } from '../agent-session-host-authority' import { isTuiAgent } from '../tui-agent-config' import type { TuiAgent } from '../tui-agent' import { WorktreeCreate } from './worktree-create-params' @@ -28,6 +29,21 @@ const LaunchAgent = z export const AgentLaunch = z.object({ agent: LaunchAgent, + /** + * Names this launch so a retry replays instead of starting a second agent. + * + * Optional, and optional forever: shipped mobile sends none, and a host that required one would + * refuse every live client. Its absence is not a silent downgrade to a weaker guarantee — it is + * the caller declining the guarantee, and the host must never mint an id on a caller's behalf + * after an ambiguous launch, because an id minted on the retry is a brand new operation. + */ + operationId: z + .string() + .refine( + (value) => parseAgentSessionOperationTimestamp(value) !== null, + 'Malformed launch operation id' + ) + .optional(), target: z.discriminatedUnion('kind', [ z.object({ kind: z.literal('existing'),