diff --git a/config/packaged-runtime-node-modules.cjs b/config/packaged-runtime-node-modules.cjs index 71ca780d1b8..83c83c33bfc 100644 --- a/config/packaged-runtime-node-modules.cjs +++ b/config/packaged-runtime-node-modules.cjs @@ -22,6 +22,7 @@ const PACKAGED_RUNTIME_PACKAGE_ROOTS = [ 'jsonc-parser', 'node-pty', 'posthog-node', + 'proper-lockfile', // serve-sim (for CLI JS entry + closure + state/middleware + to make packaged require('serve-sim') + its internal relatives work; mirrors other runtime JS like ws/yaml/zod. Natives/dylibs still via extraResources + the node_modules/serve-sim copy in resources from builder. Client if added too. 'serve-sim', 'qrcode', diff --git a/config/scripts/electron-builder-config.test.mjs b/config/scripts/electron-builder-config.test.mjs index 3227d957b24..73c51faa250 100644 --- a/config/scripts/electron-builder-config.test.mjs +++ b/config/scripts/electron-builder-config.test.mjs @@ -488,7 +488,7 @@ describe('electron-builder config', () => { } }) - it('includes @parcel/watcher in the packaged runtime closure', () => { + it('includes external main dependencies in the packaged runtime closure', () => { // Why: the main process imports '@parcel/watcher' for filesystem change // events; if it is absent from the packaged closure the serve host silently // stops propagating file changes to clients (regression guard for #4851). @@ -500,6 +500,7 @@ describe('electron-builder config', () => { target.startsWith(join('node_modules', '@parcel', 'watcher-')) ) ).toBe(true) + expect(packagedTargets).toContain(join('node_modules', 'proper-lockfile')) }) it('prunes non-target @parcel/watcher architecture subpackages', async () => { diff --git a/package.json b/package.json index 726a5081dc4..faa15fcb44c 100644 --- a/package.json +++ b/package.json @@ -137,6 +137,7 @@ "jsonc-parser": "^3.3.1", "node-pty": "^1.1.0", "posthog-node": "^5.33.3", + "proper-lockfile": "4.1.2", "psl": "1.15.0", "qrcode": "^1.5.4", "react-i18next": "^17.0.8", @@ -179,6 +180,7 @@ "@tiptap/react": "^3.22.5", "@tiptap/starter-kit": "^3.22.5", "@types/node": "^25.6.0", + "@types/proper-lockfile": "^4.1.4", "@types/qrcode": "^1.5.6", "@types/react": "^19.2.17", "@types/react-dom": "^19.2.3", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index f969e234338..80f87f7e8a6 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -67,6 +67,9 @@ importers: posthog-node: specifier: ^5.33.3 version: 5.33.3 + proper-lockfile: + specifier: 4.1.2 + version: 4.1.2 psl: specifier: 1.15.0 version: 1.15.0 @@ -188,6 +191,9 @@ importers: '@types/node': specifier: ^25.6.0 version: 25.9.5 + '@types/proper-lockfile': + specifier: ^4.1.4 + version: 4.1.4 '@types/qrcode': specifier: ^1.5.6 version: 1.5.6 @@ -3163,6 +3169,9 @@ packages: '@types/node@25.9.5': resolution: {integrity: sha512-OScDchr2fwuUmWdf4kZ9h7PcJiYDVInhJizG/biAq3cAvqwYktuy/TYGGdZNMtNTFUP7rnb0NU4TUdm82kt4Rg==} + '@types/proper-lockfile@4.1.4': + resolution: {integrity: sha512-uo2ABllncSqg9F1D4nugVl9v93RmjxF6LJzQLMLDdPaXCUIDPeOJ21Gbqi43xNKzBi/WQ0Q0dICqufzQbMjipQ==} + '@types/qrcode@1.5.6': resolution: {integrity: sha512-te7NQcV2BOvdj2b1hCAHzAoMNuj65kNBMz0KBaxM6c3VGBOhU0dURQKOtH8CFNI/dsKkwlv32p26qYQTWoB5bw==} @@ -9419,6 +9428,10 @@ snapshots: dependencies: undici-types: 7.24.6 + '@types/proper-lockfile@4.1.4': + dependencies: + '@types/retry': 0.12.0 + '@types/qrcode@1.5.6': dependencies: '@types/node': 25.9.5 diff --git a/src/main/runtime/agent-session-lease-transitions.ts b/src/main/runtime/agent-session-lease-transitions.ts new file mode 100644 index 00000000000..0541039942a --- /dev/null +++ b/src/main/runtime/agent-session-lease-transitions.ts @@ -0,0 +1,314 @@ +/** + * Pure lease state transitions. Every function returns the next record or throws a typed error; + * the store applies them inside one durable transaction so a rejected transition never lands. + * + * The invariant they exist to enforce: a session admits a writer only after a reservation, an + * observed process identity, and a proved provider handle — in that order, at one fence. + */ + +import { isDeepStrictEqual } from 'node:util' +import { + adjudicateAgentSessionRestart, + evaluateAgentSessionAcquisition, + type AgentSessionOwnerProbe +} from '../../shared/agent-session-lease-adjudication' +import { + appendAgentSessionProviderHandleLink, + type AgentSessionProviderHandleLink +} from '../../shared/agent-session-provider-handle' +import type { + AgentSessionJournalCheckpoint, + AgentSessionHandoffStage, + AgentSessionLease, + AgentSessionOwnerRuntimeKind, + AgentSessionProcessIdentity, + AgentSessionRecord +} from '../../shared/agent-session-record' + +export type AgentSessionReservation = { + runtimeKind: AgentSessionOwnerRuntimeKind + spawnToken: string + claimKeyId: string + handoffOperationId: string | null + leaseTtlMs: number + now: number +} + +function withLease(record: AgentSessionRecord, lease: AgentSessionLease): AgentSessionRecord { + return { ...record, lease, updatedAt: lease.lastRenewedAt } +} + +function assertFence(lease: AgentSessionLease, fence: number): void { + if (lease.runtimeFence !== fence) { + throw new Error('agent_session_checkpoint_stale') + } + if (lease.unreconciled) { + throw new Error('execution_owner_reconciling') + } +} + +export function agentSessionReconciliationTargetMatches( + current: AgentSessionRecord, + probed: AgentSessionRecord +): boolean { + return isDeepStrictEqual(current, probed) +} + +/** + * Compare-and-swap reservation. Writes the intent at fence + 1 before any process exists, so the + * loser of a concurrent swap is refused and never spawns. + */ +export function reserveAgentSessionOwner(args: { + record: AgentSessionRecord + expectedFence: number + probe: AgentSessionOwnerProbe + reservation: AgentSessionReservation +}): { record: AgentSessionRecord; disposition: 'reserved' | 'retry-reservation' } { + const { record, reservation } = args + const decision = evaluateAgentSessionAcquisition({ + lease: record.lease, + expectedFence: args.expectedFence, + handoffOperationId: reservation.handoffOperationId, + probe: args.probe + }) + if (decision.decision === 'refused') { + throw new Error(decision.code) + } + if (decision.decision === 'retry-reservation') { + return { record, disposition: 'retry-reservation' } + } + return { + disposition: 'reserved', + record: withLease(record, { + ...record.lease, + runtimeKind: reservation.runtimeKind, + runtimeFence: decision.nextFence, + // Why: a reserved owner is not yet a writer; it may only talk to the provider to prove resume. + handoffStage: 'new-owner-proving', + provenHandleLinkId: null, + ownerProcess: null, + reservedSpawnToken: reservation.spawnToken, + leaseDeadlineAt: reservation.now + reservation.leaseTtlMs, + lastRenewedAt: reservation.now, + handoffOperationId: reservation.handoffOperationId, + claimKeyId: reservation.claimKeyId, + claimStatus: 'reserved', + deathEvidence: null + }) + } +} + +/** Step 4 of acquisition: write the observed identity back into the same lease row. */ +export function commitAgentSessionProcessIdentity(args: { + record: AgentSessionRecord + fence: number + process: AgentSessionProcessIdentity + now: number +}): AgentSessionRecord { + const { record } = args + assertFence(record.lease, args.fence) + if (record.lease.claimStatus !== 'reserved' || record.lease.ownerProcess !== null) { + throw new Error('agent_session_ownership_unknown') + } + if (record.lease.reservedSpawnToken !== args.process.spawnToken) { + // Why: a child that cannot echo the reserved token is not the process Orca started. + throw new Error('agent_session_ownership_unknown') + } + return withLease(record, { ...record.lease, ownerProcess: args.process, lastRenewedAt: args.now }) +} + +/** + * The new runtime proved it resumed the expected provider handle. Only now does the session have + * a writer. + */ +export function proveAgentSessionOwner(args: { + record: AgentSessionRecord + fence: number + link: AgentSessionProviderHandleLink + now: number + leaseTtlMs: number +}): AgentSessionRecord { + const { record } = args + assertFence(record.lease, args.fence) + if ( + record.lease.claimStatus !== 'reserved' || + record.lease.handoffStage !== 'new-owner-proving' || + record.lease.ownerProcess === null + ) { + throw new Error('agent_session_ownership_unknown') + } + if (args.link.handle.provider !== record.provider) { + throw new Error('agent_session_provider_handle_provider_mismatch') + } + if (args.link.mintedAtFence !== args.fence) { + throw new Error('agent_session_provider_handle_stale_fence') + } + const providerHandleChain = appendAgentSessionProviderHandleLink( + record.providerHandleChain, + args.link + ) + const head = providerHandleChain.at(-1) + if (!head) { + throw new Error('agent_session_provider_handle_invalid') + } + return { + ...record, + providerHandleChain, + lease: { + ...record.lease, + handoffStage: null, + provenHandleLinkId: head.linkId, + claimStatus: 'live', + leaseDeadlineAt: args.now + args.leaseTtlMs, + lastRenewedAt: args.now, + handoffOperationId: null + }, + updatedAt: args.now + } +} + +/** + * A renewal asserts two things at once: the host is running its loop, and the child still matches + * the recorded identity. A host that cannot re-verify the child stops renewing rather than + * extending a lease it can no longer vouch for. + */ +export function renewAgentSessionLease(args: { + record: AgentSessionRecord + fence: number + childProbe: AgentSessionOwnerProbe + now: number + leaseTtlMs: number +}): AgentSessionRecord { + const { record } = args + assertFence(record.lease, args.fence) + if (record.lease.ownerProcess === null) { + throw new Error('agent_session_ownership_unknown') + } + if (args.childProbe.outcome !== 'identity-matched' || args.childProbe.matchedOn.length === 0) { + throw new Error('agent_session_ownership_unknown') + } + return withLease(record, { + ...record.lease, + leaseDeadlineAt: args.now + args.leaseTtlMs, + lastRenewedAt: args.now + }) +} + +/** Proven eviction — the only other thing besides acquisition that may move the fence. */ +export function evictAgentSessionOwner(args: { + record: AgentSessionRecord + expectedFence: number + probe: AgentSessionOwnerProbe + now: number +}): AgentSessionRecord { + const { record } = args + assertFence(record.lease, args.expectedFence) + const adjudication = adjudicateAgentSessionRestart({ + lease: record.lease, + probe: args.probe, + observedAt: args.now + }) + if (adjudication.disposition !== 'evicted') { + throw new Error('agent_session_ownership_unknown') + } + return withLease(record, { + ...record.lease, + runtimeFence: adjudication.nextFence, + handoffStage: null, + ownerProcess: null, + reservedSpawnToken: null, + claimStatus: 'released', + lastRenewedAt: args.now, + handoffOperationId: null, + deathEvidence: adjudication.evidence + }) +} + +/** Apply one restart adjudication. Never consults deadlines — only proof moves a lease. */ +export function applyAgentSessionRestartAdjudication(args: { + record: AgentSessionRecord + probe: AgentSessionOwnerProbe + now: number +}): AgentSessionRecord { + const { record } = args + const adjudication = adjudicateAgentSessionRestart({ + lease: record.lease, + probe: args.probe, + observedAt: args.now + }) + if (adjudication.disposition === 'readopt') { + // Why: re-adoption is not a new generation, so the fence does not move. + return withLease(record, { ...record.lease, unreconciled: false, lastRenewedAt: args.now }) + } + if (adjudication.disposition === 'evicted') { + return withLease(record, { + ...record.lease, + runtimeFence: adjudication.nextFence, + handoffStage: null, + ownerProcess: null, + reservedSpawnToken: null, + claimStatus: 'released', + unreconciled: false, + lastRenewedAt: args.now, + handoffOperationId: null, + deathEvidence: adjudication.evidence + }) + } + const stage: AgentSessionHandoffStage = + adjudication.disposition === 'conflicted' ? 'manual-recovery' : adjudication.stage + return withLease(record, { + ...record.lease, + handoffStage: stage, + claimStatus: + adjudication.disposition === 'conflicted' ? 'conflicted' : record.lease.claimStatus, + unreconciled: false, + lastRenewedAt: args.now + }) +} + +export function setAgentSessionHandoffStage(args: { + record: AgentSessionRecord + fence: number + stage: AgentSessionHandoffStage | null + handoffOperationId: string | null + now: number +}): AgentSessionRecord { + const { record } = args + assertFence(record.lease, args.fence) + if ( + record.lease.handoffOperationId !== null && + args.handoffOperationId !== null && + args.handoffOperationId !== record.lease.handoffOperationId + ) { + throw new Error('agent_session_operation_conflict') + } + return withLease(record, { + ...record.lease, + handoffStage: args.stage, + handoffOperationId: args.handoffOperationId, + lastRenewedAt: args.now + }) +} + +export function setAgentSessionJournalCheckpoint(args: { + record: AgentSessionRecord + fence: number + checkpoint: AgentSessionJournalCheckpoint + now: number +}): AgentSessionRecord { + const { record } = args + assertFence(record.lease, args.fence) + const current = record.lease.journalCheckpoint + if ( + current && + (current.epoch > args.checkpoint.epoch || + (current.epoch === args.checkpoint.epoch && current.sequence > args.checkpoint.sequence)) + ) { + throw new Error('agent_session_checkpoint_stale') + } + return withLease(record, { + ...record.lease, + journalCheckpoint: args.checkpoint, + lastRenewedAt: args.now + }) +} diff --git a/src/main/runtime/agent-session-process-identity-probe.test.ts b/src/main/runtime/agent-session-process-identity-probe.test.ts new file mode 100644 index 00000000000..57e0361baa7 --- /dev/null +++ b/src/main/runtime/agent-session-process-identity-probe.test.ts @@ -0,0 +1,209 @@ +import { describe, expect, it, vi } from 'vitest' +import type { AgentSessionProcessIdentity } from '../../shared/agent-session-record' +import { + PROCESS_START_TIME_TOLERANCE_MS, + probeAgentSessionProcessIdentity, + probeAgentSessionReservation, + readProcessStartTimeMs, + type AgentSessionProcessProbeDeps +} from './agent-session-process-identity-probe' + +const START_TIME = 1_700_000_000_000 + +const IDENTITY: AgentSessionProcessIdentity = { + hostId: 'local', + pid: 4242, + processStartTimeMs: START_TIME, + spawnToken: 'spawn-a' +} + +function deps(overrides: AgentSessionProcessProbeDeps = {}): AgentSessionProcessProbeDeps { + return { + isPidPresent: () => true, + readProcessStartTimeMs: async () => START_TIME, + readEchoedSpawnToken: async () => 'spawn-a', + platform: 'linux', + ...overrides + } +} + +describe('owner identity probe', () => { + it('reports an observed exit without touching the host', () => { + return expect( + probeAgentSessionProcessIdentity({ + identity: IDENTITY, + observedExit: true, + deps: deps({ + isPidPresent: () => { + throw new Error('must not probe after an observed exit') + } + }) + }) + ).resolves.toEqual({ outcome: 'exit-observed' }) + }) + + it('reports an absent pid as proven death', async () => { + await expect( + probeAgentSessionProcessIdentity({ + identity: IDENTITY, + deps: deps({ isPidPresent: () => false }) + }) + ).resolves.toEqual({ outcome: 'pid-absent' }) + }) + + it('does not call an unexpected host probe error proof of death', async () => { + const kill = vi.spyOn(process, 'kill').mockImplementation(() => { + throw Object.assign(new Error('host probe unavailable'), { code: 'EIO' }) + }) + try { + const probe = await probeAgentSessionProcessIdentity({ + identity: { ...IDENTITY, processStartTimeMs: null }, + deps: { readEchoedSpawnToken: async () => null } + }) + expect(probe.outcome).toBe('indeterminate') + } finally { + kill.mockRestore() + } + }) + + it('catches pid reuse through the spawn token', async () => { + // The pid is live and started at the recorded time, but it is a different process. + await expect( + probeAgentSessionProcessIdentity({ + identity: IDENTITY, + deps: deps({ readEchoedSpawnToken: async () => 'spawn-other' }) + }) + ).resolves.toEqual({ outcome: 'identity-mismatch', field: 'spawn-token' }) + }) + + it('catches pid reuse through the start time when no token comes back', async () => { + await expect( + probeAgentSessionProcessIdentity({ + identity: IDENTITY, + deps: deps({ + readEchoedSpawnToken: async () => null, + readProcessStartTimeMs: async () => START_TIME + PROCESS_START_TIME_TOLERANCE_MS + 1 + }) + }) + ).resolves.toEqual({ outcome: 'identity-mismatch', field: 'process-start-time' }) + }) + + it('fails closed when an exact token and the reconstructed start time disagree', async () => { + await expect( + probeAgentSessionProcessIdentity({ + identity: IDENTITY, + deps: deps({ + readProcessStartTimeMs: async () => START_TIME + PROCESS_START_TIME_TOLERANCE_MS + 1 + }) + }) + ).resolves.toEqual({ + outcome: 'indeterminate', + reason: 'process identity evidence contradicted' + }) + }) + + it('tolerates start-time jitter inside the tolerance', async () => { + await expect( + probeAgentSessionProcessIdentity({ + identity: IDENTITY, + deps: deps({ + readEchoedSpawnToken: async () => null, + readProcessStartTimeMs: async () => START_TIME + PROCESS_START_TIME_TOLERANCE_MS + }) + }) + ).resolves.toEqual({ outcome: 'identity-matched', matchedOn: ['process-start-time'] }) + }) + + it('reports every element it could actually verify', async () => { + await expect( + probeAgentSessionProcessIdentity({ identity: IDENTITY, deps: deps() }) + ).resolves.toEqual({ + outcome: 'identity-matched', + matchedOn: ['spawn-token', 'process-start-time'] + }) + }) + + it('fails closed when the pid is live but nothing PID-reuse-safe could be checked', async () => { + // The Windows case: no start time recorded and no token echo, so a bare pid match is all the + // host has — and a bare pid match is what mints a second writer. + const probe = await probeAgentSessionProcessIdentity({ + identity: { ...IDENTITY, processStartTimeMs: null }, + deps: deps({ readEchoedSpawnToken: async () => null, platform: 'win32' }) + }) + expect(probe.outcome).toBe('indeterminate') + }) + + it('fails closed when the host errors instead of answering', async () => { + const probe = await probeAgentSessionProcessIdentity({ + identity: { ...IDENTITY, processStartTimeMs: null }, + deps: deps({ + readEchoedSpawnToken: async () => { + throw new Error('handshake unavailable') + } + }) + }) + expect(probe.outcome).toBe('indeterminate') + }) + + it('still proves life from the token when the start time is unreadable', async () => { + await expect( + probeAgentSessionProcessIdentity({ + identity: IDENTITY, + deps: deps({ readProcessStartTimeMs: async () => null }) + }) + ).resolves.toEqual({ outcome: 'identity-matched', matchedOn: ['spawn-token'] }) + }) + + it('returns no start time on Windows rather than guessing one', async () => { + await expect(readProcessStartTimeMs(process.pid, 'win32')).resolves.toBeNull() + }) + + it('reads a start time for the current process on this platform', async () => { + const observed = await readProcessStartTimeMs(process.pid) + if (process.platform === 'linux' || process.platform === 'darwin') { + expect(observed).not.toBeNull() + expect(Math.abs((observed as number) - (Date.now() - process.uptime() * 1000))).toBeLessThan( + 60_000 + ) + } else { + expect(observed).toBeNull() + } + }) +}) + +describe('reservation probe', () => { + it('declares a reservation unused only with positive proof nothing started', async () => { + await expect( + probeAgentSessionReservation({ + spawnToken: 'spawn-a', + findProcessesWithSpawnToken: async () => [], + hasProviderActivitySinceReservation: async () => false + }) + ).resolves.toEqual({ outcome: 'reservation-unused' }) + }) + + it.each([ + ['a process still carries the token', async () => [999], async () => false], + ['the host cannot enumerate', async () => null, async () => false], + ['provider activity is unknown', async () => [], async () => null], + ['the provider saw activity', async () => [], async () => true] + ] as const)('stays indeterminate when %s', async (_name, findProcesses, hasActivity) => { + const probe = await probeAgentSessionReservation({ + spawnToken: 'spawn-a', + findProcessesWithSpawnToken: findProcesses, + hasProviderActivitySinceReservation: hasActivity + }) + expect(probe.outcome).toBe('indeterminate') + }) + + it('stays indeterminate when enumeration throws', async () => { + const probe = await probeAgentSessionReservation({ + spawnToken: 'spawn-a', + findProcessesWithSpawnToken: async () => { + throw new Error('ps unavailable') + }, + hasProviderActivitySinceReservation: async () => false + }) + expect(probe.outcome).toBe('indeterminate') + }) +}) diff --git a/src/main/runtime/agent-session-process-identity-probe.ts b/src/main/runtime/agent-session-process-identity-probe.ts new file mode 100644 index 00000000000..8d63fd485ac --- /dev/null +++ b/src/main/runtime/agent-session-process-identity-probe.ts @@ -0,0 +1,174 @@ +/** + * PID-reuse-safe process identity probe for the single-writer lease. + * + * Pids are reused within minutes on a busy host, and reuse happens precisely in the recovery + * case, so a bare pid match is never proof. Every element of the identity tuple is unavailable + * somewhere — start time costs a CIM query on Windows and is missing in some containers, /proc + * does not exist on macOS — so an unanswerable probe reports `indeterminate` and the lease fails + * closed to manual recovery rather than minting a second writer. + */ + +import { execFile } from 'node:child_process' +import { readFile } from 'node:fs/promises' +import { promisify } from 'node:util' +import type { + AgentSessionIdentityMatchField, + AgentSessionOwnerProbe +} from '../../shared/agent-session-lease-adjudication' +import type { AgentSessionProcessIdentity } from '../../shared/agent-session-record' + +const execFileAsync = promisify(execFile) + +/** Start times drift by scheduler granularity and clock reads; compare with a tolerance. */ +export const PROCESS_START_TIME_TOLERANCE_MS = 2_000 + +const PROCESS_START_TIME_TIMEOUT_MS = 5_000 + +export type AgentSessionProcessProbeDeps = { + /** ESRCH means gone; EPERM means present but owned by another user. */ + isPidPresent?: (pid: number) => boolean + readProcessStartTimeMs?: (pid: number, platform?: NodeJS.Platform) => Promise + /** Token the running child echoed back through the adapter handshake or provider hook. */ + readEchoedSpawnToken?: (identity: AgentSessionProcessIdentity) => Promise + platform?: NodeJS.Platform +} + +function defaultIsPidPresent(pid: number): boolean { + try { + process.kill(pid, 0) + return true + } catch (error) { + // Why: only ESRCH proves absence; permission and transient host failures must fail closed. + return (error as NodeJS.ErrnoException)?.code !== 'ESRCH' + } +} + +async function readLinuxProcessStartTimeMs(pid: number): Promise { + try { + const [stat, systemStat] = await Promise.all([ + readFile(`/proc/${pid}/stat`, 'utf-8'), + readFile('/proc/stat', 'utf-8') + ]) + // Field 22 is starttime in clock ticks; the comm field can contain spaces, so cut past ") ". + const fields = stat.slice(stat.lastIndexOf(') ') + 2).split(' ') + const ticks = Number(fields[19]) + const bootTimeSeconds = Number(/^btime\s+(\d+)$/m.exec(systemStat)?.[1]) + if (!Number.isFinite(ticks) || !Number.isFinite(bootTimeSeconds)) { + return null + } + return Math.round(bootTimeSeconds * 1000 + (ticks / 100) * 1000) + } catch { + return null + } +} + +async function readDarwinProcessStartTimeMs(pid: number): Promise { + try { + const { stdout } = await execFileAsync('ps', ['-o', 'lstart=', '-p', String(pid)], { + timeout: PROCESS_START_TIME_TIMEOUT_MS + }) + const parsed = Date.parse(stdout.trim()) + return Number.isFinite(parsed) ? parsed : null + } catch { + return null + } +} + +/** + * Windows returns null: reading a start time needs a CIM query Orca does not make here, so a + * Windows child without a token echo qualifies for manual recovery only. + */ +export async function readProcessStartTimeMs( + pid: number, + platform: NodeJS.Platform = process.platform +): Promise { + if (platform === 'linux') { + return readLinuxProcessStartTimeMs(pid) + } + if (platform === 'darwin') { + return readDarwinProcessStartTimeMs(pid) + } + return null +} + +/** + * Probe one recorded owner. `observedExit` short-circuits everything: Orca watching that exact + * process exit is the strongest evidence available. + */ +export async function probeAgentSessionProcessIdentity(args: { + identity: AgentSessionProcessIdentity + observedExit?: boolean + deps?: AgentSessionProcessProbeDeps +}): Promise { + const { identity } = args + const deps = args.deps ?? {} + if (args.observedExit) { + return { outcome: 'exit-observed' } + } + const isPidPresent = deps.isPidPresent ?? defaultIsPidPresent + if (!isPidPresent(identity.pid)) { + return { outcome: 'pid-absent' } + } + const matchedOn: AgentSessionIdentityMatchField[] = [] + const echoedToken = await deps.readEchoedSpawnToken?.(identity).catch(() => null) + if (echoedToken !== null && echoedToken !== undefined) { + if (echoedToken !== identity.spawnToken) { + return { outcome: 'identity-mismatch', field: 'spawn-token' } + } + matchedOn.push('spawn-token') + } + if (identity.processStartTimeMs !== null) { + const readStartTime = deps.readProcessStartTimeMs ?? readProcessStartTimeMs + const observed = await readStartTime(identity.pid, deps.platform ?? process.platform).catch( + () => null + ) + if (observed !== null) { + if (Math.abs(observed - identity.processStartTimeMs) > PROCESS_START_TIME_TOLERANCE_MS) { + if (matchedOn.includes('spawn-token')) { + // Why: contradictory evidence cannot prove that a token-authenticated child is dead. + return { outcome: 'indeterminate', reason: 'process identity evidence contradicted' } + } + return { outcome: 'identity-mismatch', field: 'process-start-time' } + } + matchedOn.push('process-start-time') + } + } + if (matchedOn.length === 0) { + // Why: the pid exists and nothing PID-reuse-safe could be checked. Reporting a match here is + // exactly the case that produces two writers on one provider session. + return { + outcome: 'indeterminate', + reason: 'pid present but neither spawn token nor start time could be verified' + } + } + return { outcome: 'identity-matched', matchedOn } +} + +/** + * Probe a reservation that has no recorded process. `reservation-unused` requires positive proof + * that nothing started — no process carrying the token and no provider-side activity after the + * reservation — not an assumption that the crash beat the spawn. + */ +export async function probeAgentSessionReservation(args: { + spawnToken: string + findProcessesWithSpawnToken: (spawnToken: string) => Promise + hasProviderActivitySinceReservation: () => Promise +}): Promise { + const pids = await args.findProcessesWithSpawnToken(args.spawnToken).catch(() => null) + if (pids === null) { + return { outcome: 'indeterminate', reason: 'host could not enumerate spawn tokens' } + } + if (pids.length > 0) { + return { + outcome: 'indeterminate', + reason: `reservation spawn token is live on ${pids.length} process(es)` + } + } + const providerActivity = await args.hasProviderActivitySinceReservation().catch(() => null) + if (providerActivity === null) { + return { outcome: 'indeterminate', reason: 'provider activity since reservation is unknown' } + } + return providerActivity + ? { outcome: 'indeterminate', reason: 'provider saw activity after the reservation' } + : { outcome: 'reservation-unused' } +} diff --git a/src/main/runtime/agent-session-record-store-file.ts b/src/main/runtime/agent-session-record-store-file.ts new file mode 100644 index 00000000000..57fe0001920 --- /dev/null +++ b/src/main/runtime/agent-session-record-store-file.ts @@ -0,0 +1,246 @@ +/** + * On-disk layer for the durable agent-session store. + * + * Every mutation is a whole-file atomic transaction — temp write, fsync, rename — so a SIGKILL + * at any point leaves either the previous committed state or the next one, never a torn lease. + * That matters because this host restarts its runtime often; a half-written lease would be + * indistinguishable from an owner whose identity cannot be verified. + */ + +import { createHash } from 'node:crypto' +import { mkdir, readFile, rename, rm } from 'node:fs/promises' +import { dirname, join } from 'node:path' +import { + agentSessionOperationKey, + isAgentSessionOperationRow, + type AgentSessionOperationRow +} from '../../shared/agent-session-operation-ledger' +import { isAgentSessionRecord, type AgentSessionRecord } from '../../shared/agent-session-record' +import { durableWriteTempPath, writeFileDurable } from '../durable-file-write' + +export const AGENT_SESSION_STORE_SCHEMA_VERSION = 1 as const + +export const AGENT_SESSION_STORE_FILE_NAME = 'agent-sessions.json' + +export type RetiredAgentSessionClaimKey = { keyId: string; retiredAt: number } + +export type AgentSessionStoreState = { + schemaVersion: number + hostId: string + records: Map + operations: Map + retiredClaimKeys: RetiredAgentSessionClaimKey[] + /** Rows this build cannot validate, kept verbatim so a rollback cannot delete another host's work. */ + unreadableRecords: Map +} + +export type LoadedAgentSessionStore = { + state: AgentSessionStoreState + storeFound: boolean + /** True when the file was written by a newer schema; this host reads but never writes it. */ + readOnly: boolean + /** True when the primary file was unusable and the previous committed copy was used. */ + recoveredFromBackup: boolean +} + +export function agentSessionStorePath(directory: string): string { + return join(directory, AGENT_SESSION_STORE_FILE_NAME) +} + +function backupPath(filePath: string): string { + return `${filePath}.bak` +} + +function emptyState(hostId: string): AgentSessionStoreState { + return { + schemaVersion: AGENT_SESSION_STORE_SCHEMA_VERSION, + hostId, + records: new Map(), + operations: new Map(), + retiredClaimKeys: [], + unreadableRecords: new Map() + } +} + +export function agentSessionStoreRevision(state: AgentSessionStoreState): string { + return createHash('sha256') + .update(String(state.schemaVersion)) + .update('\0') + .update(serializeState(state)) + .digest('hex') +} + +function parseState(raw: string, hostId: string): { state: AgentSessionStoreState } | null { + let parsed: unknown + try { + parsed = JSON.parse(raw) + } catch { + return null + } + if (typeof parsed !== 'object' || parsed === null) { + return null + } + const file = parsed as { + schemaVersion?: unknown + hostId?: unknown + records?: unknown + operations?: unknown + retiredClaimKeys?: unknown + } + if ( + !Number.isSafeInteger(file.schemaVersion) || + (file.schemaVersion as number) < 0 || + typeof file.hostId !== 'string' + ) { + return null + } + const schemaVersion = file.schemaVersion as number + if ( + schemaVersion <= AGENT_SESSION_STORE_SCHEMA_VERSION && + (typeof file.records !== 'object' || + file.records === null || + Array.isArray(file.records) || + typeof file.operations !== 'object' || + file.operations === null || + Array.isArray(file.operations) || + (schemaVersion === AGENT_SESSION_STORE_SCHEMA_VERSION && + !Array.isArray(file.retiredClaimKeys))) + ) { + return null + } + const state = emptyState(hostId) + state.schemaVersion = schemaVersion + state.hostId = file.hostId + if (typeof file.records === 'object' && file.records !== null) { + for (const [sessionId, value] of Object.entries(file.records)) { + if (isAgentSessionRecord(value) && value.sessionId === sessionId) { + state.records.set(sessionId, value) + } else { + // Why: a record this build cannot read must not silently vanish, and must never be + // granted a writer; keep it and refuse ownership for that session id. + state.unreadableRecords.set(sessionId, value) + } + } + } + if (typeof file.operations === 'object' && file.operations !== null) { + for (const [key, value] of Object.entries(file.operations)) { + if (!isAgentSessionOperationRow(value)) { + if (schemaVersion <= AGENT_SESSION_STORE_SCHEMA_VERSION) { + return null + } + continue + } + if (key !== agentSessionOperationKey(value.callerKey, value.operationId)) { + if (schemaVersion <= AGENT_SESSION_STORE_SCHEMA_VERSION) { + return null + } + continue + } + state.operations.set(key, value) + } + } + if (Array.isArray(file.retiredClaimKeys)) { + for (const entry of file.retiredClaimKeys) { + const key = entry as Partial + if ( + typeof key?.keyId !== 'string' || + key.keyId.length === 0 || + key.keyId.length > 512 || + !Number.isSafeInteger(key.retiredAt) || + (key.retiredAt as number) < 0 + ) { + if (schemaVersion <= AGENT_SESSION_STORE_SCHEMA_VERSION) { + return null + } + continue + } + state.retiredClaimKeys.push({ keyId: key.keyId, retiredAt: key.retiredAt as number }) + } + } + return { state } +} + +export async function loadAgentSessionStore( + filePath: string, + hostId: string +): Promise { + let unusableStoreFound = false + for (const [candidate, recoveredFromBackup] of [ + [filePath, false], + [backupPath(filePath), true] + ] as const) { + let raw: string + try { + raw = await readFile(candidate, 'utf-8') + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'ENOENT') { + unusableStoreFound = true + } + continue + } + const parsed = parseState(raw, hostId) + if (!parsed) { + unusableStoreFound = true + continue + } + return { + state: parsed.state, + storeFound: true, + readOnly: parsed.state.schemaVersion > AGENT_SESSION_STORE_SCHEMA_VERSION, + recoveredFromBackup + } + } + if (unusableStoreFound) { + throw new Error('agent_session_store_corrupt') + } + return { + state: emptyState(hostId), + storeFound: false, + readOnly: false, + recoveredFromBackup: false + } +} + +function serializeState(state: AgentSessionStoreState): string { + const records: Record = Object.create(null) + for (const [sessionId, value] of state.unreadableRecords) { + records[sessionId] = value + } + for (const [sessionId, record] of state.records) { + records[sessionId] = record + } + return JSON.stringify({ + schemaVersion: AGENT_SESSION_STORE_SCHEMA_VERSION, + hostId: state.hostId, + records, + operations: Object.fromEntries(state.operations), + retiredClaimKeys: state.retiredClaimKeys + }) +} + +/** Commit the whole state. The previous file becomes the backup before the new one lands. */ +export async function saveAgentSessionStore( + filePath: string, + state: AgentSessionStoreState, + options: { recoveredFromBackup?: boolean } = {} +): Promise { + await mkdir(dirname(filePath), { recursive: true }) + const tmpPath = durableWriteTempPath(filePath) + if (!options.recoveredFromBackup) { + try { + await rm(backupPath(filePath), { force: true }) + await rename(filePath, backupPath(filePath)) + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'ENOENT') { + throw error + } + // First write, or recovery already found the primary missing. + } + } + try { + await writeFileDurable(tmpPath, filePath, serializeState(state)) + } catch (error) { + await rm(tmpPath, { force: true }).catch(() => {}) + throw error + } +} diff --git a/src/main/runtime/agent-session-record-store.test.ts b/src/main/runtime/agent-session-record-store.test.ts new file mode 100644 index 00000000000..e7bf796704c --- /dev/null +++ b/src/main/runtime/agent-session-record-store.test.ts @@ -0,0 +1,877 @@ +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 } from 'vitest' +import type { AgentSessionOwnerProbe } from '../../shared/agent-session-lease-adjudication' +import type { + AgentSessionExecutionLocation, + AgentSessionProcessIdentity, + AgentSessionRecord +} from '../../shared/agent-session-record' +import type { AgentSessionProviderHandleLink } from '../../shared/agent-session-provider-handle' +import { + AGENT_SESSION_CLAIM_KEY_RETENTION_MS, + AgentSessionRecordStore +} from './agent-session-record-store' +import { + agentSessionStorePath, + AGENT_SESSION_STORE_FILE_NAME +} from './agent-session-record-store-file' +import type { AgentSessionReserveRequest } from './agent-session-reservation-admission' + +const NOW = 1_800_000_000_000 + +const NATIVE: AgentSessionExecutionLocation = { + executionHostId: 'local', + wslDistro: null, + workspaceId: 'workspace-1', + workspaceKind: 'git-worktree' +} +const WSL: AgentSessionExecutionLocation = { ...NATIVE, wslDistro: 'Ubuntu-22.04' } +const SSH: AgentSessionExecutionLocation = { ...NATIVE, executionHostId: 'ssh:build-box' } +const FOLDER: AgentSessionExecutionLocation = { + ...NATIVE, + workspaceId: 'workspace-2', + workspaceKind: 'folder' +} + +const MATCHED: AgentSessionOwnerProbe = { outcome: 'identity-matched', matchedOn: ['spawn-token'] } +const INDETERMINATE: AgentSessionOwnerProbe = { outcome: 'indeterminate', reason: 'no answer' } +const UNUSED: AgentSessionOwnerProbe = { outcome: 'reservation-unused' } +const BAD_OP_STORE = '{"schemaVersion":0,"hostId":"","records":{},"operations":{"x":0}}' +const BAD_KEY_STORE = + '{"schemaVersion":1,"hostId":"","records":{},"operations":{},"retiredClaimKeys":[0]}' + +let counter = 0 + +function operationId(now = NOW): string { + counter += 1 + return `${now}-${String(counter) + .padStart(32, '0') + .replaceAll(/[^0-9a-f]/g, '0')}` +} + +function reserveRequest( + overrides: Partial = {} +): AgentSessionReserveRequest { + return { + sessionId: 'session-alpha', + location: NATIVE, + provider: 'claude', + accountHome: { variable: 'CLAUDE_CONFIG_DIR', path: '/home/dev/.claude-work' }, + runtimeKind: 'native', + expectedFence: null, + spawnToken: 'spawn-a', + claimKeyId: 'key-1', + handoffOperationId: null, + probe: INDETERMINATE, + operation: { callerKey: 'client-1', operationId: operationId(), fingerprint: 'fp-1' }, + now: NOW, + ...overrides + } +} + +function processIdentity( + overrides: Partial = {} +): AgentSessionProcessIdentity { + return { + hostId: 'local', + pid: 4242, + processStartTimeMs: 1_700_000_000_000, + spawnToken: 'spawn-a', + ...overrides + } +} + +function handleLink( + overrides: Partial = {} +): AgentSessionProviderHandleLink { + return { + linkId: 'link-1', + handle: { provider: 'claude', sessionId: 'provider-session-1', leafUuid: 'leaf-1' }, + origin: 'created', + mintedAtFence: 1, + observedAt: NOW, + ...overrides + } +} + +let directory: string + +async function open(hostId = 'local'): Promise { + return AgentSessionRecordStore.open({ directory, hostId }) +} + +/** Reserve, observe the spawn, prove the handle — the full path to an admitted writer. */ +async function establishOwner( + store: AgentSessionRecordStore, + overrides: Partial = {} +): Promise { + const reserved = await store.reserveOwner(reserveRequest(overrides)) + const fence = reserved.record.lease.runtimeFence + const sessionId = reserved.record.sessionId + await store.commitProcessIdentity({ + sessionId, + fence, + process: processIdentity({ spawnToken: reserved.record.lease.reservedSpawnToken ?? 'spawn-a' }), + now: NOW + }) + return store.proveOwner({ + sessionId, + fence, + link: handleLink({ mintedAtFence: fence }), + now: NOW + }) +} + +beforeEach(async () => { + directory = await mkdtemp(join(tmpdir(), 'orca-agent-session-store-')) +}) + +afterEach(async () => { + await rm(directory, { recursive: true, force: true }) +}) + +describe('acquisition path', () => { + it('admits a writer only after reservation, observed identity, and a proved handle', async () => { + const store = await open() + const reserved = await store.reserveOwner(reserveRequest()) + expect(reserved.disposition).toBe('created') + expect(reserved.record.lease).toMatchObject({ + runtimeFence: 1, + claimStatus: 'reserved', + handoffStage: 'new-owner-proving', + ownerProcess: null, + reservedSpawnToken: 'spawn-a' + }) + + // Proving before the spawn is observed is refused. + await expect( + store.proveOwner({ sessionId: 'session-alpha', fence: 1, link: handleLink(), now: NOW }) + ).rejects.toThrow('agent_session_ownership_unknown') + + await store.commitProcessIdentity({ + sessionId: 'session-alpha', + fence: 1, + process: processIdentity(), + now: NOW + }) + const proved = await store.proveOwner({ + sessionId: 'session-alpha', + fence: 1, + link: handleLink(), + now: NOW + }) + expect(proved.lease).toMatchObject({ + claimStatus: 'live', + handoffStage: null, + provenHandleLinkId: 'link-1', + runtimeFence: 1 + }) + expect(proved.providerHandleChain).toHaveLength(1) + }) + + it('refuses a child that cannot echo the reserved spawn token', async () => { + const store = await open() + await store.reserveOwner(reserveRequest()) + await expect( + store.commitProcessIdentity({ + sessionId: 'session-alpha', + fence: 1, + process: processIdentity({ spawnToken: 'spawn-other' }), + now: NOW + }) + ).rejects.toThrow('agent_session_ownership_unknown') + }) + + it('accepts provider proof only for the reserved provider at the current fence', async () => { + const store = await open() + await store.reserveOwner(reserveRequest()) + await store.commitProcessIdentity({ + sessionId: 'session-alpha', + fence: 1, + process: processIdentity(), + now: NOW + }) + + await expect( + store.proveOwner({ + sessionId: 'session-alpha', + fence: 1, + link: handleLink({ + handle: { provider: 'codex', threadId: 'thread-1' } + }), + now: NOW + }) + ).rejects.toThrow('agent_session_provider_handle_provider_mismatch') + await expect( + store.proveOwner({ + sessionId: 'session-alpha', + fence: 1, + link: handleLink({ mintedAtFence: 2 }), + now: NOW + }) + ).rejects.toThrow('agent_session_provider_handle_stale_fence') + }) + + it('does not let an established owner re-enter the proof transition', async () => { + const store = await open() + await establishOwner(store) + + await expect( + store.proveOwner({ + sessionId: 'session-alpha', + fence: 1, + link: handleLink({ + linkId: 'link-2', + origin: 'resumed', + observedAt: NOW + 1 + }), + now: NOW + 1 + }) + ).rejects.toThrow('agent_session_ownership_unknown') + }) + + it('refuses a create that carries a fence and a re-create that does not', async () => { + const store = await open() + await expect(store.reserveOwner(reserveRequest({ expectedFence: 0 }))).rejects.toThrow( + 'agent_session_checkpoint_stale' + ) + await establishOwner(store) + await expect(store.reserveOwner(reserveRequest({ expectedFence: null }))).rejects.toThrow( + 'agent_session_conflict' + ) + }) + + it.each([ + [ + 'provider', + { provider: 'codex', accountHome: { variable: 'CODEX_HOME', path: '/home/dev/.codex' } } + ], + ['account', { accountHome: { variable: 'CLAUDE_CONFIG_DIR', path: '/home/dev/.claude-other' } }] + ] as const)("refuses to change a session's pinned %s", async (_name, overrides) => { + const store = await open() + await establishOwner(store) + await expect( + store.reserveOwner( + reserveRequest({ + ...overrides, + expectedFence: 1, + probe: { outcome: 'pid-absent' }, + operation: { callerKey: 'client-1', operationId: operationId(), fingerprint: 'fp-2' } + }) + ) + ).rejects.toThrow('agent_session_conflict') + }) +}) + +describe('concurrent claims', () => { + it('lets only one store instance reserve a session from the same disk snapshot', async () => { + const [first, second] = await Promise.all([open(), open()]) + const results = await Promise.allSettled([ + first.reserveOwner(reserveRequest()), + second.reserveOwner(reserveRequest()) + ]) + + expect(results.filter((result) => result.status === 'fulfilled')).toHaveLength(1) + const refused = results.find((result) => result.status === 'rejected') + expect((refused as PromiseRejectedResult).reason.message).toBe('agent_session_conflict') + + const persisted = await open() + expect(persisted.getRecord('session-alpha')?.lease.runtimeFence).toBe(1) + expect(persisted.listOperationRows()).toHaveLength(1) + }) + + it('lets exactly one of two concurrent reservations win and never spawns the loser', async () => { + const store = await open() + await establishOwner(store) + const request = () => + reserveRequest({ + expectedFence: 1, + probe: { outcome: 'pid-absent' }, + spawnToken: 'spawn-b', + operation: { callerKey: 'client-1', operationId: operationId(), fingerprint: 'fp-2' } + }) + const results = await Promise.allSettled([ + store.reserveOwner(request()), + store.reserveOwner(request()) + ]) + const granted = results.filter((result) => result.status === 'fulfilled') + expect(granted).toHaveLength(1) + const refused = results.find((result) => result.status === 'rejected') + expect((refused as PromiseRejectedResult).reason.message).toBe('agent_session_checkpoint_stale') + expect(store.getRecord('session-alpha')?.lease.runtimeFence).toBe(2) + }) + + it('serializes concurrent creates of the same session id', async () => { + const store = await open() + const results = await Promise.allSettled([ + store.reserveOwner(reserveRequest()), + store.reserveOwner(reserveRequest()) + ]) + expect(results.filter((result) => result.status === 'fulfilled')).toHaveLength(1) + expect(store.getRecord('session-alpha')?.lease.runtimeFence).toBe(1) + }) + + it('replays a retried operation id instead of reserving twice', async () => { + const store = await open() + const operation = { callerKey: 'client-1', operationId: operationId(), fingerprint: 'fp-1' } + const first = await store.reserveOwner(reserveRequest({ operation })) + const second = await store.reserveOwner(reserveRequest({ operation })) + expect(second.disposition).toBe('replayed') + expect(second.record.lease.runtimeFence).toBe(first.record.lease.runtimeFence) + expect(store.listOperationRows()).toHaveLength(1) + }) + + it('refuses the same operation id carrying different parameters', async () => { + const store = await open() + const operationId_ = operationId() + await store.reserveOwner( + reserveRequest({ + operation: { callerKey: 'client-1', operationId: operationId_, fingerprint: 'fp-1' } + }) + ) + await expect( + store.reserveOwner( + reserveRequest({ + sessionId: 'session-beta', + operation: { callerKey: 'client-1', operationId: operationId_, fingerprint: 'fp-9' } + }) + ) + ).rejects.toThrow('agent_session_operation_conflict') + }) + + it('rolls the in-memory state back when a transaction throws', async () => { + const store = await open() + await establishOwner(store) + const before = store.getRecord('session-alpha') + await expect( + store.reserveOwner(reserveRequest({ expectedFence: 1, probe: INDETERMINATE })) + ).rejects.toThrow('agent_session_ownership_unknown') + expect(store.getRecord('session-alpha')).toEqual(before) + expect(store.listOperationRows()).toHaveLength(1) + }) + + it('never keeps a change in memory that failed to commit to disk', async () => { + const store = await open() + await establishOwner(store) + const before = store.getRecord('session-alpha') + // Losing both committed copies must not reset the live store to empty authority. + await rm(directory, { recursive: true, force: true }) + await expect( + store.setJournalCheckpoint({ + sessionId: 'session-alpha', + fence: 1, + checkpoint: { epoch: 9, sequence: 9 }, + now: NOW + }) + ).rejects.toThrow() + expect(store.getRecord('session-alpha')).toEqual(before) + expect(store.getRecord('session-alpha')?.lease.journalCheckpoint).toBeNull() + }) +}) + +describe('expiry is not eviction', () => { + it('never grants a second owner on a lapsed deadline alone', async () => { + const store = await open() + const owned = await establishOwner(store) + const wellPastDeadline = owned.lease.leaseDeadlineAt + 60 * 60 * 1000 + for (const probe of [INDETERMINATE, MATCHED] as const) { + await expect( + store.reserveOwner( + reserveRequest({ + expectedFence: 1, + probe, + now: wellPastDeadline, + operation: { + callerKey: 'client-1', + operationId: operationId(wellPastDeadline), + fingerprint: 'fp-2' + } + }) + ) + ).rejects.toThrow(/agent_session_(ownership_unknown|conflict)/) + } + expect(store.getRecord('session-alpha')?.lease.runtimeFence).toBe(1) + + // Only proof of death moves it. + const evicted = await store.evictProvenDeadOwner({ + sessionId: 'session-alpha', + expectedFence: 1, + probe: { outcome: 'pid-absent' }, + now: wellPastDeadline + }) + expect(evicted.lease).toMatchObject({ runtimeFence: 2, claimStatus: 'released' }) + expect(evicted.lease.deathEvidence?.kind).toBe('pid-absent') + }) + + it('refuses to evict an owner it cannot prove dead', async () => { + const store = await open() + await establishOwner(store) + await expect( + store.evictProvenDeadOwner({ + sessionId: 'session-alpha', + expectedFence: 1, + probe: INDETERMINATE, + now: NOW + }) + ).rejects.toThrow('agent_session_ownership_unknown') + }) + + it('stops renewing a lease it can no longer vouch for', async () => { + const store = await open() + await establishOwner(store) + const renewed = await store.renewLease({ + sessionId: 'session-alpha', + fence: 1, + childProbe: MATCHED, + now: NOW + 5_000 + }) + expect(renewed.lease.lastRenewedAt).toBe(NOW + 5_000) + await expect( + store.renewLease({ + sessionId: 'session-alpha', + fence: 1, + childProbe: { outcome: 'identity-matched', matchedOn: [] }, + now: NOW + 10_000 + }) + ).rejects.toThrow('agent_session_ownership_unknown') + }) +}) + +describe('restart reconciliation', () => { + it('survives a restart and grants no writer until adjudicated', async () => { + const first = await open() + await establishOwner(first) + + const reopened = await open() + const loaded = reopened.getRecord('session-alpha') + expect(loaded?.lease.unreconciled).toBe(true) + expect(loaded?.providerHandleChain).toHaveLength(1) + expect(loaded?.accountHome).toEqual({ + variable: 'CLAUDE_CONFIG_DIR', + path: '/home/dev/.claude-work' + }) + // Every mutating path is closed while unreconciled. + await expect( + reopened.renewLease({ sessionId: 'session-alpha', fence: 1, childProbe: MATCHED, now: NOW }) + ).rejects.toThrow('execution_owner_reconciling') + await expect( + reopened.reserveOwner( + reserveRequest({ + expectedFence: 1, + probe: { outcome: 'pid-absent' }, + operation: { callerKey: 'client-1', operationId: operationId(), fingerprint: 'fp-2' } + }) + ) + ).rejects.toThrow('execution_owner_reconciling') + }) + + it('re-adopts a live owner without moving the fence', async () => { + const first = await open() + await establishOwner(first) + const reopened = await open() + await reopened.reconcileOnRestart({ probe: async () => MATCHED, now: NOW + 1_000 }) + const record = reopened.getRecord('session-alpha') + expect(record?.lease).toMatchObject({ + unreconciled: false, + runtimeFence: 1, + claimStatus: 'live' + }) + expect(record?.lease.provenHandleLinkId).toBe('link-1') + }) + + it('never applies a stale restart probe to a replacement owner', async () => { + const writer = await open() + await establishOwner(writer) + const reconciler = await open() + let releaseProbe!: (probe: AgentSessionOwnerProbe) => void + let markProbeStarted!: () => void + const probeStarted = new Promise((resolve) => (markProbeStarted = resolve)) + const probeResult = new Promise((resolve) => (releaseProbe = resolve)) + const reconciliation = reconciler.reconcileOnRestart({ + probe: async () => { + markProbeStarted() + return probeResult + }, + now: NOW + 1_000 + }) + + await probeStarted + await writer.evictProvenDeadOwner({ + sessionId: 'session-alpha', + expectedFence: 1, + probe: { outcome: 'pid-absent' }, + now: NOW + 100 + }) + const replacement = await writer.reserveOwner( + reserveRequest({ + expectedFence: 2, + probe: UNUSED, + spawnToken: 'spawn-b', + operation: { + callerKey: 'client-1', + operationId: operationId(NOW + 200), + fingerprint: 'fp-2' + }, + now: NOW + 200 + }) + ) + await writer.commitProcessIdentity({ + sessionId: 'session-alpha', + fence: replacement.record.lease.runtimeFence, + process: processIdentity({ pid: 5252, spawnToken: 'spawn-b' }), + now: NOW + 300 + }) + await writer.proveOwner({ + sessionId: 'session-alpha', + fence: replacement.record.lease.runtimeFence, + link: handleLink({ linkId: 'link-2', origin: 'resumed', mintedAtFence: 3 }), + now: NOW + 300 + }) + + releaseProbe({ outcome: 'pid-absent' }) + expect(await reconciliation).toEqual(new Map()) + const persisted = JSON.parse(await readFile(agentSessionStorePath(directory), 'utf-8')) + expect(persisted.records['session-alpha'].lease).toMatchObject({ + runtimeFence: 3, + claimStatus: 'live', + ownerProcess: { pid: 5252, spawnToken: 'spawn-b' }, + unreconciled: false + }) + }) + + it('keeps the fence monotonic across a restart and never reuses a retired fence', async () => { + const first = await open() + await establishOwner(first) + await first.evictProvenDeadOwner({ + sessionId: 'session-alpha', + expectedFence: 1, + probe: { outcome: 'exit-observed' }, + now: NOW + }) + + const second = await open() + await second.reconcileOnRestart({ probe: async () => UNUSED, now: NOW + 1_000 }) + const afterRestart = second.getRecord('session-alpha')?.lease.runtimeFence ?? 0 + expect(afterRestart).toBeGreaterThanOrEqual(2) + + const reacquired = await second.reserveOwner( + reserveRequest({ + expectedFence: afterRestart, + probe: UNUSED, + spawnToken: 'spawn-b', + operation: { + callerKey: 'client-1', + operationId: operationId(NOW + 1_000), + fingerprint: 'fp-2' + }, + now: NOW + 1_000 + }) + ) + expect(reacquired.record.lease.runtimeFence).toBe(afterRestart + 1) + + const third = await open() + expect(third.getRecord('session-alpha')?.lease.runtimeFence).toBe(afterRestart + 1) + }) + + it('sends an unverifiable owner to recovery rather than releasing it', async () => { + const first = await open() + await establishOwner(first) + const reopened = await open() + await reopened.reconcileOnRestart({ probe: async () => INDETERMINATE, now: NOW + 1_000 }) + const lease = reopened.getRecord('session-alpha')?.lease + expect(lease).toMatchObject({ handoffStage: 'recovering', runtimeFence: 1 }) + expect(lease?.ownerProcess).not.toBeNull() + await expect( + reopened.reserveOwner( + reserveRequest({ + expectedFence: 1, + probe: { outcome: 'pid-absent' }, + operation: { + callerKey: 'client-1', + operationId: operationId(NOW + 1_000), + fingerprint: 'fp-2' + }, + now: NOW + 1_000 + }) + ) + ).rejects.toThrow('agent_session_ownership_unknown') + }) + + it('keeps a conflict conflicted across a restart', async () => { + const first = await open() + await establishOwner(first) + await first.markClaimConflicted('session-alpha', NOW) + + const reopened = await open() + await reopened.reconcileOnRestart({ probe: async () => ({ outcome: 'pid-absent' }), now: NOW }) + expect(reopened.getRecord('session-alpha')?.lease).toMatchObject({ + claimStatus: 'conflicted', + handoffStage: 'manual-recovery' + }) + await expect( + reopened.reserveOwner( + reserveRequest({ + expectedFence: 1, + probe: { outcome: 'pid-absent' }, + operation: { callerKey: 'client-1', operationId: operationId(), fingerprint: 'fp-2' } + }) + ) + ).rejects.toThrow('agent_session_conflict') + }) + + it('frees a reservation that provably never spawned', async () => { + const first = await open() + await first.reserveOwner(reserveRequest()) + const reopened = await open() + await reopened.reconcileOnRestart({ probe: async () => UNUSED, now: NOW + 1_000 }) + expect(reopened.getRecord('session-alpha')?.lease).toMatchObject({ + claimStatus: 'released', + runtimeFence: 2, + reservedSpawnToken: null + }) + }) + + it('carries the operation ledger across a restart so a retry is still a replay', async () => { + const operation = { callerKey: 'client-1', operationId: operationId(), fingerprint: 'fp-1' } + const first = await open() + await first.reserveOwner(reserveRequest({ operation })) + await first.recordOperationOutcome({ + callerKey: operation.callerKey, + operationId: operation.operationId, + outcome: { status: 'succeeded', sessionId: 'session-alpha' } + }) + + const reopened = await open() + expect(reopened.listOperationRows()).toHaveLength(1) + const replayed = await reopened.reserveOwner(reserveRequest({ operation })) + expect(replayed.disposition).toBe('replayed') + expect(replayed.record.sessionId).toBe('session-alpha') + }) +}) + +describe('host and workspace isolation', () => { + it.each([ + ['WSL', WSL], + ['SSH', SSH], + ['another workspace', FOLDER], + ['another workspace kind', { ...NATIVE, workspaceKind: 'folder' }] + ] as const)('refuses to move one session id to %s', async (_name, location) => { + const store = await open() + await establishOwner(store) + await expect( + store.reserveOwner( + reserveRequest({ + location, + expectedFence: 1, + probe: { outcome: 'pid-absent' }, + operation: { callerKey: 'client-1', operationId: operationId(), fingerprint: 'fp-2' } + }) + ) + ).rejects.toThrow('agent_session_conflict') + }) + + it('keeps native, WSL, and SSH sessions in separate scopes', async () => { + const store = await open() + await establishOwner(store, { sessionId: '__proto__' }) + await establishOwner(store, { sessionId: 'session-wsl', location: WSL, spawnToken: 'spawn-b' }) + await establishOwner(store, { sessionId: 'session-ssh', location: SSH, spawnToken: 'spawn-c' }) + await establishOwner(store, { + sessionId: 'session-folder', + location: FOLDER, + spawnToken: 'spawn-d' + }) + + expect(store.listByScope(NATIVE).map((record) => record.sessionId)).toEqual(['__proto__']) + expect(store.listByScope(WSL).map((record) => record.sessionId)).toEqual(['session-wsl']) + expect(store.listByScope(SSH).map((record) => record.sessionId)).toEqual(['session-ssh']) + expect(store.listByScope(FOLDER).map((record) => record.sessionId)).toEqual(['session-folder']) + expect((await open()).getRecord('__proto__')).not.toBeNull() + }) + + it('preserves the workspace kind so a folder workspace is never read back as a worktree', async () => { + const first = await open() + await establishOwner(first, { sessionId: 'session-folder', location: FOLDER }) + const reopened = await open() + expect(reopened.getRecord('session-folder')?.location).toEqual(FOLDER) + }) +}) + +describe('orphans, claim keys, checkpoints, and unreadable rows', () => { + it('calls a spawn token with no lease an orphan', async () => { + const store = await open() + await establishOwner(store) + expect(store.listOrphanSpawnTokens(['spawn-a', 'spawn-z'])).toEqual(['spawn-z']) + }) + + it('keeps a retired claim key verifiable for the retention window', async () => { + const store = await open() + await store.retireClaimKey('key-1', NOW) + expect(store.isClaimKeyVerifiable('key-1', NOW + AGENT_SESSION_CLAIM_KEY_RETENTION_MS)).toBe( + true + ) + expect( + store.isClaimKeyVerifiable('key-1', NOW + AGENT_SESSION_CLAIM_KEY_RETENTION_MS + 1) + ).toBe(false) + expect(store.isClaimKeyVerifiable('key-unknown', NOW)).toBe(true) + }) + + it('refuses a journal checkpoint that moves backwards', async () => { + const store = await open() + await establishOwner(store) + await store.setJournalCheckpoint({ + sessionId: 'session-alpha', + fence: 1, + checkpoint: { epoch: 2, sequence: 10 }, + now: NOW + }) + await expect( + store.setJournalCheckpoint({ + sessionId: 'session-alpha', + fence: 1, + checkpoint: { epoch: 2, sequence: 9 }, + now: NOW + }) + ).rejects.toThrow('agent_session_checkpoint_stale') + await expect( + store.setJournalCheckpoint({ + sessionId: 'session-alpha', + fence: 1, + checkpoint: { epoch: 1, sequence: 999 }, + now: NOW + }) + ).rejects.toThrow('agent_session_checkpoint_stale') + const advanced = await store.setJournalCheckpoint({ + sessionId: 'session-alpha', + fence: 1, + checkpoint: { epoch: 3, sequence: 0 }, + now: NOW + }) + expect(advanced.lease.journalCheckpoint).toEqual({ epoch: 3, sequence: 0 }) + }) + + it('rejects a handoff stage change under a different operation id', async () => { + const store = await open() + await establishOwner(store) + await store.setHandoffStage({ + sessionId: 'session-alpha', + fence: 1, + stage: 'preparing', + handoffOperationId: 'op-1', + now: NOW + }) + await expect( + store.setHandoffStage({ + sessionId: 'session-alpha', + fence: 1, + stage: 'old-owner-stopped', + handoffOperationId: 'op-2', + now: NOW + }) + ).rejects.toThrow('agent_session_operation_conflict') + }) + + it('quarantines a record it cannot validate and refuses to own that session id', async () => { + const first = await open() + await establishOwner(first) + const filePath = agentSessionStorePath(directory) + const raw = JSON.parse(await readFile(filePath, 'utf-8')) + raw.records['session-alpha'].lease.runtimeFence = 'not-a-number' + await writeFile(filePath, JSON.stringify(raw)) + + const reopened = await open() + expect(reopened.getRecord('session-alpha')).toBeNull() + expect(reopened.isSessionUnreadable('session-alpha')).toBe(true) + await expect(reopened.reserveOwner(reserveRequest())).rejects.toThrow( + 'execution_owner_reconciling' + ) + // The row is kept verbatim so a rollback does not delete another build's session. + await reopened.retireClaimKey('key-2', NOW) + const persisted = JSON.parse(await readFile(filePath, 'utf-8')) + expect(persisted.records['session-alpha'].lease.runtimeFence).toBe('not-a-number') + }) + + it.each([ + [ + 'invalid checkpoint', + (record: AgentSessionRecord) => + Object.assign(record.lease, { journalCheckpoint: { epoch: 'bad', sequence: 1 } }) + ], + [ + 'missing live proof', + (record: AgentSessionRecord) => Object.assign(record.lease, { provenHandleLinkId: null }) + ] + ])('quarantines a record with %s', async (_name, corrupt) => { + const first = await open() + await establishOwner(first) + const filePath = agentSessionStorePath(directory) + const raw = JSON.parse(await readFile(filePath, 'utf-8')) + corrupt(raw.records['session-alpha']) + await writeFile(filePath, JSON.stringify(raw)) + expect((await open()).isSessionUnreadable('session-alpha')).toBe(true) + }) + + it('recovers the previous committed state when the primary file is corrupt', async () => { + const first = await open() + await establishOwner(first) + // A second commit leaves the first as the backup. + await first.setJournalCheckpoint({ + sessionId: 'session-alpha', + fence: 1, + checkpoint: { epoch: 1, sequence: 1 }, + now: NOW + }) + await writeFile(join(directory, AGENT_SESSION_STORE_FILE_NAME), '{ truncated') + + const reopened = await open() + expect(reopened.recoveredFromBackup).toBe(true) + expect(reopened.getRecord('session-alpha')?.lease.runtimeFence).toBe(1) + + await expect(reopened.retireClaimKey('key-2', NOW)).rejects.toThrow() + const backup = JSON.parse(await readFile(`${agentSessionStorePath(directory)}.bak`, 'utf-8')) + expect(backup.records['session-alpha'].lease.runtimeFence).toBe(1) + }) + + it.each([ + ['corrupt', ['{ truncated']], + ['missing required collections', ['{"schemaVersion":1,"hostId":"local"}']], + ['invalid operation row', [BAD_OP_STORE]], + ['invalid retired key', [BAD_KEY_STORE]], + ['corrupt in both committed copies', ['{ truncated', '{ also truncated']] + ])('fails closed when the store is %s', async (_name, copies) => { + const filePath = agentSessionStorePath(directory) + await writeFile(filePath, copies[0]) + if (copies[1]) { + await writeFile(`${filePath}.bak`, copies[1]) + } + await expect(open()).rejects.toThrow('agent_session_store_corrupt') + }) + + it('refuses to write a store written by a newer schema', async () => { + const filePath = agentSessionStorePath(directory) + await writeFile( + filePath, + JSON.stringify({ schemaVersion: 99, hostId: 'local', records: {}, operations: {} }) + ) + const store = await open() + expect(store.readOnly).toBe(true) + await expect(store.reserveOwner(reserveRequest())).rejects.toThrow( + 'agent_session_legacy_required' + ) + }) + + it('migrates an older store schema on its next commit', async () => { + const filePath = agentSessionStorePath(directory) + await writeFile( + filePath, + JSON.stringify({ schemaVersion: 0, hostId: 'local', records: {}, operations: {} }) + ) + + const store = await open() + await store.retireClaimKey('key-1', NOW) + + const persisted = JSON.parse(await readFile(filePath, 'utf-8')) + expect(persisted.schemaVersion).toBe(1) + }) +}) diff --git a/src/main/runtime/agent-session-record-store.ts b/src/main/runtime/agent-session-record-store.ts new file mode 100644 index 00000000000..3f70fa44a73 --- /dev/null +++ b/src/main/runtime/agent-session-record-store.ts @@ -0,0 +1,337 @@ +/** + * Durable agent-session store for one execution host. + * + * Holds the session records, their single-writer leases, and the client-operation ledger that + * used to live only in memory — all in one file so a reservation and its operation row commit in + * the same atomic transaction. Mutations are serialized through a single write queue, so two + * concurrent compare-and-swaps in one or multiple Orca processes cannot both commit one pre-state. + * + * Nothing here is wired into the runtime yet; the store is dormant until later parts adopt it. + */ + +import { + pruneAgentSessionOperationRows, + agentSessionOperationKey, + type AgentSessionOperationOutcome, + type AgentSessionOperationRow +} from '../../shared/agent-session-operation-ledger' +import type { AgentSessionOwnerProbe } from '../../shared/agent-session-lease-adjudication' +import { classifyObservedAgentSessionSpawnToken } from '../../shared/agent-session-lease-adjudication' +import type { AgentSessionProviderHandleLink } from '../../shared/agent-session-provider-handle' +import { + agentSessionScopeKey, + type AgentSessionExecutionLocation, + type AgentSessionHandoffStage, + type AgentSessionJournalCheckpoint, + type AgentSessionProcessIdentity, + type AgentSessionRecord +} from '../../shared/agent-session-record' +import { + applyAgentSessionRestartAdjudication, + agentSessionReconciliationTargetMatches, + commitAgentSessionProcessIdentity, + evictAgentSessionOwner, + proveAgentSessionOwner, + renewAgentSessionLease, + setAgentSessionHandoffStage, + setAgentSessionJournalCheckpoint +} from './agent-session-lease-transitions' +import { + applyAgentSessionReservation, + evaluateAgentSessionReserveOperation, + requireAgentSessionRecordForReplay, + type AgentSessionReserveRequest, + type AgentSessionReserveResult +} from './agent-session-reservation-admission' +import { + agentSessionStoreRevision, + agentSessionStorePath, + loadAgentSessionStore, + type AgentSessionStoreState +} from './agent-session-record-store-file' +import { + AgentSessionStoreTransactionQueue, + markAgentSessionStoreLeasesUnreconciled +} from './agent-session-store-transaction-queue' + +export type { + AgentSessionReserveRequest, + AgentSessionReserveResult +} from './agent-session-reservation-admission' + +export const AGENT_SESSION_LEASE_TTL_MS = 30_000 +export const 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) {} + + static async open(args: { directory: string; hostId: string }): Promise { + const filePath = agentSessionStorePath(args.directory) + const loaded = await loadAgentSessionStore(filePath, args.hostId) + // Why: every persisted lease is unreconciled until this host adjudicates it, so a restart + // grants no writer on the strength of what the previous process wrote. + const diskRevision = agentSessionStoreRevision(loaded.state) + markAgentSessionStoreLeasesUnreconciled(loaded.state) + return new AgentSessionRecordStore( + new AgentSessionStoreTransactionQueue( + filePath, + args.hostId, + loaded.readOnly, + loaded.recoveredFromBackup, + loaded.storeFound, + loaded.state, + diskRevision + ) + ) + } + + private get state(): AgentSessionStoreState { + return this.transactions.state + } + + get readOnly(): boolean { + return this.transactions.readOnly + } + + get recoveredFromBackup(): boolean { + return this.transactions.recoveredFromBackup + } + + get hostId(): string { + return this.state.hostId + } + + getRecord(sessionId: string): AgentSessionRecord | null { + return this.state.records.get(sessionId) ?? null + } + + listRecords(): AgentSessionRecord[] { + return [...this.state.records.values()] + } + + listByScope(location: AgentSessionExecutionLocation): AgentSessionRecord[] { + const scope = agentSessionScopeKey(location) + return this.listRecords().filter((record) => agentSessionScopeKey(record.location) === scope) + } + + /** A record this build cannot validate: readable as present, never grantable as a writer. */ + isSessionUnreadable(sessionId: string): boolean { + return this.state.unreadableRecords.has(sessionId) + } + + listOperationRows(): AgentSessionOperationRow[] { + return [...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 + } + + /** Spawn tokens observed on the host with no matching lease. Stop them; never adopt them. */ + listOrphanSpawnTokens(observedTokens: readonly string[]): string[] { + const leases = this.listRecords().map((record) => record.lease) + return observedTokens.filter( + (spawnToken) => classifyObservedAgentSessionSpawnToken({ spawnToken, leases }) === 'orphan' + ) + } + + /** + * Compare-and-swap reservation plus its client-operation row, committed together. A replayed + * operation returns the recorded outcome and never reaches the reservation. + */ + async reserveOwner(request: AgentSessionReserveRequest): Promise { + return this.transact(() => { + const decision = evaluateAgentSessionReserveOperation(this.state, request) + if (decision.decision === 'refused') { + throw new Error(decision.code) + } + if (decision.decision === 'replay') { + const record = requireAgentSessionRecordForReplay( + this.state, + decision.row, + request.sessionId + ) + return { record, disposition: 'replayed' as const, operationRow: decision.row } + } + const result = applyAgentSessionReservation(this.state, request, AGENT_SESSION_LEASE_TTL_MS) + this.state.operations.set( + agentSessionOperationKey(request.operation.callerKey, request.operation.operationId), + decision.row + ) + this.state.records.set(result.record.sessionId, result.record) + return { ...result, operationRow: decision.row } + }) + } + + async commitProcessIdentity(args: { + sessionId: string + fence: number + process: AgentSessionProcessIdentity + now: number + }): Promise { + return this.mutate(args.sessionId, (record) => + commitAgentSessionProcessIdentity({ ...args, record }) + ) + } + + async proveOwner(args: { + sessionId: string + fence: number + link: AgentSessionProviderHandleLink + now: number + leaseTtlMs?: number + }): Promise { + return this.mutate(args.sessionId, (record) => + proveAgentSessionOwner({ + record, + fence: args.fence, + link: args.link, + now: args.now, + leaseTtlMs: args.leaseTtlMs ?? AGENT_SESSION_LEASE_TTL_MS + }) + ) + } + + async renewLease(args: { + sessionId: string + fence: number + childProbe: AgentSessionOwnerProbe + now: number + leaseTtlMs?: number + }): Promise { + return this.mutate(args.sessionId, (record) => + renewAgentSessionLease({ + record, + fence: args.fence, + childProbe: args.childProbe, + now: args.now, + leaseTtlMs: args.leaseTtlMs ?? AGENT_SESSION_LEASE_TTL_MS + }) + ) + } + + async evictProvenDeadOwner(args: { + sessionId: string + expectedFence: number + probe: AgentSessionOwnerProbe + now: number + }): Promise { + return this.mutate(args.sessionId, (record) => evictAgentSessionOwner({ ...args, record })) + } + + async setHandoffStage(args: { + sessionId: string + fence: number + stage: AgentSessionHandoffStage | null + handoffOperationId: string | null + now: number + }): Promise { + return this.mutate(args.sessionId, (record) => setAgentSessionHandoffStage({ ...args, record })) + } + + async setJournalCheckpoint(args: { + sessionId: string + fence: number + checkpoint: AgentSessionJournalCheckpoint + now: number + }): Promise { + return this.mutate(args.sessionId, (record) => + setAgentSessionJournalCheckpoint({ ...args, record }) + ) + } + + /** Adjudicate every lease this host loaded. No lease grants a writer until it appears here. */ + async reconcileOnRestart(args: { + probe: (record: AgentSessionRecord) => Promise + now: number + }): Promise> { + const pending = this.listRecords().filter((record) => record.lease.unreconciled) + const probes = new Map() + for (const record of pending) { + probes.set(record.sessionId, { record, probe: await args.probe(record) }) + } + return this.transact(() => { + const reconciled = new Map() + for (const [sessionId, probed] of probes) { + const record = this.state.records.get(sessionId) + if ( + !record?.lease.unreconciled || + !agentSessionReconciliationTargetMatches(record, probed.record) + ) { + continue + } + const next = applyAgentSessionRestartAdjudication({ + record, + probe: probed.probe, + now: args.now + }) + this.state.records.set(sessionId, next) + reconciled.set(sessionId, next) + } + this.state.operations = pruneAgentSessionOperationRows(this.state.operations, args.now) + return reconciled + }) + } + + async recordOperationOutcome(args: { + callerKey: string + operationId: string + outcome: AgentSessionOperationOutcome + }): Promise { + await this.transact(() => { + const key = agentSessionOperationKey(args.callerKey, args.operationId) + const row = this.state.operations.get(key) + if (row) { + this.state.operations.set(key, { ...row, outcome: args.outcome }) + } + }) + } + + async markClaimConflicted(sessionId: string, now: number): Promise { + return this.mutate(sessionId, (record) => ({ + ...record, + updatedAt: now, + // Why: a conflicted key must stay conflicted across a restart; it cannot resolve to free + // merely because the process that observed the conflict is gone. + lease: { ...record.lease, claimStatus: 'conflicted', handoffStage: 'manual-recovery' } + })) + } + + 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 + ) + }) + } + + private async mutate( + sessionId: string, + apply: (record: AgentSessionRecord) => AgentSessionRecord + ): Promise { + return this.transact(() => { + const record = this.state.records.get(sessionId) + if (!record) { + throw new Error( + this.isSessionUnreadable(sessionId) + ? 'execution_owner_reconciling' + : 'agent_session_identity_required' + ) + } + const next = apply(record) + this.state.records.set(sessionId, next) + return next + }) + } + + /** Serialize every mutation against the latest committed disk state. */ + private async transact(apply: () => T): Promise { + return this.transactions.transact(apply) + } +} diff --git a/src/main/runtime/agent-session-reservation-admission.ts b/src/main/runtime/agent-session-reservation-admission.ts new file mode 100644 index 00000000000..7f5c60b2a7a --- /dev/null +++ b/src/main/runtime/agent-session-reservation-admission.ts @@ -0,0 +1,166 @@ +/** + * Reservation admission: what a reserve request means against the persisted state. + * + * Pure over a store snapshot so the compare-and-swap, the idempotency replay, and the + * location-immutability check can be reasoned about without touching the disk. The store applies + * the result inside one transaction; nothing here mutates. + */ + +import { + evaluateAgentSessionOperation, + pruneAgentSessionOperationRows, + type AgentSessionOperationDecision, + type AgentSessionOperationRow +} from '../../shared/agent-session-operation-ledger' +import type { AgentSessionOwnerProbe } from '../../shared/agent-session-lease-adjudication' +import { + AGENT_SESSION_RECORD_SCHEMA_VERSION, + agentSessionExecutionLocationsEqual, + type AgentSessionAccountHome, + type AgentSessionExecutionLocation, + type AgentSessionRecord +} from '../../shared/agent-session-record' +import type { AgentSessionHandleProvider } from '../../shared/agent-session-provider-handle' +import { + reserveAgentSessionOwner, + type AgentSessionReservation +} from './agent-session-lease-transitions' +import type { AgentSessionStoreState } from './agent-session-record-store-file' + +export type AgentSessionReserveRequest = { + sessionId: string + location: AgentSessionExecutionLocation + provider: AgentSessionHandleProvider + accountHome: AgentSessionAccountHome + runtimeKind: AgentSessionReservation['runtimeKind'] + /** Null when the session does not exist yet; otherwise the fence the caller last observed. */ + expectedFence: number | null + spawnToken: string + claimKeyId: string + handoffOperationId: string | null + probe: AgentSessionOwnerProbe + operation: { callerKey: string; operationId: string; fingerprint: string } + now: number + leaseTtlMs?: number +} + +export type AgentSessionReserveDisposition = + | 'created' + | 'reserved' + | 'retry-reservation' + | 'replayed' + +export type AgentSessionReserveResult = { + record: AgentSessionRecord + disposition: AgentSessionReserveDisposition + operationRow: AgentSessionOperationRow +} + +export function evaluateAgentSessionReserveOperation( + state: AgentSessionStoreState, + request: AgentSessionReserveRequest +): AgentSessionOperationDecision { + state.operations = pruneAgentSessionOperationRows(state.operations, request.now) + return evaluateAgentSessionOperation({ + rows: state.operations, + callerKey: request.operation.callerKey, + operationId: request.operation.operationId, + fingerprint: request.operation.fingerprint, + now: request.now + }) +} + +export function requireAgentSessionRecordForReplay( + state: AgentSessionStoreState, + row: AgentSessionOperationRow, + sessionId: string +): AgentSessionRecord { + const replayedId = row.outcome.status === 'succeeded' ? row.outcome.sessionId : sessionId + const record = state.records.get(replayedId) + if (!record) { + // Why: the recorded effect is no longer reconstructable, and re-running it would be a second + // spawn rather than a replay. + throw new Error('agent_session_ownership_unknown') + } + return record +} + +export function applyAgentSessionReservation( + state: AgentSessionStoreState, + request: AgentSessionReserveRequest, + leaseTtlMs: number +): { + record: AgentSessionRecord + disposition: Exclude +} { + const reservation: AgentSessionReservation = { + runtimeKind: request.runtimeKind, + spawnToken: request.spawnToken, + claimKeyId: request.claimKeyId, + handoffOperationId: request.handoffOperationId, + leaseTtlMs: request.leaseTtlMs ?? leaseTtlMs, + now: request.now + } + const existing = state.records.get(request.sessionId) + if (!existing) { + if (state.unreadableRecords.has(request.sessionId)) { + throw new Error('execution_owner_reconciling') + } + if (request.expectedFence !== null) { + throw new Error('agent_session_checkpoint_stale') + } + return { record: createAgentSessionRecord(request, reservation), disposition: 'created' } + } + if ( + !agentSessionExecutionLocationsEqual(existing.location, request.location) || + existing.provider !== request.provider || + existing.accountHome.variable !== request.accountHome.variable || + existing.accountHome.path !== request.accountHome.path + ) { + // Why: location, provider, and account are the session identity; changing one is a fork. + throw new Error('agent_session_conflict') + } + if (request.expectedFence === null) { + throw new Error('agent_session_conflict') + } + return reserveAgentSessionOwner({ + record: existing, + expectedFence: request.expectedFence, + probe: request.probe, + reservation + }) +} + +function createAgentSessionRecord( + request: AgentSessionReserveRequest, + reservation: AgentSessionReservation +): AgentSessionRecord { + return { + schemaVersion: AGENT_SESSION_RECORD_SCHEMA_VERSION, + sessionId: request.sessionId, + location: request.location, + provider: request.provider, + providerHandleChain: [], + accountHome: request.accountHome, + createdAt: request.now, + updatedAt: request.now, + lease: { + sessionId: request.sessionId, + runtimeKind: reservation.runtimeKind, + // Why: fence 1 is the first reservation; 0 is reserved for "no owner has ever existed". + runtimeFence: 1, + handoffStage: 'new-owner-proving', + provenHandleLinkId: null, + ownerProcess: null, + reservedSpawnToken: reservation.spawnToken, + leaseDeadlineAt: reservation.now + reservation.leaseTtlMs, + lastRenewedAt: reservation.now, + handoffOperationId: reservation.handoffOperationId, + journalCheckpoint: null, + claimKeyId: reservation.claimKeyId, + claimStatus: 'reserved', + unreconciled: false, + deathEvidence: null + } + } +} diff --git a/src/main/runtime/agent-session-store-transaction-lock.ts b/src/main/runtime/agent-session-store-transaction-lock.ts new file mode 100644 index 00000000000..1586bcb61f8 --- /dev/null +++ b/src/main/runtime/agent-session-store-transaction-lock.ts @@ -0,0 +1,25 @@ +import { mkdir } from 'node:fs/promises' +import { dirname } from 'node:path' +import { lock } from 'proper-lockfile' + +const LOCK_RETRIES = { + retries: 20, + factor: 1.3, + minTimeout: 10, + maxTimeout: 250, + randomize: true +} + +/** Serialize whole-file transactions across Orca processes sharing one execution host. */ +export async function withAgentSessionStoreTransactionLock( + filePath: string, + apply: () => Promise +): Promise { + await mkdir(dirname(filePath), { recursive: true }) + const release = await lock(filePath, { realpath: false, retries: LOCK_RETRIES }) + try { + return await apply() + } finally { + await release() + } +} diff --git a/src/main/runtime/agent-session-store-transaction-queue.ts b/src/main/runtime/agent-session-store-transaction-queue.ts new file mode 100644 index 00000000000..230ac6093a4 --- /dev/null +++ b/src/main/runtime/agent-session-store-transaction-queue.ts @@ -0,0 +1,124 @@ +import type { AgentSessionOperationRow } from '../../shared/agent-session-operation-ledger' +import type { AgentSessionRecord } from '../../shared/agent-session-record' +import { + AGENT_SESSION_STORE_SCHEMA_VERSION, + agentSessionStoreRevision, + loadAgentSessionStore, + saveAgentSessionStore, + type AgentSessionStoreState +} from './agent-session-record-store-file' +import { withAgentSessionStoreTransactionLock } from './agent-session-store-transaction-lock' + +function markLoadedLeasesUnreconciled(state: AgentSessionStoreState): void { + for (const [sessionId, record] of state.records) { + state.records.set(sessionId, { + ...record, + lease: { ...record.lease, unreconciled: true } + }) + } +} + +function mapEntriesMatch(left: ReadonlyMap, right: ReadonlyMap): boolean { + if (left.size !== right.size) { + return false + } + for (const [key, value] of left) { + if (right.get(key) !== value) { + return false + } + } + return true +} + +function agentSessionStoreStateChanged( + state: AgentSessionStoreState, + records: ReadonlyMap, + operations: ReadonlyMap, + retiredClaimKeys: AgentSessionStoreState['retiredClaimKeys'] +): boolean { + return ( + !mapEntriesMatch(state.records, records) || + !mapEntriesMatch(state.operations, operations) || + state.retiredClaimKeys.length !== retiredClaimKeys.length || + state.retiredClaimKeys.some((entry, index) => entry !== retiredClaimKeys[index]) + ) +} + +export class AgentSessionStoreTransactionQueue { + private queue: Promise = Promise.resolve() + private diskRecoveredFromBackup: boolean + + constructor( + private readonly filePath: string, + readonly hostId: string, + readonly readOnly: boolean, + readonly recoveredFromBackup: boolean, + private diskStoreFound: boolean, + public state: AgentSessionStoreState, + private diskRevision: string + ) { + this.diskRecoveredFromBackup = recoveredFromBackup + } + + transact(apply: () => T): Promise { + const run = this.queue.then(() => + withAgentSessionStoreTransactionLock(this.filePath, async () => { + if (this.readOnly) { + throw new Error('agent_session_legacy_required') + } + await this.refreshExternallyChangedState() + if (this.diskRecoveredFromBackup) { + // Why: the missing commit may hold a newer fence, so rollback cannot mint another writer. + throw new Error('execution_owner_reconciling') + } + const records = new Map(this.state.records) + const operations = new Map(this.state.operations) + const retiredClaimKeys = [...this.state.retiredClaimKeys] + try { + const result = apply() + if (!agentSessionStoreStateChanged(this.state, records, operations, retiredClaimKeys)) { + return result + } + await saveAgentSessionStore(this.filePath, this.state, { + recoveredFromBackup: this.diskRecoveredFromBackup + }) + this.state.schemaVersion = AGENT_SESSION_STORE_SCHEMA_VERSION + this.diskRevision = agentSessionStoreRevision(this.state) + this.diskRecoveredFromBackup = false + this.diskStoreFound = true + return result + } catch (error) { + this.state.records = records + this.state.operations = operations + this.state.retiredClaimKeys = retiredClaimKeys + throw error + } + }) + ) + this.queue = run.catch(() => {}) + return run + } + + private async refreshExternallyChangedState(): Promise { + const loaded = await loadAgentSessionStore(this.filePath, this.hostId) + if (this.diskStoreFound && !loaded.storeFound) { + throw new Error('agent_session_store_corrupt') + } + this.diskStoreFound ||= loaded.storeFound + const diskRevision = agentSessionStoreRevision(loaded.state) + this.diskRecoveredFromBackup = loaded.recoveredFromBackup + if (diskRevision === this.diskRevision) { + return + } + if (loaded.readOnly) { + throw new Error('agent_session_legacy_required') + } + markLoadedLeasesUnreconciled(loaded.state) + this.state = loaded.state + this.diskRevision = diskRevision + } +} + +export function markAgentSessionStoreLeasesUnreconciled(state: AgentSessionStoreState): void { + markLoadedLeasesUnreconciled(state) +} diff --git a/src/shared/agent-session-lease-adjudication.test.ts b/src/shared/agent-session-lease-adjudication.test.ts new file mode 100644 index 00000000000..782701e71c6 --- /dev/null +++ b/src/shared/agent-session-lease-adjudication.test.ts @@ -0,0 +1,266 @@ +import { describe, expect, it } from 'vitest' +import { + adjudicateAgentSessionRestart, + agentSessionLeaseAdmitsWriter, + classifyObservedAgentSessionSpawnToken, + evaluateAgentSessionAcquisition, + isProvenAliveProbe, + isProvenDeadProbe, + type AgentSessionOwnerProbe +} from './agent-session-lease-adjudication' +import type { AgentSessionLease } from './agent-session-record' + +const OWNER = { + hostId: 'local', + pid: 4242, + processStartTimeMs: 1_700_000_000_000, + spawnToken: 'spawn-a' +} + +function lease(overrides: Partial = {}): AgentSessionLease { + return { + sessionId: 'session-alpha-1', + runtimeKind: 'native', + runtimeFence: 7, + handoffStage: null, + provenHandleLinkId: 'link-1', + ownerProcess: OWNER, + reservedSpawnToken: 'spawn-a', + leaseDeadlineAt: 1_000, + lastRenewedAt: 500, + handoffOperationId: null, + journalCheckpoint: null, + claimKeyId: 'key-1', + claimStatus: 'live', + unreconciled: false, + deathEvidence: null, + ...overrides + } +} + +const MATCHED: AgentSessionOwnerProbe = { outcome: 'identity-matched', matchedOn: ['spawn-token'] } +const INDETERMINATE: AgentSessionOwnerProbe = { outcome: 'indeterminate', reason: 'no answer' } + +function acquire( + leaseState: AgentSessionLease, + probe: AgentSessionOwnerProbe, + handoffOperationId: string | null = null +) { + return evaluateAgentSessionAcquisition({ + lease: leaseState, + expectedFence: leaseState.runtimeFence, + handoffOperationId, + probe + }) +} + +describe('proof classification', () => { + it('treats a pid match with nothing PID-reuse-safe as no proof at all', () => { + // A bare pid match is exactly the case that mints a second writer after pid reuse. + expect(isProvenAliveProbe({ outcome: 'identity-matched', matchedOn: [] })).toBe(false) + expect(isProvenAliveProbe(MATCHED)).toBe(true) + expect(isProvenDeadProbe(INDETERMINATE)).toBe(false) + expect(isProvenAliveProbe(INDETERMINATE)).toBe(false) + }) + + it.each([ + ['exit-observed', { outcome: 'exit-observed' } as AgentSessionOwnerProbe], + ['pid-absent', { outcome: 'pid-absent' } as AgentSessionOwnerProbe], + [ + 'identity-mismatch', + { outcome: 'identity-mismatch', field: 'spawn-token' } as AgentSessionOwnerProbe + ] + ])('accepts %s as proof of death', (_name, probe) => { + expect(isProvenDeadProbe(probe)).toBe(true) + }) + + it('never counts a reservation probe or an indeterminate answer as death', () => { + expect(isProvenDeadProbe({ outcome: 'reservation-unused' })).toBe(false) + expect(isProvenDeadProbe(INDETERMINATE)).toBe(false) + }) +}) + +describe('acquisition compare-and-swap', () => { + it('refuses a stale fence and grants at exactly fence + 1', () => { + const held = lease({ ownerProcess: null, claimStatus: 'released', reservedSpawnToken: null }) + expect( + evaluateAgentSessionAcquisition({ + lease: held, + expectedFence: held.runtimeFence - 1, + handoffOperationId: null, + probe: MATCHED + }) + ).toEqual({ decision: 'refused', code: 'agent_session_checkpoint_stale' }) + expect(acquire(held, MATCHED)).toEqual({ decision: 'granted', nextFence: 8 }) + }) + + it('refuses the loser of a concurrent swap: only one caller sees the pre-state fence', () => { + const before = lease({ ownerProcess: null, claimStatus: 'released', reservedSpawnToken: null }) + const winner = acquire(before, MATCHED) + expect(winner).toEqual({ decision: 'granted', nextFence: 8 }) + // The loser still holds the pre-swap fence, which is no longer current. + const after = lease({ ...before, runtimeFence: 8, claimStatus: 'reserved' }) + expect( + evaluateAgentSessionAcquisition({ + lease: after, + expectedFence: 7, + handoffOperationId: null, + probe: MATCHED + }) + ).toEqual({ decision: 'refused', code: 'agent_session_checkpoint_stale' }) + }) + + it('never grants a second owner on expiry alone', () => { + // The recorded owner is long past its deadline; nothing here may consult that deadline. + const expired = lease({ leaseDeadlineAt: 1, lastRenewedAt: 1 }) + expect(acquire(expired, INDETERMINATE)).toEqual({ + decision: 'refused', + code: 'agent_session_ownership_unknown' + }) + expect(acquire(expired, MATCHED)).toEqual({ + decision: 'refused', + code: 'agent_session_conflict' + }) + expect(acquire(expired, { outcome: 'pid-absent' })).toEqual({ + decision: 'granted', + nextFence: 8 + }) + }) + + it('refuses while unreconciled even with proof the owner is dead', () => { + expect(acquire(lease({ unreconciled: true }), { outcome: 'pid-absent' })).toEqual({ + decision: 'refused', + code: 'execution_owner_reconciling' + }) + }) + + it('keeps a conflicted claim conflicted regardless of proof', () => { + expect(acquire(lease({ claimStatus: 'conflicted' }), { outcome: 'exit-observed' })).toEqual({ + decision: 'refused', + code: 'agent_session_conflict' + }) + }) + + it.each([ + ['recovering', 'agent_session_ownership_unknown'], + ['manual-recovery', 'agent_session_ownership_unknown'], + ['preparing', 'agent_session_conflict'] + ] as const)('refuses acquisition in stage %s', (handoffStage, code) => { + expect(acquire(lease({ handoffStage }), { outcome: 'pid-absent' })).toEqual({ + decision: 'refused', + code + }) + }) + + it('refuses a different handoff operation and replays the matching one', () => { + const mid = lease({ + handoffStage: 'old-owner-stopped', + handoffOperationId: 'op-1', + ownerProcess: null, + claimStatus: 'reserved' + }) + expect(acquire(mid, { outcome: 'reservation-unused' }, 'op-2')).toEqual({ + decision: 'refused', + code: 'agent_session_operation_conflict' + }) + expect(acquire(mid, { outcome: 'reservation-unused' }, 'op-1')).toEqual({ + decision: 'retry-reservation', + fence: 7 + }) + }) + + it('refuses a reservation whose spawn may have won the race with the crash', () => { + const reserved = lease({ ownerProcess: null, claimStatus: 'reserved', handoffStage: null }) + expect(acquire(reserved, INDETERMINATE)).toEqual({ + decision: 'refused', + code: 'agent_session_ownership_unknown' + }) + expect(acquire(reserved, { outcome: 'reservation-unused' })).toEqual({ + decision: 'granted', + nextFence: 8 + }) + }) +}) + +describe('restart reconciliation', () => { + it('re-adopts a proven-live owner without moving the fence', () => { + expect( + adjudicateAgentSessionRestart({ lease: lease(), probe: MATCHED, observedAt: 9_000 }) + ).toEqual({ disposition: 'readopt' }) + }) + + it('bumps the fence exactly once for a proven-dead owner and records the evidence', () => { + const result = adjudicateAgentSessionRestart({ + lease: lease(), + probe: { outcome: 'identity-mismatch', field: 'process-start-time' }, + observedAt: 9_000 + }) + expect(result).toEqual({ + disposition: 'evicted', + nextFence: 8, + evidence: { + kind: 'identity-mismatch', + detail: 'mismatched process-start-time', + observedAt: 9_000 + } + }) + }) + + it('sends an unverifiable owner to recovery instead of evicting it', () => { + expect( + adjudicateAgentSessionRestart({ + lease: lease({ leaseDeadlineAt: 1 }), + probe: INDETERMINATE, + observedAt: 9_000 + }) + ).toEqual({ disposition: 'recovering', stage: 'recovering', reason: 'no answer' }) + }) + + it('keeps a pre-restart conflict conflicted', () => { + expect( + adjudicateAgentSessionRestart({ + lease: lease({ claimStatus: 'conflicted' }), + probe: { outcome: 'pid-absent' }, + observedAt: 9_000 + }) + ).toEqual({ disposition: 'conflicted', reason: 'claim conflicted before restart' }) + }) + + it('frees a reservation only when nothing ever spawned', () => { + const reserved = lease({ ownerProcess: null, claimStatus: 'reserved' }) + expect( + adjudicateAgentSessionRestart({ + lease: reserved, + probe: { outcome: 'reservation-unused' }, + observedAt: 9_000 + }) + ).toMatchObject({ disposition: 'evicted', nextFence: 8 }) + expect( + adjudicateAgentSessionRestart({ lease: reserved, probe: INDETERMINATE, observedAt: 9_000 }) + ).toMatchObject({ disposition: 'recovering' }) + }) +}) + +describe('writer admission and orphan spawn tokens', () => { + it('admits a writer only when reconciled, settled, live, and holding a process', () => { + expect(agentSessionLeaseAdmitsWriter(lease())).toBe(true) + expect(agentSessionLeaseAdmitsWriter(lease({ unreconciled: true }))).toBe(false) + expect(agentSessionLeaseAdmitsWriter(lease({ handoffStage: 'new-owner-proving' }))).toBe(false) + expect(agentSessionLeaseAdmitsWriter(lease({ claimStatus: 'reserved' }))).toBe(false) + expect(agentSessionLeaseAdmitsWriter(lease({ ownerProcess: null }))).toBe(false) + }) + + it('calls a spawn token with no matching lease an orphan', () => { + const leases = [lease(), lease({ sessionId: 'session-beta-1', reservedSpawnToken: 'spawn-b' })] + expect(classifyObservedAgentSessionSpawnToken({ spawnToken: 'spawn-a', leases })).toBe('owned') + expect(classifyObservedAgentSessionSpawnToken({ spawnToken: 'spawn-b', leases })).toBe('owned') + expect(classifyObservedAgentSessionSpawnToken({ spawnToken: 'spawn-z', leases })).toBe('orphan') + }) + + it('still recognises an owner whose reservation token was cleared after proving', () => { + const proved = lease({ reservedSpawnToken: null }) + expect( + classifyObservedAgentSessionSpawnToken({ spawnToken: 'spawn-a', leases: [proved] }) + ).toBe('owned') + }) +}) diff --git a/src/shared/agent-session-lease-adjudication.ts b/src/shared/agent-session-lease-adjudication.ts new file mode 100644 index 00000000000..c847c473147 --- /dev/null +++ b/src/shared/agent-session-lease-adjudication.ts @@ -0,0 +1,225 @@ +/** + * Single-writer lease adjudication. + * + * Every decision here fails closed: expiry alone never grants a second owner, an unverifiable + * process counts as possibly alive, and a stage that cannot prove an owner ends in manual + * recovery rather than handing the session to the other runtime. This is the opposite polarity + * from daemon adoption checks, which fail open on a missing start time — a wrong answer there + * refuses an adoption, a wrong answer here creates two writers on one provider session. + */ + +import type { + AgentSessionDeathEvidence, + AgentSessionHandoffStage, + AgentSessionLease +} from './agent-session-record' + +export type AgentSessionIdentityMatchField = 'process-start-time' | 'spawn-token' + +export type AgentSessionOwnerProbe = + /** Orca watched this exact process exit. */ + | { outcome: 'exit-observed' } + /** The recorded pid is not present on the host. */ + | { outcome: 'pid-absent' } + /** The pid is present but is a different process. */ + | { outcome: 'identity-mismatch'; field: AgentSessionIdentityMatchField | 'command-line' } + /** The pid is present and at least one identity element was verified. */ + | { outcome: 'identity-matched'; matchedOn: readonly AgentSessionIdentityMatchField[] } + /** No process carries the reserved spawn token and the provider saw no activity after it. */ + | { outcome: 'reservation-unused' } + /** The host could not answer — restricted container, no start time, no token echo. */ + | { outcome: 'indeterminate'; reason: string } + +export type AgentSessionLeaseRefusalCode = + | 'agent_session_checkpoint_stale' + | 'agent_session_conflict' + | 'agent_session_ownership_unknown' + | 'agent_session_operation_conflict' + | 'execution_owner_reconciling' + +export type AgentSessionAcquisitionDecision = + | { decision: 'granted'; nextFence: number } + /** The same handoff operation re-entering its own reservation; no new fence, no new spawn. */ + | { decision: 'retry-reservation'; fence: number } + | { decision: 'refused'; code: AgentSessionLeaseRefusalCode } + +export type AgentSessionRestartAdjudication = + | { disposition: 'readopt' } + | { disposition: 'evicted'; nextFence: number; evidence: AgentSessionDeathEvidence } + | { disposition: 'recovering'; stage: AgentSessionHandoffStage; reason: string } + | { disposition: 'conflicted'; reason: string } + +/** Stages that can legally admit a new owner at all; the rest have an owner or no evidence. */ +const STAGES_ADMITTING_NEW_OWNER: ReadonlySet = new Set([ + 'old-owner-stopped', + 'new-owner-proving' +]) + +export function isProvenDeadProbe(probe: AgentSessionOwnerProbe): boolean { + return ( + probe.outcome === 'exit-observed' || + probe.outcome === 'pid-absent' || + probe.outcome === 'identity-mismatch' + ) +} + +/** + * A matched pid is only proof of life when something PID-reuse-safe matched with it. A bare pid + * match on a host that can produce neither a start time nor a token echo is indeterminate. + */ +export function isProvenAliveProbe(probe: AgentSessionOwnerProbe): boolean { + return probe.outcome === 'identity-matched' && probe.matchedOn.length > 0 +} + +function deathEvidenceFor( + probe: AgentSessionOwnerProbe, + observedAt: number +): AgentSessionDeathEvidence | null { + if (probe.outcome === 'exit-observed') { + return { kind: 'exit-observed', detail: 'observed process exit', observedAt } + } + if (probe.outcome === 'pid-absent') { + return { kind: 'pid-absent', detail: 'recorded pid absent on host', observedAt } + } + if (probe.outcome === 'identity-mismatch') { + return { kind: 'identity-mismatch', detail: `mismatched ${probe.field}`, observedAt } + } + return null +} + +/** True when the recorded owner may write right now. Used by every mutating path in later parts. */ +export function agentSessionLeaseAdmitsWriter(lease: AgentSessionLease): boolean { + return ( + !lease.unreconciled && + lease.handoffStage === null && + lease.claimStatus === 'live' && + lease.ownerProcess !== null + ) +} + +export function isAgentSessionFenceCurrent(lease: AgentSessionLease, fence: number): boolean { + return Number.isSafeInteger(fence) && fence === lease.runtimeFence +} + +/** + * Compare-and-swap acquisition. `probe` describes what the host could prove about the recorded + * owner; it is only consulted when a recorded owner or an unused reservation stands in the way. + */ +export function evaluateAgentSessionAcquisition(args: { + lease: AgentSessionLease + expectedFence: number + handoffOperationId: string | null + probe: AgentSessionOwnerProbe +}): AgentSessionAcquisitionDecision { + const { lease, expectedFence, handoffOperationId, probe } = args + if (lease.unreconciled) { + return { decision: 'refused', code: 'execution_owner_reconciling' } + } + if (!isAgentSessionFenceCurrent(lease, expectedFence)) { + return { decision: 'refused', code: 'agent_session_checkpoint_stale' } + } + if (lease.claimStatus === 'conflicted') { + return { decision: 'refused', code: 'agent_session_conflict' } + } + if (lease.handoffStage === 'recovering' || lease.handoffStage === 'manual-recovery') { + // Why: no stage expires into an owner; recovery is resolved by proof or by the user. + return { decision: 'refused', code: 'agent_session_ownership_unknown' } + } + if (lease.handoffStage === 'preparing') { + // Why: the old owner is quiesced but alive and still authoritative. + return { decision: 'refused', code: 'agent_session_conflict' } + } + if (lease.handoffStage !== null && !STAGES_ADMITTING_NEW_OWNER.has(lease.handoffStage)) { + return { decision: 'refused', code: 'agent_session_conflict' } + } + if (lease.handoffStage !== null && lease.handoffOperationId !== null) { + if (handoffOperationId !== lease.handoffOperationId) { + // Why: the retry key is operation id + fence + stage; a different id is a different intent. + return { decision: 'refused', code: 'agent_session_operation_conflict' } + } + if (lease.ownerProcess === null && lease.handoffStage === 'old-owner-stopped') { + // Why: an idempotent re-run of a reservation that already exists at this fence. + return { decision: 'retry-reservation', fence: lease.runtimeFence } + } + } + if (lease.ownerProcess !== null) { + if (!isProvenDeadProbe(probe)) { + // Why: a lapsed deadline means Orca stopped hearing from the owner, not that the child + // stopped editing files and spending tokens. + return { + decision: 'refused', + code: isProvenAliveProbe(probe) + ? 'agent_session_conflict' + : 'agent_session_ownership_unknown' + } + } + return { decision: 'granted', nextFence: lease.runtimeFence + 1 } + } + if (lease.claimStatus === 'reserved' && probe.outcome !== 'reservation-unused') { + // Why: a reservation with no proven process is not a free lease — the crash may have lost + // the race with the spawn rather than beaten it. + return { decision: 'refused', code: 'agent_session_ownership_unknown' } + } + return { decision: 'granted', nextFence: lease.runtimeFence + 1 } +} + +/** + * Host-restart reconciliation for one persisted lease. Every lease is unreconciled at load and + * grants no writer until this returns. + */ +export function adjudicateAgentSessionRestart(args: { + lease: AgentSessionLease + probe: AgentSessionOwnerProbe + observedAt: number +}): AgentSessionRestartAdjudication { + const { lease, probe, observedAt } = args + if (lease.claimStatus === 'conflicted') { + // Why: the conflict outlives the process that observed it; resolving to free would hand the + // provider session to whichever side restarted first. + return { disposition: 'conflicted', reason: 'claim conflicted before restart' } + } + if (lease.ownerProcess === null) { + if (probe.outcome === 'reservation-unused') { + return { + disposition: 'evicted', + nextFence: lease.runtimeFence + 1, + evidence: { kind: 'pid-absent', detail: 'reservation never spawned', observedAt } + } + } + return { + disposition: 'recovering', + stage: 'recovering', + reason: 'reservation with no proven process' + } + } + if (isProvenAliveProbe(probe)) { + // Why: re-adoption is not a new generation, so the fence does not move. + return { disposition: 'readopt' } + } + const evidence = deathEvidenceFor(probe, observedAt) + if (evidence) { + return { disposition: 'evicted', nextFence: lease.runtimeFence + 1, evidence } + } + return { + disposition: 'recovering', + stage: 'recovering', + reason: + probe.outcome === 'indeterminate' ? probe.reason : 'process identity could not be verified' + } +} + +/** + * A process carrying an Orca spawn token with no matching lease is an orphan: stop it, never + * adopt it. Neither age nor CPU is evidence — only a token match justifies acting on a process. + */ +export function classifyObservedAgentSessionSpawnToken(args: { + spawnToken: string + leases: readonly AgentSessionLease[] +}): 'owned' | 'orphan' { + const owned = args.leases.some( + (lease) => + lease.reservedSpawnToken === args.spawnToken || + lease.ownerProcess?.spawnToken === args.spawnToken + ) + return owned ? 'owned' : 'orphan' +} diff --git a/src/shared/agent-session-operation-ledger.test.ts b/src/shared/agent-session-operation-ledger.test.ts new file mode 100644 index 00000000000..0710eade461 --- /dev/null +++ b/src/shared/agent-session-operation-ledger.test.ts @@ -0,0 +1,174 @@ +import { describe, expect, it } from 'vitest' +import { + AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS, + AGENT_SESSION_OPERATION_FUTURE_SKEW_MS +} from './agent-session-host-authority' +import { + agentSessionOperationExpiry, + agentSessionOperationKey, + evaluateAgentSessionOperation, + isAgentSessionOperationRow, + pruneAgentSessionOperationRows, + type AgentSessionOperationRow +} from './agent-session-operation-ledger' + +const NOW = 1_800_000_000_000 + +function operationId(timestamp: number, suffix = 'a'.repeat(32)): string { + return `${String(timestamp).padStart(13, '0')}-${suffix}` +} + +function evaluate( + rows: Map, + overrides: Partial<{ + callerKey: string + operationId: string + fingerprint: string + now: number + perClientLimit: number + globalLimit: number + }> = {} +) { + return evaluateAgentSessionOperation({ + rows, + callerKey: 'client-1', + operationId: operationId(NOW), + fingerprint: 'fp-1', + now: NOW, + ...overrides + }) +} + +function admit( + rows: Map, + overrides: Parameters[1] = {} +): AgentSessionOperationRow { + const decision = evaluate(rows, overrides) + if (decision.decision !== 'admit') { + throw new Error(`expected admit, got ${decision.decision}`) + } + rows.set(agentSessionOperationKey(decision.row.callerKey, decision.row.operationId), decision.row) + return decision.row +} + +describe('operation admission', () => { + it('admits a fresh id once and replays the identical retry', () => { + const rows = new Map() + const row = admit(rows) + const replay = evaluate(rows) + expect(replay).toEqual({ decision: 'replay', row }) + }) + + it('refuses the same id carrying different parameters', () => { + const rows = new Map() + admit(rows) + expect(evaluate(rows, { fingerprint: 'fp-2' })).toEqual({ + decision: 'refused', + code: 'agent_session_operation_conflict' + }) + }) + + it('scopes ids per caller so two clients cannot collide or replay each other', () => { + const rows = new Map() + admit(rows) + expect(evaluate(rows, { callerKey: 'client-2' }).decision).toBe('admit') + }) + + it('refuses a malformed or future-dated id', () => { + const rows = new Map() + expect(evaluate(rows, { operationId: 'not-an-operation-id' })).toEqual({ + decision: 'refused', + code: 'agent_session_operation_invalid' + }) + // Why: a future-dated id would look new again after its own tombstone is collected. + expect( + evaluate(rows, { + operationId: operationId(NOW + AGENT_SESSION_OPERATION_FUTURE_SKEW_MS + 1) + }) + ).toEqual({ decision: 'refused', code: 'agent_session_operation_invalid' }) + expect( + evaluate(rows, { operationId: operationId(NOW + AGENT_SESSION_OPERATION_FUTURE_SKEW_MS) }) + .decision + ).toBe('admit') + }) + + it('refuses an id older than the admission window instead of treating it as new', () => { + const rows = new Map() + const stale = operationId(NOW - AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS - 1) + expect(evaluate(rows, { operationId: stale })).toEqual({ + decision: 'refused', + code: 'agent_session_operation_expired' + }) + expect( + evaluate(rows, { operationId: operationId(NOW - AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS) }) + .decision + ).toBe('admit') + }) + + it('refuses new ids at the per-client and global caps rather than evicting tombstones', () => { + const rows = new Map() + admit(rows, { operationId: operationId(NOW, 'b'.repeat(32)) }) + expect(evaluate(rows, { perClientLimit: 1 })).toEqual({ + decision: 'refused', + code: 'agent_session_operation_capacity' + }) + // A different caller is still refused once the global cap is reached. + expect(evaluate(rows, { callerKey: 'client-2', globalLimit: 1 })).toEqual({ + decision: 'refused', + code: 'agent_session_operation_capacity' + }) + expect(evaluate(rows, { callerKey: 'client-2', perClientLimit: 1 }).decision).toBe('admit') + }) +}) + +describe('retention', () => { + it('keeps a tombstone strictly longer than its id can be admitted as new', () => { + const expiry = agentSessionOperationExpiry(NOW, NOW) + const lastAdmissibleAt = NOW + AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS + expect(expiry).toBeGreaterThan(lastAdmissibleAt) + // Why: a retry landing in that gap would become a second spawn instead of a replay. + const rows = new Map() + admit(rows) + expect(pruneAgentSessionOperationRows(rows, lastAdmissibleAt).size).toBe(1) + expect(evaluate(pruneAgentSessionOperationRows(rows, lastAdmissibleAt)).decision).toBe('replay') + }) + + it('anchors retention to the later of recording and stamping', () => { + const late = agentSessionOperationExpiry(NOW + 10_000, NOW) + expect(late).toBe(agentSessionOperationExpiry(NOW + 10_000, NOW + 10_000)) + expect(late).toBeGreaterThan(agentSessionOperationExpiry(NOW, NOW)) + }) + + it('drops only rows past their own expiry', () => { + const rows = new Map() + const row = admit(rows) + expect(pruneAgentSessionOperationRows(rows, row.expiresAt).size).toBe(0) + expect(pruneAgentSessionOperationRows(rows, row.expiresAt - 1).size).toBe(1) + }) +}) + +describe('persisted row validation', () => { + it('accepts every recorded outcome shape', () => { + const rows = new Map() + const row = admit(rows) + expect(isAgentSessionOperationRow(row)).toBe(true) + expect( + isAgentSessionOperationRow({ ...row, outcome: { status: 'succeeded', sessionId: 's-1' } }) + ).toBe(true) + expect(isAgentSessionOperationRow({ ...row, outcome: { status: 'unknown' } })).toBe(true) + expect(isAgentSessionOperationRow({ ...row, outcome: { status: 'failed', code: 'x' } })).toBe( + true + ) + }) + + it('rejects rows a later build could misread', () => { + const rows = new Map() + const row = admit(rows) + expect(isAgentSessionOperationRow({ ...row, operationId: 'garbage' })).toBe(false) + expect(isAgentSessionOperationRow({ ...row, callerKey: '' })).toBe(false) + expect(isAgentSessionOperationRow({ ...row, expiresAt: 1.5 })).toBe(false) + expect(isAgentSessionOperationRow({ ...row, outcome: { status: 'succeeded' } })).toBe(false) + expect(isAgentSessionOperationRow({ ...row, outcome: null })).toBe(false) + expect(isAgentSessionOperationRow(null)).toBe(false) + }) +}) diff --git a/src/shared/agent-session-operation-ledger.ts b/src/shared/agent-session-operation-ledger.ts new file mode 100644 index 00000000000..864c2c66485 --- /dev/null +++ b/src/shared/agent-session-operation-ledger.ts @@ -0,0 +1,173 @@ +/** + * Durable client-operation ledger. + * + * `terminal.ensureAgentSession` / `terminal.createAgentSession` already enforce timestamped + * operation ids with fingerprint conflict detection, age expiry, capacity limits, and tombstone + * retention — but in memory, so a host restart turns "replay this create" into "spawn another + * agent". These are the same rules over rows that survive a restart; the store writes a row in + * the same atomic transaction as the lease reservation. + */ + +import { + AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS, + AGENT_SESSION_OPERATION_FUTURE_SKEW_MS, + parseAgentSessionOperationTimestamp +} from './agent-session-host-authority' + +export const AGENT_SESSION_DURABLE_OPERATION_PER_CLIENT_LIMIT = 512 +export const AGENT_SESSION_DURABLE_OPERATION_GLOBAL_LIMIT = 4_096 + +export type AgentSessionOperationOutcome = + | { status: 'pending' } + | { status: 'succeeded'; sessionId: string } + | { status: 'failed'; code: string } + /** The effect may or may not have happened; replay this answer instead of spawning again. */ + | { status: 'unknown' } + +export type AgentSessionOperationRow = { + callerKey: string + operationId: string + fingerprint: string + operationTimestamp: number + recordedAt: number + expiresAt: number + outcome: AgentSessionOperationOutcome +} + +export type AgentSessionOperationRefusalCode = + | 'agent_session_operation_invalid' + | 'agent_session_operation_conflict' + | 'agent_session_operation_expired' + | 'agent_session_operation_capacity' + +export type AgentSessionOperationDecision = + | { decision: 'replay'; row: AgentSessionOperationRow } + | { decision: 'admit'; row: AgentSessionOperationRow } + | { decision: 'refused'; code: AgentSessionOperationRefusalCode } + +/** NUL cannot occur in a caller key or operation id, so no pair can forge another pair's key. */ +const OPERATION_KEY_SEPARATOR = '\u0000' + +export function agentSessionOperationKey(callerKey: string, operationId: string): string { + return `${callerKey}${OPERATION_KEY_SEPARATOR}${operationId}` +} + +/** + * 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 + * spawn instead of a replay. + */ +export function agentSessionOperationExpiry( + operationTimestamp: number, + recordedAt: number +): number { + return ( + Math.max(recordedAt, operationTimestamp) + + AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS + + AGENT_SESSION_OPERATION_FUTURE_SKEW_MS + ) +} + +export function pruneAgentSessionOperationRows( + rows: ReadonlyMap, + now: number +): Map { + const kept = new Map() + for (const [key, row] of rows) { + if (row.expiresAt > now) { + kept.set(key, row) + } + } + return kept +} + +/** + * Decide what a mutating call with this operation id means against the persisted ledger. Callers + * must prune first; a row that is present is a row that is still authoritative. + */ +export function evaluateAgentSessionOperation(args: { + rows: ReadonlyMap + callerKey: string + operationId: string + fingerprint: string + now: number + perClientLimit?: number + globalLimit?: number +}): AgentSessionOperationDecision { + const { rows, callerKey, operationId, fingerprint, now } = args + const operationTimestamp = parseAgentSessionOperationTimestamp(operationId) + if ( + operationTimestamp === null || + operationTimestamp > now + AGENT_SESSION_OPERATION_FUTURE_SKEW_MS + ) { + // Why: a future-dated id could look new again after its tombstone is collected. + return { decision: 'refused', code: 'agent_session_operation_invalid' } + } + const key = agentSessionOperationKey(callerKey, operationId) + const existing = rows.get(key) + if (existing) { + return existing.fingerprint === fingerprint + ? { decision: 'replay', row: existing } + : { decision: 'refused', code: 'agent_session_operation_conflict' } + } + if (now - operationTimestamp > AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS) { + // Why: once a tombstone could have expired, an unseen replay must never be reinterpreted as + // permission to start another fresh agent. + return { decision: 'refused', code: 'agent_session_operation_expired' } + } + const perClientLimit = args.perClientLimit ?? AGENT_SESSION_DURABLE_OPERATION_PER_CLIENT_LIMIT + const globalLimit = args.globalLimit ?? AGENT_SESSION_DURABLE_OPERATION_GLOBAL_LIMIT + let callerCount = 0 + for (const row of rows.values()) { + if (row.callerKey === callerKey) { + callerCount += 1 + } + } + if (callerCount >= perClientLimit || rows.size >= globalLimit) { + // Why: tombstones cannot be evicted early without making an old replay capable of spawning + // again; reject new ids until retained rows age out. + return { decision: 'refused', code: 'agent_session_operation_capacity' } + } + return { + decision: 'admit', + row: { + callerKey, + operationId, + fingerprint, + operationTimestamp, + recordedAt: now, + expiresAt: agentSessionOperationExpiry(operationTimestamp, now), + outcome: { status: 'pending' } + } + } +} + +const OPERATION_ID_MAX_LENGTH = 128 + +export function isAgentSessionOperationRow(value: unknown): value is AgentSessionOperationRow { + if (typeof value !== 'object' || value === null) { + return false + } + const row = value as Partial + const outcome = row.outcome as AgentSessionOperationOutcome | undefined + const outcomeValid = + typeof outcome === 'object' && + outcome !== null && + ((outcome.status === 'pending' && true) || + (outcome.status === 'succeeded' && typeof outcome.sessionId === 'string') || + (outcome.status === 'failed' && typeof outcome.code === 'string') || + outcome.status === 'unknown') + return ( + typeof row.callerKey === 'string' && + row.callerKey.length > 0 && + typeof row.operationId === 'string' && + row.operationId.length <= OPERATION_ID_MAX_LENGTH && + parseAgentSessionOperationTimestamp(row.operationId) !== null && + typeof row.fingerprint === 'string' && + row.fingerprint.length > 0 && + Number.isSafeInteger(row.operationTimestamp) && + Number.isSafeInteger(row.recordedAt) && + Number.isSafeInteger(row.expiresAt) && + outcomeValid + ) +} diff --git a/src/shared/agent-session-provider-handle.test.ts b/src/shared/agent-session-provider-handle.test.ts new file mode 100644 index 00000000000..4605a2c1129 --- /dev/null +++ b/src/shared/agent-session-provider-handle.test.ts @@ -0,0 +1,294 @@ +import { describe, expect, it } from 'vitest' +import { + agentSessionProviderHandleChainHead, + agentSessionProviderHandleKey, + agentSessionProviderHandleRoot, + agentSessionProviderHandlesEqual, + appendAgentSessionProviderHandleLink, + findAgentSessionProviderHandleLink, + isAgentSessionProviderHandle, + isAgentSessionProviderHandleChain, + MAX_AGENT_SESSION_PROVIDER_HANDLE_LINKS, + type AgentSessionProviderHandle, + type AgentSessionProviderHandleLink +} from './agent-session-provider-handle' + +const CLAUDE: AgentSessionProviderHandle = { + provider: 'claude', + sessionId: 'sess-1', + leafUuid: 'leaf-1' +} + +function link(overrides: Partial = {}) { + return { + linkId: 'link-1', + handle: CLAUDE, + origin: 'created', + mintedAtFence: 1, + observedAt: 1_000, + ...overrides + } as AgentSessionProviderHandleLink +} + +describe('handle identity', () => { + it('keys a Claude handle by session id AND leaf, so two branches are two handles', () => { + // Concurrent resumes branch one transcript silently; the session id alone cannot name a writer. + const branchA = agentSessionProviderHandleKey(CLAUDE) + const branchB = agentSessionProviderHandleKey({ ...CLAUDE, leafUuid: 'leaf-2' }) + expect(branchA).not.toEqual(branchB) + expect(agentSessionProviderHandleRoot(CLAUDE)).toEqual( + agentSessionProviderHandleRoot({ ...CLAUDE, leafUuid: 'leaf-2' }) + ) + }) + + it('keys a Codex handle by thread id alone', () => { + const codex: AgentSessionProviderHandle = { provider: 'codex', threadId: 'thread-1' } + expect(agentSessionProviderHandleKey(codex)).toBe('codex:"thread-1"') + expect(agentSessionProviderHandleRoot(codex)).toBe('codex:"thread-1"') + expect( + agentSessionProviderHandlesEqual(codex, { provider: 'codex', threadId: 'thread-2' }) + ).toBe(false) + }) + + it('distinguishes a null leaf from an empty-string leaf and rejects malformed handles', () => { + expect(isAgentSessionProviderHandle({ ...CLAUDE, leafUuid: null })).toBe(true) + expect(isAgentSessionProviderHandle({ ...CLAUDE, leafUuid: '' })).toBe(false) + expect(isAgentSessionProviderHandle({ provider: 'claude', sessionId: '' })).toBe(false) + expect(isAgentSessionProviderHandle({ provider: 'gemini', sessionId: 'x' })).toBe(false) + expect(isAgentSessionProviderHandle({ ...CLAUDE, sessionId: ' sess-1 ' })).toBe(false) + }) + + it('uses collision-free keys when Claude ids contain delimiters', () => { + const left: AgentSessionProviderHandle = { + provider: 'claude', + sessionId: 'a#b', + leafUuid: 'c' + } + const right: AgentSessionProviderHandle = { + provider: 'claude', + sessionId: 'a', + leafUuid: 'b#c' + } + expect(agentSessionProviderHandleKey(left)).not.toBe(agentSessionProviderHandleKey(right)) + expect(agentSessionProviderHandlesEqual(left, right)).toBe(false) + }) +}) + +describe('chain append', () => { + it('starts only from a created or adopted link', () => { + expect(appendAgentSessionProviderHandleLink([], link())).toEqual([link()]) + expect(appendAgentSessionProviderHandleLink([], link({ origin: 'adopted' }))).toHaveLength(1) + expect(() => appendAgentSessionProviderHandleLink([], link({ origin: 'resumed' }))).toThrow( + 'agent_session_provider_handle_invalid' + ) + }) + + it('refuses to record a fork as a resume', () => { + // --fork-session keeps the original item ids; calling it a resume would claim continuity the + // provider never gave. + const chain = [link()] + expect(() => + appendAgentSessionProviderHandleLink( + chain, + link({ + linkId: 'link-2', + origin: 'resumed', + handle: { provider: 'claude', sessionId: 'sess-2', leafUuid: 'leaf-9' }, + mintedAtFence: 2 + }) + ) + ).toThrow('agent_session_provider_handle_forked') + }) + + it('records a fork only with a new root and the seed it came from', () => { + const chain = [link()] + const forked = link({ + linkId: 'link-2', + origin: 'forked', + handle: { provider: 'claude', sessionId: 'sess-2', leafUuid: 'leaf-9' }, + mintedAtFence: 2, + forkedFromKey: agentSessionProviderHandleKey(CLAUDE) + }) + expect(appendAgentSessionProviderHandleLink(chain, forked)).toHaveLength(2) + expect(() => + appendAgentSessionProviderHandleLink(chain, { ...forked, forkedFromKey: 'claude:other' }) + ).toThrow('agent_session_provider_handle_invalid') + expect(() => + appendAgentSessionProviderHandleLink(chain, { + ...forked, + handle: CLAUDE, + forkedFromKey: agentSessionProviderHandleKey(CLAUDE) + }) + ).toThrow('agent_session_provider_handle_invalid') + }) + + it('rejects a link minted under an older fence', () => { + const chain = [link({ mintedAtFence: 5 })] + expect(() => + appendAgentSessionProviderHandleLink( + chain, + link({ + linkId: 'link-2', + origin: 'resumed', + handle: { ...CLAUDE, leafUuid: 'leaf-2' }, + mintedAtFence: 4 + }) + ) + ).toThrow('agent_session_provider_handle_stale_fence') + }) + + it('rejects a provider change mid-chain', () => { + expect(() => + appendAgentSessionProviderHandleLink( + [link()], + link({ + linkId: 'link-2', + origin: 'resumed', + handle: { provider: 'codex', threadId: 'thread-1' }, + mintedAtFence: 2 + }) + ) + ).toThrow('agent_session_provider_handle_provider_mismatch') + }) + + it('treats re-proving the same handle at the same fence as a retry, not a new link', () => { + const chain = [link({ mintedAtFence: 3 })] + const retried = appendAgentSessionProviderHandleLink( + chain, + link({ linkId: 'link-2', origin: 'resumed', mintedAtFence: 3 }) + ) + expect(retried).toHaveLength(1) + expect(retried[0]?.linkId).toBe('link-1') + // A later fence on the same handle is a genuine re-acquisition and does append. + expect( + appendAgentSessionProviderHandleLink( + chain, + link({ linkId: 'link-2', origin: 'resumed', mintedAtFence: 4 }) + ) + ).toHaveLength(2) + }) + + it('rejects reuse of a stable link id for a different proof', () => { + expect(() => + appendAgentSessionProviderHandleLink( + [link()], + link({ + origin: 'resumed', + handle: { ...CLAUDE, leafUuid: 'leaf-2' }, + mintedAtFence: 2 + }) + ) + ).toThrow('agent_session_provider_handle_invalid') + }) + + it('refuses to grow past the cap rather than dropping fork provenance', () => { + const chain: AgentSessionProviderHandleLink[] = [link()] + for (let index = 1; index < MAX_AGENT_SESSION_PROVIDER_HANDLE_LINKS; index += 1) { + chain.push( + link({ + linkId: `link-${index + 1}`, + origin: 'resumed', + handle: { ...CLAUDE, leafUuid: `leaf-${index + 1}` }, + mintedAtFence: index + 1 + }) + ) + } + expect(chain).toHaveLength(MAX_AGENT_SESSION_PROVIDER_HANDLE_LINKS) + expect(() => + appendAgentSessionProviderHandleLink( + chain, + link({ + linkId: 'link-overflow', + origin: 'resumed', + handle: { ...CLAUDE, leafUuid: 'leaf-overflow' }, + mintedAtFence: 999 + }) + ) + ).toThrow('agent_session_provider_handle_chain_overflow') + expect(isAgentSessionProviderHandleChain(chain)).toBe(true) + expect(agentSessionProviderHandleChainHead(chain)?.linkId).toBe( + `link-${MAX_AGENT_SESSION_PROVIDER_HANDLE_LINKS}` + ) + }) + + it('never mutates the chain it was given', () => { + const chain = [link()] + appendAgentSessionProviderHandleLink( + chain, + link({ + linkId: 'link-2', + origin: 'resumed', + handle: { ...CLAUDE, leafUuid: 'leaf-2' }, + mintedAtFence: 2 + }) + ) + expect(chain).toHaveLength(1) + }) +}) + +describe('chain lookup and validation', () => { + it('finds a link by id and reports the head', () => { + const chain = appendAgentSessionProviderHandleLink( + [link()], + link({ + linkId: 'link-2', + origin: 'resumed', + handle: { ...CLAUDE, leafUuid: 'leaf-2' }, + mintedAtFence: 2 + }) + ) + expect(findAgentSessionProviderHandleLink(chain, 'link-1')?.origin).toBe('created') + expect(findAgentSessionProviderHandleLink(chain, 'missing')).toBeNull() + expect(agentSessionProviderHandleChainHead(chain)?.linkId).toBe('link-2') + expect(agentSessionProviderHandleChainHead([])).toBeNull() + }) + + it('rejects a persisted chain that is over the cap or holds a malformed link', () => { + expect(isAgentSessionProviderHandleChain([{ ...link(), mintedAtFence: -1 }])).toBe(false) + expect(isAgentSessionProviderHandleChain([{ ...link(), linkId: 'not a link id!' }])).toBe(false) + expect(isAgentSessionProviderHandleChain([{ ...link(), forkedFromKey: 'claude:seed' }])).toBe( + false + ) + expect( + isAgentSessionProviderHandleChain( + Array.from({ length: MAX_AGENT_SESSION_PROVIDER_HANDLE_LINKS + 1 }, (_value, index) => + link({ linkId: `link-${index}` }) + ) + ) + ).toBe(false) + }) + + it('rejects persisted chains that bypass append invariants', () => { + expect( + isAgentSessionProviderHandleChain([ + link(), + link({ + linkId: 'link-2', + origin: 'created', + mintedAtFence: 2 + }) + ]) + ).toBe(false) + expect( + isAgentSessionProviderHandleChain([ + link(), + link({ + origin: 'resumed', + handle: { ...CLAUDE, leafUuid: 'leaf-2' }, + mintedAtFence: 2 + }) + ]) + ).toBe(false) + expect( + isAgentSessionProviderHandleChain([ + link(), + link({ + linkId: 'link-2', + origin: 'resumed', + handle: { provider: 'claude', sessionId: 'sess-2', leafUuid: 'leaf-2' }, + mintedAtFence: 2 + }) + ]) + ).toBe(false) + }) +}) diff --git a/src/shared/agent-session-provider-handle.ts b/src/shared/agent-session-provider-handle.ts new file mode 100644 index 00000000000..f36be80672c --- /dev/null +++ b/src/shared/agent-session-provider-handle.ts @@ -0,0 +1,212 @@ +/** + * Durable provider handle chain for an agent session. + * + * Handles are keyed per provider because the two structured lanes disagree about what + * identifies a conversation: a Claude session id alone can name two sibling branches inside one + * transcript, so the leaf uuid is part of the key; a Codex thread id is the whole key. Resumes + * extend the chain, forks start a new identity root, and the chain records which is which so a + * fork is never presented as a resume. + */ + +export const AGENT_SESSION_PROVIDER_HANDLE_PROVIDERS = ['claude', 'codex'] as const + +export type AgentSessionHandleProvider = (typeof AGENT_SESSION_PROVIDER_HANDLE_PROVIDERS)[number] + +export type AgentSessionProviderHandle = + | { provider: 'claude'; sessionId: string; leafUuid: string | null } + | { provider: 'codex'; threadId: string } + +export type AgentSessionProviderHandleOrigin = 'created' | 'adopted' | 'resumed' | 'forked' + +export type AgentSessionProviderHandleLink = { + /** Stable id so a lease can name the exact link its owner proved. */ + linkId: string + handle: AgentSessionProviderHandle + origin: AgentSessionProviderHandleOrigin + /** Runtime fence in force when this link was minted; never decreases along the chain. */ + mintedAtFence: number + observedAt: number + /** Key of the link a fork was seeded from. Only set when `origin` is `forked`. */ + forkedFromKey?: string +} + +export type AgentSessionProviderHandleChain = readonly AgentSessionProviderHandleLink[] + +/** Bounded so one session cannot grow an unbounded persisted record. */ +export const MAX_AGENT_SESSION_PROVIDER_HANDLE_LINKS = 256 + +const MAX_HANDLE_FIELD_LENGTH = 512 +const LINK_ID_PATTERN = /^[A-Za-z0-9_-]{1,128}$/ + +function isHandleField(value: unknown): value is string { + return ( + typeof value === 'string' && + value.length > 0 && + value.length <= MAX_HANDLE_FIELD_LENGTH && + value === value.trim() + ) +} + +export function isAgentSessionProviderHandle(value: unknown): value is AgentSessionProviderHandle { + if (typeof value !== 'object' || value === null) { + return false + } + const handle = value as Partial & Record + if (handle.provider === 'claude') { + return ( + isHandleField(handle.sessionId) && + (handle.leafUuid === null || isHandleField(handle.leafUuid)) + ) + } + return handle.provider === 'codex' && isHandleField(handle.threadId) +} + +/** Stable string identity for one handle. Two handles with the same key name the same writer target. */ +export function agentSessionProviderHandleKey(handle: AgentSessionProviderHandle): string { + return handle.provider === 'claude' + ? `claude:${JSON.stringify([handle.sessionId, handle.leafUuid])}` + : `codex:${JSON.stringify(handle.threadId)}` +} + +/** + * Identity root: the part that a resume must preserve. A resume that changes the root is a fork, + * whatever the provider called it. + */ +export function agentSessionProviderHandleRoot(handle: AgentSessionProviderHandle): string { + return handle.provider === 'claude' + ? `claude:${JSON.stringify(handle.sessionId)}` + : `codex:${JSON.stringify(handle.threadId)}` +} + +export function agentSessionProviderHandlesEqual( + left: AgentSessionProviderHandle, + right: AgentSessionProviderHandle +): boolean { + return agentSessionProviderHandleKey(left) === agentSessionProviderHandleKey(right) +} + +export function agentSessionProviderHandleChainHead( + chain: AgentSessionProviderHandleChain +): AgentSessionProviderHandleLink | null { + return chain.at(-1) ?? null +} + +export function findAgentSessionProviderHandleLink( + chain: AgentSessionProviderHandleChain, + linkId: string +): AgentSessionProviderHandleLink | null { + return chain.find((link) => link.linkId === linkId) ?? null +} + +export function isAgentSessionProviderHandleLink( + value: unknown +): value is AgentSessionProviderHandleLink { + if (typeof value !== 'object' || value === null) { + return false + } + const link = value as Partial + const originValid = + link.origin === 'created' || + link.origin === 'adopted' || + link.origin === 'resumed' || + link.origin === 'forked' + return ( + typeof link.linkId === 'string' && + LINK_ID_PATTERN.test(link.linkId) && + isAgentSessionProviderHandle(link.handle) && + originValid && + Number.isSafeInteger(link.mintedAtFence) && + (link.mintedAtFence as number) >= 0 && + Number.isSafeInteger(link.observedAt) && + (link.origin === 'forked' + ? isHandleField(link.forkedFromKey) + : link.forkedFromKey === undefined) + ) +} + +export function isAgentSessionProviderHandleChain( + value: unknown +): value is AgentSessionProviderHandleLink[] { + if (!Array.isArray(value) || value.length > MAX_AGENT_SESSION_PROVIDER_HANDLE_LINKS) { + return false + } + let validated: AgentSessionProviderHandleLink[] = [] + try { + for (const link of value) { + if (!isAgentSessionProviderHandleLink(link)) { + return false + } + const next = appendAgentSessionProviderHandleLink(validated, link) + // A persisted chain must name every link exactly once; retry elision belongs at append time. + if (next.length !== validated.length + 1) { + return false + } + validated = next + } + return true + } catch { + return false + } +} + +/** + * Append one link, rejecting anything that would let a fork masquerade as a resume or let a + * late writer rewrite the chain under an older fence. + */ +export function appendAgentSessionProviderHandleLink( + chain: AgentSessionProviderHandleChain, + link: AgentSessionProviderHandleLink +): AgentSessionProviderHandleLink[] { + if (!isAgentSessionProviderHandleLink(link)) { + throw new Error('agent_session_provider_handle_invalid') + } + const head = agentSessionProviderHandleChainHead(chain) + if (!head) { + if (link.origin !== 'created' && link.origin !== 'adopted') { + throw new Error('agent_session_provider_handle_invalid') + } + return [link] + } + if (link.handle.provider !== head.handle.provider) { + throw new Error('agent_session_provider_handle_provider_mismatch') + } + if (link.mintedAtFence < head.mintedAtFence) { + throw new Error('agent_session_provider_handle_stale_fence') + } + if (link.origin === 'created' || link.origin === 'adopted') { + throw new Error('agent_session_provider_handle_invalid') + } + const sameRoot = + agentSessionProviderHandleRoot(link.handle) === agentSessionProviderHandleRoot(head.handle) + if (link.origin === 'resumed' && !sameRoot) { + // Why: a resume that lands on another identity root forked; recording it as a resume would + // make Orca claim continuity the provider never gave. + throw new Error('agent_session_provider_handle_forked') + } + if (link.origin === 'forked') { + if (sameRoot) { + throw new Error('agent_session_provider_handle_invalid') + } + if (link.forkedFromKey !== agentSessionProviderHandleKey(head.handle)) { + throw new Error('agent_session_provider_handle_invalid') + } + } + if ( + link.origin === 'resumed' && + agentSessionProviderHandlesEqual(link.handle, head.handle) && + link.mintedAtFence === head.mintedAtFence + ) { + // Why: re-proving the same handle at the same fence is a retry, not a new identity. + return [...chain] + } + if (findAgentSessionProviderHandleLink(chain, link.linkId)) { + // Why: the lease names its exact proof by link id; reuse would make that reference ambiguous. + throw new Error('agent_session_provider_handle_invalid') + } + if (chain.length >= MAX_AGENT_SESSION_PROVIDER_HANDLE_LINKS) { + // Why: dropping older links would erase fork provenance, so refuse and let the caller roll + // the journal epoch instead of silently losing where this conversation came from. + throw new Error('agent_session_provider_handle_chain_overflow') + } + return [...chain, link] +} diff --git a/src/shared/agent-session-record.ts b/src/shared/agent-session-record.ts new file mode 100644 index 00000000000..761cde5ecb1 --- /dev/null +++ b/src/shared/agent-session-record.ts @@ -0,0 +1,283 @@ +/** + * Durable agent-session record and its single-writer lease. + * + * The record is the session's identity — where it runs, which provider it talks to, which account + * home is pinned to it — and is independent of any terminal tab. The lease is the separate + * question of which process is currently allowed to write to it. + */ + +import type { ExecutionHostId } from './execution-host' +import { + isAgentSessionProviderHandleChain, + type AgentSessionHandleProvider, + type AgentSessionProviderHandleLink +} from './agent-session-provider-handle' + +export const AGENT_SESSION_RECORD_SCHEMA_VERSION = 1 as const + +export type AgentSessionWorkspaceKind = 'git-worktree' | 'folder' + +/** + * Where the provider process actually runs. WSL is called out separately from the execution host + * id because a WSL workspace is served by the local host but is a distinct filesystem, account + * root, and process namespace — two sessions there must never collide with their native twins. + */ +export type AgentSessionExecutionLocation = { + executionHostId: ExecutionHostId + /** Distro name when the provider runs inside WSL; null for native and remote hosts. */ + wslDistro: string | null + workspaceId: string + workspaceKind: AgentSessionWorkspaceKind +} + +/** Account root pinned at launch by the account selector, so a resume cannot drift to another login. */ +export type AgentSessionAccountHome = { + variable: 'CLAUDE_CONFIG_DIR' | 'CODEX_HOME' + /** Host-resolved absolute path in the execution host's own path syntax. */ + path: string +} + +export type AgentSessionOwnerRuntimeKind = 'native' | 'tui' + +export type AgentSessionHandoffStage = + | 'preparing' + | 'old-owner-stopped' + | 'new-owner-proving' + | 'recovering' + | 'manual-recovery' + +/** + * PID-reuse-safe process identity. `spawnToken` is the only element available on every platform: + * process start time costs a CIM query on Windows and is absent in some containers, so a runtime + * that cannot echo the token back qualifies for manual recovery only. + */ +export type AgentSessionProcessIdentity = { + hostId: string + pid: number + processStartTimeMs: number | null + spawnToken: string +} + +export type AgentSessionJournalCheckpoint = { epoch: number; sequence: number } + +/** + * Mirrors the in-memory claim registry's reserved / live / conflicted states so a conflict + * survives a restart. `released` has no registry equivalent: the registry expresses "no owner" by + * deleting the entry, and a durable record that outlives its owner needs a name for that. + */ +export type AgentSessionClaimStatus = 'reserved' | 'live' | 'conflicted' | 'released' + +export type AgentSessionDeathEvidence = { + kind: 'exit-observed' | 'pid-absent' | 'identity-mismatch' + detail: string + observedAt: number +} + +export type AgentSessionLease = { + sessionId: string + runtimeKind: AgentSessionOwnerRuntimeKind + /** Durable monotonic integer; only acquisition CAS and proven eviction move it. */ + runtimeFence: number + handoffStage: AgentSessionHandoffStage | null + /** Link id of the provider handle this owner proved; the full chain lives on the record. */ + provenHandleLinkId: string | null + /** Null between the durable reservation and the observed spawn. */ + ownerProcess: AgentSessionProcessIdentity | null + /** Reserved before any process exists, then matched against the child's environment. */ + reservedSpawnToken: string | null + leaseDeadlineAt: number + lastRenewedAt: number + handoffOperationId: string | null + journalCheckpoint: AgentSessionJournalCheckpoint | null + /** Key id that minted the HMAC claim this lease was granted under. */ + claimKeyId: string + claimStatus: AgentSessionClaimStatus + /** True from load until the host adjudicates it; no writer is granted while set. */ + unreconciled: boolean + deathEvidence: AgentSessionDeathEvidence | null +} + +export type AgentSessionRecord = { + schemaVersion: typeof AGENT_SESSION_RECORD_SCHEMA_VERSION + sessionId: string + location: AgentSessionExecutionLocation + provider: AgentSessionHandleProvider + providerHandleChain: AgentSessionProviderHandleLink[] + accountHome: AgentSessionAccountHome + lease: AgentSessionLease + createdAt: number + updatedAt: number +} + +const MAX_ID_LENGTH = 512 +const MAX_PATH_LENGTH = 4096 +const SESSION_ID_PATTERN = /^[A-Za-z0-9_-]{8,128}$/ + +function isBoundedString(value: unknown, max: number): value is string { + return typeof value === 'string' && value.length > 0 && value.length <= max +} + +export function isAgentSessionId(value: unknown): value is string { + return typeof value === 'string' && SESSION_ID_PATTERN.test(value) +} + +/** NUL cannot occur in a host id, distro name, or workspace id, so no component can forge a join. */ +const SCOPE_KEY_SEPARATOR = '\u0000' + +/** + * Scope key for host-and-workspace isolation. Native, WSL, and SSH copies of one workspace id are + * different sessions; collapsing them would let one host adjudicate another host's lease. + */ +export function agentSessionScopeKey(location: AgentSessionExecutionLocation): string { + return [location.executionHostId, location.wslDistro ?? '', location.workspaceId].join( + SCOPE_KEY_SEPARATOR + ) +} + +export function agentSessionExecutionLocationsEqual( + left: AgentSessionExecutionLocation, + right: AgentSessionExecutionLocation +): boolean { + return ( + agentSessionScopeKey(left) === agentSessionScopeKey(right) && + left.workspaceKind === right.workspaceKind + ) +} + +export function isAgentSessionExecutionLocation( + value: unknown +): value is AgentSessionExecutionLocation { + if (typeof value !== 'object' || value === null) { + return false + } + const location = value as Partial + return ( + isBoundedString(location.executionHostId, MAX_ID_LENGTH) && + (location.wslDistro === null || isBoundedString(location.wslDistro, MAX_ID_LENGTH)) && + isBoundedString(location.workspaceId, MAX_ID_LENGTH) && + (location.workspaceKind === 'git-worktree' || location.workspaceKind === 'folder') + ) +} + +export function isAgentSessionProcessIdentity( + value: unknown +): value is AgentSessionProcessIdentity { + if (typeof value !== 'object' || value === null) { + return false + } + const identity = value as Partial + return ( + isBoundedString(identity.hostId, MAX_ID_LENGTH) && + Number.isSafeInteger(identity.pid) && + (identity.pid as number) > 0 && + (identity.processStartTimeMs === null || + (Number.isSafeInteger(identity.processStartTimeMs) && + (identity.processStartTimeMs as number) >= 0)) && + isBoundedString(identity.spawnToken, MAX_ID_LENGTH) + ) +} + +function isAgentSessionAccountHome(value: unknown): value is AgentSessionAccountHome { + if (typeof value !== 'object' || value === null) { + return false + } + const home = value as Partial + return ( + (home.variable === 'CLAUDE_CONFIG_DIR' || home.variable === 'CODEX_HOME') && + isBoundedString(home.path, MAX_PATH_LENGTH) + ) +} + +function isAgentSessionJournalCheckpoint(value: unknown): value is AgentSessionJournalCheckpoint { + if (typeof value !== 'object' || value === null) { + return false + } + const checkpoint = value as Partial + return ( + Number.isSafeInteger(checkpoint.epoch) && + (checkpoint.epoch as number) >= 0 && + Number.isSafeInteger(checkpoint.sequence) && + (checkpoint.sequence as number) >= 0 + ) +} + +function isAgentSessionDeathEvidence(value: unknown): value is AgentSessionDeathEvidence { + if (typeof value !== 'object' || value === null) { + return false + } + const evidence = value as Partial + return ( + (evidence.kind === 'exit-observed' || + evidence.kind === 'pid-absent' || + evidence.kind === 'identity-mismatch') && + isBoundedString(evidence.detail, MAX_ID_LENGTH) && + Number.isSafeInteger(evidence.observedAt) && + (evidence.observedAt as number) >= 0 + ) +} + +function isAgentSessionLease(value: unknown): value is AgentSessionLease { + if (typeof value !== 'object' || value === null) { + return false + } + const lease = value as Partial + return ( + isAgentSessionId(lease.sessionId) && + (lease.runtimeKind === 'native' || lease.runtimeKind === 'tui') && + Number.isSafeInteger(lease.runtimeFence) && + (lease.runtimeFence as number) >= 0 && + (lease.handoffStage === null || + lease.handoffStage === 'preparing' || + lease.handoffStage === 'old-owner-stopped' || + lease.handoffStage === 'new-owner-proving' || + lease.handoffStage === 'recovering' || + lease.handoffStage === 'manual-recovery') && + (lease.provenHandleLinkId === null || isBoundedString(lease.provenHandleLinkId, 128)) && + (lease.ownerProcess === null || isAgentSessionProcessIdentity(lease.ownerProcess)) && + (lease.reservedSpawnToken === null || + isBoundedString(lease.reservedSpawnToken, MAX_ID_LENGTH)) && + Number.isSafeInteger(lease.leaseDeadlineAt) && + Number.isSafeInteger(lease.lastRenewedAt) && + (lease.handoffOperationId === null || + isBoundedString(lease.handoffOperationId, MAX_ID_LENGTH)) && + (lease.journalCheckpoint === null || + isAgentSessionJournalCheckpoint(lease.journalCheckpoint)) && + isBoundedString(lease.claimKeyId, MAX_ID_LENGTH) && + (lease.claimStatus === 'reserved' || + lease.claimStatus === 'live' || + lease.claimStatus === 'conflicted' || + lease.claimStatus === 'released') && + typeof lease.unreconciled === 'boolean' && + (lease.deathEvidence === null || isAgentSessionDeathEvidence(lease.deathEvidence)) + ) +} + +export function isAgentSessionRecord(value: unknown): value is AgentSessionRecord { + if (typeof value !== 'object' || value === null) { + return false + } + const record = value as Partial + const shapeValid = + record.schemaVersion === AGENT_SESSION_RECORD_SCHEMA_VERSION && + isAgentSessionId(record.sessionId) && + isAgentSessionExecutionLocation(record.location) && + (record.provider === 'claude' || record.provider === 'codex') && + isAgentSessionProviderHandleChain(record.providerHandleChain) && + isAgentSessionAccountHome(record.accountHome) && + isAgentSessionLease(record.lease) && + record.lease.sessionId === record.sessionId && + Number.isSafeInteger(record.createdAt) && + Number.isSafeInteger(record.updatedAt) + if (!shapeValid) { + return false + } + const validated = record as AgentSessionRecord + const head = validated.providerHandleChain.at(-1) + return ( + validated.providerHandleChain.every((link) => link.handle.provider === validated.provider) && + (validated.lease.claimStatus !== 'live' || + (validated.lease.ownerProcess !== null && + head?.linkId === validated.lease.provenHandleLinkId && + head.mintedAtFence === validated.lease.runtimeFence)) + ) +}