From 0bf815a48063eb4ef2dff1f363f108ee7c17cc4f Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Wed, 16 Sep 2026 18:19:02 -0700 Subject: [PATCH] fix(agent-launch): make a lost launch safe to retry (#21106) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(agent-launch): make a lost launch safe to retry `agent.launch` could not be retried safely. Only a create-worktree target carrying a clientMutationId got any idempotency at all, and that was a 60s in-memory cache with no caller partition that dies with the process; an existing-workspace launch got none. Mobile retries a lost create by design, so the retry is the ordinary case — and a retry past that cache meant a second worktree and a second agent. A caller may now name its launch with an optional `operationId` and get one execution, the recorded answer on every replay, and a truthful refusal when the outcome is unknown. Admission runs before the worktree selector is resolved, so a replay answers from the record rather than re-deciding against today's world. The core is an atomic claim. Admission alone cannot decide who runs: two replays both read `pending`, and settling `unknown` replaces the outcome blind, so two serialized writes are not a compare-and-swap and both callers execute. A conditional current-state swap now reports which caller won, and settlement is monotone so a late `unknown` cannot erase a recorded success. Also here: a host-computed fingerprint over the launch intent that excludes mutable settings, the full launch result persisted so a replay returns the receipt and warning that cannot be recomputed once settings move, and a derived child operation id for the inner attach — the ledger key carries no method, so forwarding the launch id would make the attach conflict with its own launch. Safety, not recovery. Nothing here probes for a surface a dead attempt left behind, adopts one, or finishes an interrupted publication. Callers that send no `operationId` keep today's behaviour exactly, which is why the field is optional and the host advertises `agent.launch.replay.v1`: an older host strips an unknown param and launches anyway, so a client may only treat a retry as safe once the host has said it enforces the ledger. * fix(agent-launch): keep an unreadable launch payload from costing the store Review follow-ups on the replay-safety ledger. A recorded `launch` payload must not gate row validity. `isAgentLaunchResult` is a hand-maintained mirror of a result type later work will edit, and `isAgentSessionOperationRow` is consulted by the store loader, where one rejected row makes the whole file unparseable — a primary and backup that both fail to parse raise `agent_session_store_corrupt` and the profile loses every lease. That is the same argument the row already makes for keeping `sessionId` required, applied to the field this PR added. The payload is now typed `unknown`, left out of the row guard, and narrowed where it is read, so a payload this build cannot read refuses exactly one replay. A recorded failure now replays as the code the launch raised. Narrowing it through the closed `agentSession.*` refusal list answered `worktree_not_found` with `agent_session_operation_invalid` — the ledger's "your id is malformed" signal, which invites a client to mint a fresh id when the truthful answer is that this launch definitively did not run and the same id is safe to retry. The persisted failure code is bounded on the way in. A code is an identifier, but `error.message` is free text: an errno sentence carrying an absolute path arrived here as one and was written into a file re-serialized whole on every later operation. Bounded on write only — a length check in the row validator would reject rows this same build wrote, which is the hazard above. Comments: the caller key does not give one client a single namespace across surfaces, because the structured attach this launch performs partitions under `structuredCallerFor`; the two coincide only for a bearer-identity caller with no paired device, which is exactly when the derived child id is load-bearing. Recorded as a known limit that a `lost` claim cannot tell a sibling executing now from one a restart abandoned; telling them apart needs execution-generation tagging, which is recovery. Tests: the store-level ablation was inert — it defined a local stand-in and passed identically with and without the guard. It now substitutes the non-atomic composition into the handler's own store and watches one tap create two workspaces. Each of the four new guards was watched failing against the unfixed code: `agent_session_store_corrupt` on reopen, `expected false to be true` on the row guard, `agent_session_operation_invalid` in place of `worktree_not_found`, and a 6042-character code where 128 is the bound. * fix(agent-launch): keep live retries in one execution * docs(agent-launch): clarify failed replay guidance --- .../agent-session-claim-key-retention.ts | 29 + .../agent-session-record-store.test.ts | 6 +- .../runtime/agent-session-record-store.ts | 50 +- .../rpc/methods/agent-launch-replay.test.ts | 603 ++++++++++++++++++ .../rpc/methods/agent-launch-replay.ts | 205 ++++++ .../rpc/methods/agent-launch-surfaces.ts | 14 +- .../rpc/methods/agent-launch.test-fixture.ts | 109 ++++ .../runtime/rpc/methods/agent-launch.test.ts | 103 +-- src/main/runtime/rpc/methods/agent-launch.ts | 199 +++++- src/shared/agent-launch-intent.ts | 71 +++ src/shared/agent-launch-operation.ts | 76 +++ src/shared/agent-session-host-authority.ts | 1 + src/shared/agent-session-mutation-envelope.ts | 9 +- src/shared/agent-session-operation-ledger.ts | 80 ++- src/shared/protocol-version.ts | 16 +- .../rpc-contract/agent-launch-params.ts | 16 + 16 files changed, 1437 insertions(+), 150 deletions(-) create mode 100644 src/main/runtime/agent-session-claim-key-retention.ts create mode 100644 src/main/runtime/rpc/methods/agent-launch-replay.test.ts create mode 100644 src/main/runtime/rpc/methods/agent-launch-replay.ts create mode 100644 src/main/runtime/rpc/methods/agent-launch.test-fixture.ts create mode 100644 src/shared/agent-launch-operation.ts 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'),