diff --git a/src/main/codex/codex-structured-item-translation.ts b/src/main/codex/codex-structured-item-translation.ts index d1c468db82a..887e593342f 100644 --- a/src/main/codex/codex-structured-item-translation.ts +++ b/src/main/codex/codex-structured-item-translation.ts @@ -170,16 +170,26 @@ function commandState(item: CodexThreadItem): 'running' | 'completed' | 'failed' return typeof exitCode === 'number' && exitCode !== 0 ? 'failed' : 'completed' } -function commandBody(item: CodexThreadItem): AgentJournalItemBody { +export type CodexJournalItem = { + body: AgentJournalItemBody | null + blobs: { digest: string; payload: string }[] +} + +function commandItem(item: CodexThreadItem): CodexJournalItem { const output = readString(item, 'aggregatedOutput') + const bounded = output === null ? null : boundInlineText(output, DEFAULT_JOURNAL_PAYLOAD_LIMITS) return { - kind: 'tool-call', - name: 'shell', - input: { command: item.command ?? null, cwd: item.cwd ?? null }, - state: commandState(item), - ...(output === null - ? {} - : { output: boundInlineText(output, DEFAULT_JOURNAL_PAYLOAD_LIMITS).bounded }) + body: { + kind: 'tool-call', + name: 'shell', + input: { command: item.command ?? null, cwd: item.cwd ?? null }, + state: commandState(item), + ...(bounded === null ? {} : { output: bounded.bounded }) + }, + blobs: + output !== null && bounded?.bounded.truncated + ? [{ digest: bounded.bounded.digest, payload: output }] + : [] } } @@ -190,22 +200,29 @@ function commandBody(item: CodexThreadItem): AgentJournalItemBody { * row the client cannot render is worse than a gap, and the ordinal projection * already guarantees skipping it costs the following messages nothing. */ -export function codexItemBody(item: CodexThreadItem): AgentJournalItemBody | null { +export function codexJournalItem(item: CodexThreadItem): CodexJournalItem { if (item.type === 'userMessage' || item.type === 'agentMessage') { const blocks = codexMessageBlocks(item) - return blocks.length === 0 - ? null - : { kind: 'message', role: item.type === 'userMessage' ? 'user' : 'assistant', blocks } + return { + body: + blocks.length === 0 + ? null + : { kind: 'message', role: item.type === 'userMessage' ? 'user' : 'assistant', blocks }, + blobs: [] + } } if (item.type === 'commandExecution') { - return commandBody(item) + return commandItem(item) } if (item.type === 'fileChange') { return { - kind: 'tool-call', - name: 'apply_patch', - input: { changes: item.changes ?? null }, - state: commandState(item) + body: { + kind: 'tool-call', + name: 'apply_patch', + input: { changes: item.changes ?? null }, + state: commandState(item) + }, + blobs: [] } } if (item.type === 'reasoning') { @@ -213,11 +230,19 @@ export function codexItemBody(item: CodexThreadItem): AgentJournalItemBody | nul readTextContent(item, 'text') ?? readTextContent(item, 'summary') ?? readTextContent(item, 'content') - return text === null - ? null - : { kind: 'status', text: boundInlineText(text, DEFAULT_JOURNAL_PAYLOAD_LIMITS).text } + return { + body: + text === null + ? null + : { kind: 'status', text: boundInlineText(text, DEFAULT_JOURNAL_PAYLOAD_LIMITS).text }, + blobs: [] + } } - return null + return { body: null, blobs: [] } +} + +export function codexItemBody(item: CodexThreadItem): AgentJournalItemBody | null { + return codexJournalItem(item).body } /** Snapshot body for text still streaming, before its item completes. */ diff --git a/src/main/codex/codex-structured-journal-translation.ts b/src/main/codex/codex-structured-journal-translation.ts index 54957121da8..a5450a90fd0 100644 --- a/src/main/codex/codex-structured-journal-translation.ts +++ b/src/main/codex/codex-structured-journal-translation.ts @@ -7,8 +7,8 @@ import { import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' import type { CodexStructuredSessionEvent } from './codex-structured-session-adapter' import { - codexItemBody, codexItemIdentity, + codexJournalItem, codexStreamingMessageBody, CodexTurnOrdinals, readCodexThreadItem @@ -157,7 +157,7 @@ export function createCodexJournalTranslator( } const turnId = readCodexTurnId(event.params) ?? currentTurnIds.get(event.threadId) ?? null const identity = identityFor(event.threadId, turnId, item) - const body = codexItemBody(item) + const translated = codexJournalItem(item) const command = readString(item, 'command') if (command) { details.set(itemKey(event.threadId, item.id), command) @@ -168,10 +168,10 @@ export function createCodexJournalTranslator( latestStreamText.delete(itemKey(event.threadId, item.id)) checkpointLengths.delete(itemKey(event.threadId, item.id)) } - if (!body) { + if (!translated.body) { return } - deps.sink.appendItem(identity, body) + deps.sink.appendItem(identity, translated.body, translated.blobs) deps.sink.publish() } diff --git a/src/main/codex/codex-structured-session-adapter.test.ts b/src/main/codex/codex-structured-session-adapter.test.ts index 177f69db260..8c4b30116e9 100644 --- a/src/main/codex/codex-structured-session-adapter.test.ts +++ b/src/main/codex/codex-structured-session-adapter.test.ts @@ -273,6 +273,26 @@ describe('CodexStructuredSessionAdapter.acquire', () => { ).rejects.toThrow('no live codex app-server') }) + it('classifies launch validation failure as pre-spawn without opening a child', async () => { + const codex = fakeCodex() + const adapter = new CodexStructuredSessionAdapter({ + resolveLaunch: async () => { + throw new Error('workspace no longer exists') + }, + openConnection: codex.openConnection + }) + + const error = await adapter + .acquire({ identity: identityFor('session-1'), fence: 7, spawnToken: 'spawn-9' }) + .catch((cause: unknown) => cause) + + expect(error).toMatchObject({ + name: 'AgentSessionPreSpawnError', + message: 'workspace no longer exists' + }) + expect(codex.connections).toHaveLength(0) + }) + it('reports the rollout path Codex named, and null when it named none', async () => { const withPath = fakeCodex() const adapter = await acquired(withPath) diff --git a/src/main/codex/codex-structured-session-adapter.ts b/src/main/codex/codex-structured-session-adapter.ts index 6a2d207803b..27894390ee4 100644 --- a/src/main/codex/codex-structured-session-adapter.ts +++ b/src/main/codex/codex-structured-session-adapter.ts @@ -2,10 +2,11 @@ import type { AgentJournalMessageItem, AgentSessionJournalIdentity } from '../../shared/agent-session-journal-types' -import type { - AgentSessionAcquisition, - AgentSessionDispatchOutcome, - StructuredAgentSessionAdapter +import { + AgentSessionPreSpawnError, + type AgentSessionAcquisition, + type AgentSessionDispatchOutcome, + type StructuredAgentSessionAdapter } from '../native-chat/agent-session-wire/structured-agent-session-adapter' import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' import { createCodexJournalTranslator } from './codex-structured-journal-translation' @@ -79,7 +80,11 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap this.acquisitions.assertCurrent(sessionId, attempt) await closeCodexPublishedSession(this.sessions, sessionId, this.deps.onEvent) this.acquisitions.assertCurrent(sessionId, attempt) - const launch = await this.deps.resolveLaunch({ identity: input.identity }) + const launch = await this.deps + .resolveLaunch({ identity: input.identity }) + .catch((error: unknown) => { + throw new AgentSessionPreSpawnError(error) + }) this.acquisitions.assertCurrent(sessionId, attempt) const connection = await open( { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts index eea191011c5..002a5f47292 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts @@ -26,6 +26,18 @@ export type AgentSessionAcquisition = { link: AgentSessionProviderHandleLink } +/** Acquisition validation failed before the adapter attempted to spawn. */ +export class AgentSessionPreSpawnError extends Error { + constructor(cause: unknown) { + super(cause instanceof Error ? cause.message : String(cause), { cause }) + this.name = 'AgentSessionPreSpawnError' + } +} + +export function isAgentSessionPreSpawnError(error: unknown): error is AgentSessionPreSpawnError { + return error instanceof Error && error.name === 'AgentSessionPreSpawnError' +} + export type AgentSessionDispatchOutcome = /** The provider owns the turn now, under this identity. */ | { state: 'accepted'; providerIdentity: AgentJournalItemIdentity } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-attach-flow.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-flow.ts index b75990348a8..49ce093c199 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-attach-flow.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-flow.ts @@ -24,6 +24,7 @@ import { } from './structured-agent-session-attach' import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter' +import { isAgentSessionPreSpawnError } from './structured-agent-session-adapter' import type { StructuredAgentSessionEventSink } from './structured-agent-session-event-sink' export type AttachFlowInput = { @@ -58,6 +59,7 @@ export async function performAttach( } let record: AgentSessionRecord + let reservedRecord: AgentSessionRecord | null = null let replayed = false try { const reserved = await store.reserveOwner( @@ -71,11 +73,23 @@ export async function performAttach( }) ) record = reserved.record + reservedRecord = record replayed = reserved.disposition === 'replayed' if (!agentSessionLeaseAdmitsWriter(record.lease)) { record = await acquireOwner(input, record) } } catch (error) { + if (reservedRecord && isAgentSessionPreSpawnError(error)) { + const spawnToken = reservedRecord.lease.reservedSpawnToken + if (spawnToken) { + await store.markReservationProcessless({ + sessionId, + fence: reservedRecord.lease.runtimeFence, + spawnToken, + now: input.now() + }) + } + } return { ok: false, refusal: classifyStoreFailure(error, store.getRecord(sessionId)?.lease.runtimeFence ?? null) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.ts index 1bfe625c58d..c717dc22660 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.ts @@ -16,12 +16,19 @@ import type { AgentJournalItemIdentity } from '../../../shared/agent-session-journal-types' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' +import { putJournalBlob } from '../agent-session-journal/journal-blob-store' + +export type StructuredAgentSessionJournalBlob = { digest: string; payload: string } /** The only journal surface an adapter gets: append and publish, no reads. An * adapter that could read the journal would start reconciling against it, and * reconciliation is the wire's job, not the provider's. */ export type StructuredAgentSessionEventSink = { - appendItem(identity: AgentJournalItemIdentity, body: AgentJournalItemBody): void + appendItem( + identity: AgentJournalItemIdentity, + body: AgentJournalItemBody, + blobs?: readonly StructuredAgentSessionJournalBlob[] + ): void appendTombstone(identity: AgentJournalItemIdentity): void /** Fan the journal out to subscribers. Cheap and idempotent. */ publish(): void @@ -88,8 +95,13 @@ export function createDeferredStructuredAgentSessionEventSink( return { sink: { - appendItem: (identity, body: AgentJournalItemBody) => { - submit((bound) => bound.journal.appendItem(identity, body, { fence: bound.fence })) + appendItem: (identity, body: AgentJournalItemBody, blobs = []) => { + submit(async (bound) => { + for (const blob of blobs) { + await putJournalBlob(bound.journal.directory, blob.digest, blob.payload) + } + await bound.journal.appendItem(identity, body, { fence: bound.fence }) + }) }, appendTombstone: (identity) => { submit((bound) => bound.journal.appendTombstone(identity, { fence: bound.fence })) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts index b3cd6d78f93..6123bbf95b7 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts @@ -269,6 +269,7 @@ export class StructuredAgentSessionHost { envelope: AgentSessionMutationEnvelope body: AgentJournalMessageItem retryUnknown?: true + beforeRun?: () => void } ): Promise> { return this.mutate(caller, params.envelope, sendPlan(params)) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-admission.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-admission.ts index b43fa331270..04a0d6aaf48 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-admission.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-admission.ts @@ -107,6 +107,7 @@ export async function admitAndRunAgentSessionMutation( } } + plan.beforeRun?.() const outcome = await runSettledAgentSessionMutation({ store: request.store, callerKey: request.callerKey, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-plans.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-plans.ts index b87bba19a0d..b15c3a0e03a 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-plans.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-plans.ts @@ -27,6 +27,7 @@ import { export type MutationPlan = { method: string fields: Record + beforeRun?: () => void run: (ctx: AgentSessionTurnContext) => Promise> replay: (ctx: AgentSessionTurnContext, outcome: AgentSessionOperationOutcome) => TValue | null rerunWhenReplayMissing?: (ctx: AgentSessionTurnContext) => boolean @@ -36,6 +37,7 @@ export function sendPlan(params: { envelope: AgentSessionMutationEnvelope body: AgentJournalMessageItem retryUnknown?: true + beforeRun?: () => void }): MutationPlan { // The operation id IS the client message id: one send, one durable row, one // key the client reconciles its optimistic bubble against. @@ -44,6 +46,7 @@ export function sendPlan(params: { method: 'agentSession.send', // A control signal is not payload; only the matching durable unknown unlocks redispatch. fields: { body: params.body }, + ...(params.beforeRun ? { beforeRun: params.beforeRun } : {}), rerunWhenReplayMissing: (ctx) => params.retryUnknown === true && ctx.journal diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-processless-reservation.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-processless-reservation.test.ts new file mode 100644 index 00000000000..d33980d9ff5 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-processless-reservation.test.ts @@ -0,0 +1,106 @@ +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, describe, expect, it, vi } from 'vitest' +import { computeAgentSessionPayloadFingerprint } from '../../../shared/agent-session-mutation-envelope' +import { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' +import { + AgentSessionPreSpawnError, + type StructuredAgentSessionAdapter +} from './structured-agent-session-adapter' +import { + attachFingerprintFields, + type AgentSessionAttachParams +} from './structured-agent-session-attach' +import { performAttach } from './structured-agent-session-attach-flow' + +const NOW = 1_800_000_000_000 +const SESSION = 'session-alpha' +const OPERATION = `${NOW}-${'1'.padStart(32, '0')}` +let root: string | null = null + +afterEach(async () => { + if (root) { + await rm(root, { recursive: true, force: true }) + } + root = null +}) + +function attachParams(): AgentSessionAttachParams { + const params: AgentSessionAttachParams = { + envelope: { + sessionId: SESSION, + clientOperationId: OPERATION, + expectedRuntimeFence: null, + payloadFingerprint: '' + }, + location: { + executionHostId: 'local', + wslDistro: null, + workspaceId: 'workspace-1', + workspaceKind: 'folder' + }, + provider: 'codex', + agent: 'codex', + accountHome: { variable: 'CODEX_HOME', path: '/home/dev/.codex' }, + runtimeKind: 'native', + providerHandle: { kind: 'codex', threadId: 'thread-1' } + } + return { + ...params, + envelope: { + ...params.envelope, + payloadFingerprint: computeAgentSessionPayloadFingerprint({ + method: 'agentSession.attach', + sessionId: SESSION, + fields: attachFingerprintFields(params) + }) + } + } +} + +describe('processless structured session reservation', () => { + it('persists pre-spawn proof so restart can release only the proven reservation', async () => { + root = await mkdtemp(join(tmpdir(), 'orca-processless-reservation-')) + const storeDir = join(root, 'store') + const store = await AgentSessionRecordStore.open({ directory: storeDir, hostId: 'local' }) + const adapter = { + acquire: vi.fn(async () => { + throw new AgentSessionPreSpawnError(new Error('workspace no longer exists')) + }) + } as unknown as StructuredAgentSessionAdapter + + await expect( + performAttach({ + store, + adapter, + journalRoot: root, + authority: { + spawnToken: 'spawn-a', + claimKeyId: 'key-1', + handoffOperationId: OPERATION, + probe: { outcome: 'reservation-unused' } + }, + callerKey: 'client-1', + params: attachParams(), + now: () => NOW, + onAttached: () => {} + }) + ).rejects.toThrow('workspace no longer exists') + expect(store.getRecord(SESSION)?.lease.processlessAt).toBe(NOW) + + const reopened = await AgentSessionRecordStore.open({ directory: storeDir, hostId: 'local' }) + await reopened.reconcileOnRestart({ + probe: async (record) => + record.lease.processlessAt === NOW + ? { outcome: 'reservation-unused' } + : { outcome: 'indeterminate', reason: 'missing pre-spawn proof' }, + now: NOW + 1 + }) + expect(reopened.getRecord(SESSION)?.lease).toMatchObject({ + claimStatus: 'released', + runtimeFence: 2, + reservedSpawnToken: null + }) + }) +}) diff --git a/src/main/runtime/agent-session-lease-transitions.ts b/src/main/runtime/agent-session-lease-transitions.ts index 0541039942a..01ec81dbc25 100644 --- a/src/main/runtime/agent-session-lease-transitions.ts +++ b/src/main/runtime/agent-session-lease-transitions.ts @@ -88,6 +88,7 @@ export function reserveAgentSessionOwner(args: { provenHandleLinkId: null, ownerProcess: null, reservedSpawnToken: reservation.spawnToken, + processlessAt: null, leaseDeadlineAt: reservation.now + reservation.leaseTtlMs, lastRenewedAt: reservation.now, handoffOperationId: reservation.handoffOperationId, @@ -99,12 +100,16 @@ export function reserveAgentSessionOwner(args: { } /** Step 4 of acquisition: write the observed identity back into the same lease row. */ -export function commitAgentSessionProcessIdentity(args: { - record: AgentSessionRecord +export type AgentSessionProcessIdentityCommit = { + sessionId: string fence: number process: AgentSessionProcessIdentity now: number -}): AgentSessionRecord { +} + +export function commitAgentSessionProcessIdentity( + args: AgentSessionProcessIdentityCommit & { record: AgentSessionRecord } +): AgentSessionRecord { const { record } = args assertFence(record.lease, args.fence) if (record.lease.claimStatus !== 'reserved' || record.lease.ownerProcess !== null) { @@ -114,7 +119,12 @@ export function commitAgentSessionProcessIdentity(args: { // 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 }) + return withLease(record, { + ...record.lease, + ownerProcess: args.process, + processlessAt: null, + lastRenewedAt: args.now + }) } /** @@ -217,6 +227,7 @@ export function evictAgentSessionOwner(args: { handoffStage: null, ownerProcess: null, reservedSpawnToken: null, + processlessAt: null, claimStatus: 'released', lastRenewedAt: args.now, handoffOperationId: null, @@ -247,6 +258,7 @@ export function applyAgentSessionRestartAdjudication(args: { handoffStage: null, ownerProcess: null, reservedSpawnToken: null, + processlessAt: null, claimStatus: 'released', unreconciled: false, lastRenewedAt: args.now, diff --git a/src/main/runtime/agent-session-processless-reservation.ts b/src/main/runtime/agent-session-processless-reservation.ts new file mode 100644 index 00000000000..52ab279e01d --- /dev/null +++ b/src/main/runtime/agent-session-processless-reservation.ts @@ -0,0 +1,29 @@ +import type { AgentSessionRecord } from '../../shared/agent-session-record' + +export type AgentSessionReservationProcesslessProof = { + sessionId: string + fence: number + spawnToken: string + now: number +} + +export function markAgentSessionReservationProcessless( + args: AgentSessionReservationProcesslessProof & { record: AgentSessionRecord } +): AgentSessionRecord { + const { record } = args + if (record.lease.runtimeFence !== args.fence || record.lease.unreconciled) { + throw new Error('agent_session_checkpoint_stale') + } + if ( + record.lease.claimStatus !== 'reserved' || + record.lease.ownerProcess !== null || + record.lease.reservedSpawnToken !== args.spawnToken + ) { + throw new Error('agent_session_ownership_unknown') + } + return { + ...record, + lease: { ...record.lease, processlessAt: args.now }, + updatedAt: args.now + } +} diff --git a/src/main/runtime/agent-session-record-store.ts b/src/main/runtime/agent-session-record-store.ts index 15fcef3c85e..f148d0064c9 100644 --- a/src/main/runtime/agent-session-record-store.ts +++ b/src/main/runtime/agent-session-record-store.ts @@ -28,7 +28,6 @@ import { type AgentSessionExecutionLocation, type AgentSessionHandoffStage, type AgentSessionJournalCheckpoint, - type AgentSessionProcessIdentity, type AgentSessionRecord } from '../../shared/agent-session-record' import { @@ -39,8 +38,13 @@ import { proveAgentSessionOwner, renewAgentSessionLease, setAgentSessionHandoffStage, - setAgentSessionJournalCheckpoint + setAgentSessionJournalCheckpoint, + type AgentSessionProcessIdentityCommit } from './agent-session-lease-transitions' +import { + markAgentSessionReservationProcessless, + type AgentSessionReservationProcesslessProof +} from './agent-session-processless-reservation' import { applyAgentSessionReservation, evaluateAgentSessionReserveOperation, @@ -59,11 +63,6 @@ import { 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. */ @@ -171,17 +170,21 @@ export class AgentSessionRecordStore { }) } - async commitProcessIdentity(args: { - sessionId: string - fence: number - process: AgentSessionProcessIdentity - now: number - }): Promise { + async commitProcessIdentity( + args: AgentSessionProcessIdentityCommit + ): Promise { return this.mutate(args.sessionId, (record) => commitAgentSessionProcessIdentity({ ...args, record }) ) } + markReservationProcessless = ( + args: AgentSessionReservationProcesslessProof + ): Promise => + this.mutate(args.sessionId, (record) => + markAgentSessionReservationProcessless({ ...args, record }) + ) + async proveOwner(args: { sessionId: string fence: number @@ -347,7 +350,5 @@ export class AgentSessionRecordStore { } /** Serialize every mutation against the latest committed disk state. */ - private async transact(apply: () => T): Promise { - return this.transactions.transact(apply) - } + private transact = (apply: () => T): Promise => this.transactions.transact(apply) } diff --git a/src/main/runtime/rpc/methods/clipboard.test.ts b/src/main/runtime/rpc/methods/clipboard.test.ts index a28daf4f1af..c0d224b85c6 100644 --- a/src/main/runtime/rpc/methods/clipboard.test.ts +++ b/src/main/runtime/rpc/methods/clipboard.test.ts @@ -105,6 +105,21 @@ describe('clipboard RPC methods', () => { expect(hasMobileClipboardImagePath('device-b', path)).toBe(false) }) + it('does not authorize a remote-host clipboard path for local structured delivery', async () => { + const path = '/tmp/orca-paste-image.png' + saveClipboardImageBufferAsTempFile.mockResolvedValue(path) + const dispatcher = makeDispatcher() + + await expect( + callMobile(dispatcher, 'clipboard.saveImageAsTempFile', { + contentBase64: Buffer.from('png-bytes').toString('base64'), + connectionId: 'ssh-1' + }) + ).resolves.toMatchObject({ ok: true, result: path }) + + expect(hasMobileClipboardImagePath('device-a', path)).toBe(false) + }) + it('rejects non-base64 clipboard image payloads', async () => { const dispatcher = makeDispatcher() @@ -174,12 +189,55 @@ describe('clipboard RPC methods', () => { ).resolves.toMatchObject({ ok: true, result: { receivedBase64Length: contentBase64.length } }) await expect( - callMobile(dispatcher, 'clipboard.commitImageUpload', { uploadId: uploadId.uploadId }) + dispatcher.dispatch( + makeRequest('clipboard.commitImageUpload', { uploadId: uploadId.uploadId }) + ) ).resolves.toMatchObject({ ok: true, result: '/tmp/orca-paste-image.png' }) expect(saveClipboardImageBufferAsTempFile).toHaveBeenCalledWith(Buffer.from('png-bytes'), { connectionId: 'ssh-1' }) + expect(hasMobileClipboardImagePath('device-a', '/tmp/orca-paste-image.png')).toBe(false) + }) + + it('binds chunk mutation and provenance to the mobile client that started the upload', async () => { + saveClipboardImageBufferAsTempFile.mockResolvedValue('/tmp/orca-paste-image.png') + const dispatcher = makeDispatcher() + const contentBase64 = Buffer.from('png-bytes').toString('base64') + const start = await callMobile(dispatcher, 'clipboard.startImageUpload', { + expectedBase64Length: contentBase64.length, + connectionId: null + }) + const uploadId = (start.ok ? start.result : null) as { uploadId: string } + + for (const method of [ + 'clipboard.appendImageUploadChunk', + 'clipboard.commitImageUpload', + 'clipboard.abortImageUpload' + ]) { + const params = + method === 'clipboard.appendImageUploadChunk' + ? { uploadId: uploadId.uploadId, offset: 0, contentBase64 } + : { uploadId: uploadId.uploadId } + await expect(callMobile(dispatcher, method, params, 'device-b')).resolves.toMatchObject({ + ok: false + }) + } + + await expect( + callMobile(dispatcher, 'clipboard.appendImageUploadChunk', { + uploadId: uploadId.uploadId, + offset: 0, + contentBase64 + }) + ).resolves.toMatchObject({ + ok: true, + result: { receivedBase64Length: contentBase64.length } + }) + await expect( + callMobile(dispatcher, 'clipboard.commitImageUpload', { uploadId: uploadId.uploadId }) + ).resolves.toMatchObject({ ok: true, result: '/tmp/orca-paste-image.png' }) expect(hasMobileClipboardImagePath('device-a', '/tmp/orca-paste-image.png')).toBe(true) + expect(hasMobileClipboardImagePath('device-b', '/tmp/orca-paste-image.png')).toBe(false) }) it('rejects out-of-order chunk offsets', async () => { diff --git a/src/main/runtime/rpc/methods/clipboard.ts b/src/main/runtime/rpc/methods/clipboard.ts index 71c3e0cc57c..e6b487d7761 100644 --- a/src/main/runtime/rpc/methods/clipboard.ts +++ b/src/main/runtime/rpc/methods/clipboard.ts @@ -1,5 +1,5 @@ import { z } from 'zod' -import { defineMethod, type RpcMethod } from '../core' +import { defineMethod, type RpcContext, type RpcMethod } from '../core' import { saveClipboardImageBufferAsTempFile } from '../../../window/clipboard-image-temp-file' import { randomUUID } from 'node:crypto' import { @@ -17,6 +17,7 @@ const BASE64_PATTERN = /^[A-Za-z0-9+/]*={0,2}$/ type ClipboardImageUpload = { expectedBase64Length: number connectionId?: string | null + mobileClientId?: string chunks: string[] receivedBase64Length: number expiresAt: number @@ -70,6 +71,28 @@ function getUpload(uploadId: string): ClipboardImageUpload { return upload } +function mobileClientId(ctx: RpcContext): string | undefined { + if (ctx.clientKind !== 'mobile') { + return undefined + } + const clientId = ctx.clientId?.trim() + if (!clientId) { + throw new Error('Clipboard image upload requires an authenticated mobile client') + } + return clientId +} + +function assertMobileUploadOwner( + upload: ClipboardImageUpload, + ctx: RpcContext +): string | undefined { + const clientId = mobileClientId(ctx) + if (clientId && upload.mobileClientId !== clientId) { + throw new Error('Clipboard image upload was not found') + } + return clientId +} + function assertValidBase64Content(value: string): void { if (!isValidBase64(value)) { throw new Error('Clipboard image content must be base64') @@ -133,14 +156,15 @@ export const CLIPBOARD_METHODS: RpcMethod[] = [ name: 'clipboard.saveImageAsTempFile', params: SaveImageAsTempFile, handler: async (params, ctx) => { + const clientId = mobileClientId(ctx) const path = await saveClipboardImageBufferAsTempFile( Buffer.from(params.contentBase64, 'base64'), { connectionId: params.connectionId } ) - if (ctx.clientKind === 'mobile') { - recordMobileClipboardImagePath(ctx.clientId, path) + if (clientId && !params.connectionId) { + recordMobileClipboardImagePath(clientId, path) } return path } @@ -148,7 +172,7 @@ export const CLIPBOARD_METHODS: RpcMethod[] = [ defineMethod({ name: 'clipboard.startImageUpload', params: StartImageUpload, - handler: (params) => { + handler: (params, ctx) => { pruneExpiredUploads() if (clipboardImageUploads.size >= CLIPBOARD_IMAGE_UPLOAD_MAX_CONCURRENT) { throw new Error('Too many clipboard image uploads are in progress') @@ -157,6 +181,7 @@ export const CLIPBOARD_METHODS: RpcMethod[] = [ clipboardImageUploads.set(uploadId, { expectedBase64Length: params.expectedBase64Length, connectionId: params.connectionId, + mobileClientId: mobileClientId(ctx), chunks: [], receivedBase64Length: 0, expiresAt: Date.now() + CLIPBOARD_IMAGE_UPLOAD_TTL_MS, @@ -168,8 +193,9 @@ export const CLIPBOARD_METHODS: RpcMethod[] = [ defineMethod({ name: 'clipboard.appendImageUploadChunk', params: AppendImageUploadChunk, - handler: (params) => { + handler: (params, ctx) => { const upload = getUpload(params.uploadId) + assertMobileUploadOwner(upload, ctx) if (params.offset !== upload.receivedBase64Length) { throw new Error('Clipboard image chunk offset is out of order') } @@ -188,6 +214,7 @@ export const CLIPBOARD_METHODS: RpcMethod[] = [ params: CommitImageUpload, handler: async (params, ctx) => { const upload = getUpload(params.uploadId) + const clientId = assertMobileUploadOwner(upload, ctx) try { if (upload.receivedBase64Length !== upload.expectedBase64Length) { throw new Error('Clipboard image upload is incomplete') @@ -200,8 +227,8 @@ export const CLIPBOARD_METHODS: RpcMethod[] = [ connectionId: upload.connectionId } ) - if (ctx.clientKind === 'mobile') { - recordMobileClipboardImagePath(ctx.clientId, path) + if (clientId && !upload.connectionId) { + recordMobileClipboardImagePath(clientId, path) } return path } finally { @@ -214,7 +241,12 @@ export const CLIPBOARD_METHODS: RpcMethod[] = [ defineMethod({ name: 'clipboard.abortImageUpload', params: AbortImageUpload, - handler: (params) => { + handler: (params, ctx) => { + pruneExpiredUploads() + const upload = clipboardImageUploads.get(params.uploadId) + if (upload) { + assertMobileUploadOwner(upload, ctx) + } deleteUpload(params.uploadId) return { aborted: true } } diff --git a/src/main/runtime/rpc/methods/structured-agent-session.test.ts b/src/main/runtime/rpc/methods/structured-agent-session.test.ts index 29418f609f1..40b5cbb1280 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session.test.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session.test.ts @@ -86,7 +86,10 @@ function hostStub(): StructuredAgentSessionHost { unconfirmedClientMessageIds: [] } })), - send: vi.fn(async () => ({ ok: true, replayed: false })), + send: vi.fn(async (_caller, params) => { + params.beforeRun?.() + return { ok: true, replayed: false } + }), cancel: vi.fn(async () => ({ ok: true, replayed: false })), respondToPrompt: vi.fn(async () => ({ ok: true, replayed: false })), setOption: vi.fn(async () => ({ ok: true, replayed: false })), @@ -355,7 +358,7 @@ describe('mobile image provenance', () => { error: { message: expect.stringContaining('agent_session_image_untrusted') } }) } - expect(hostCalls.send).not.toHaveBeenCalled() + expect(hostCalls.send).toHaveBeenCalledTimes(3) }) it('keeps trusted in-process image paths unchanged', async () => { diff --git a/src/main/runtime/rpc/methods/structured-agent-session.ts b/src/main/runtime/rpc/methods/structured-agent-session.ts index 2daa2867236..d0100b997df 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session.ts @@ -181,8 +181,10 @@ export const STRUCTURED_AGENT_SESSION_METHODS: RpcAnyMethod[] = [ params: SendParams, handler: async (params, ctx) => { const host = requireHost(ctx) - assertMobileImageProvenance(ctx, params.body) - return host.send(callerFor(ctx), params) + return host.send(callerFor(ctx), { + ...params, + beforeRun: () => assertMobileImageProvenance(ctx, params.body) + }) } }), defineMethod({ diff --git a/src/main/runtime/structured-agent-session-integration.test.ts b/src/main/runtime/structured-agent-session-integration.test.ts index 2f71b1a1be6..0007cc0c333 100644 --- a/src/main/runtime/structured-agent-session-integration.test.ts +++ b/src/main/runtime/structured-agent-session-integration.test.ts @@ -26,6 +26,7 @@ import type { import { attachFingerprintFields } from '../native-chat/agent-session-wire/structured-agent-session-attach' import { getStructuredAgentSessionHost } from '../native-chat/agent-session-wire/structured-agent-session-registry' import { journalDirectoryFor } from '../native-chat/agent-session-journal/journal-paths' +import { readJournalBlob } from '../native-chat/agent-session-journal/journal-blob-store' import { openAgentSessionJournal, type AgentSessionJournal @@ -34,6 +35,10 @@ import type { OrcaRuntimeService } from './orca-runtime' import type { RpcRequest, RpcResponse } from './rpc/core' import { RpcDispatcher } from './rpc/dispatcher' import { STRUCTURED_AGENT_SESSION_METHODS } from './rpc/methods/structured-agent-session' +import { + recordMobileClipboardImagePath, + resetMobileClipboardImageProvenanceForTest +} from './rpc/mobile-clipboard-image-provenance' import { ensureStructuredAgentSessionHost, stopStructuredAgentSessionRuntime @@ -44,6 +49,7 @@ const THREAD = 'thread-integration' const TURN = 'turn-1' const WORKSPACE = 'workspace-1' const CLIENT = { + clientId: 'device-a', clientKind: 'mobile' as const, clientCapabilities: [STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY] } @@ -258,6 +264,7 @@ async function historyPage( } beforeEach(async () => { + resetMobileClipboardImageProvenanceForTest() operations = 0 root = await mkdtemp(join(tmpdir(), 'orca-structured-integration-')) codex = fakeCodex() @@ -307,6 +314,7 @@ beforeEach(async () => { afterEach(async () => { await stopStructuredAgentSessionRuntime() + resetMobileClipboardImageProvenanceForTest() await rm(root, { recursive: true, force: true }) }) @@ -567,6 +575,56 @@ describe('a structured codex session over agentSession.*', () => { ).toBe(false) }) + it('persists truncated command output before publishing its journal row', async () => { + await ok<{ fence: number }>('agentSession.create', createIntentParams()) + const output = 'large command output\n'.repeat(2_000) + + codex.notify('item/completed', { + threadId: THREAD, + turnId: TURN, + item: { + type: 'commandExecution', + id: 'item-large-output', + command: 'print-many-lines', + status: 'completed', + exitCode: 0, + aggregatedOutput: output + } + }) + await drainStreamedEvents() + + const host = getStructuredAgentSessionHost() + const journal = ( + host as unknown as { sessions: Map } + ).sessions.get(SESSION)!.journal + const item = journal.snapshot().items.find((candidate) => candidate.body?.kind === 'tool-call') + const bounded = item?.body?.kind === 'tool-call' ? item.body.output : undefined + expect(bounded).toMatchObject({ truncated: true, byteLength: Buffer.byteLength(output) }) + expect(await readJournalBlob(journal.directory, bounded?.digest ?? '')).toBe(output) + }) + + it('replays a durable image send after ephemeral upload provenance is gone', async () => { + const created = await ok<{ fence: number }>('agentSession.create', createIntentParams()) + const path = '/tmp/orca-paste-image.png' + recordMobileClipboardImagePath('device-a', path) + const body = { + kind: 'message' as const, + role: 'user' as const, + blocks: [{ type: 'image-ref' as const, path }] + } + const params = { + envelope: envelope('agentSession.send', { body }, created.fence), + body + } + + await ok('agentSession.send', params) + resetMobileClipboardImageProvenanceForTest() + const replay = await call('agentSession.send', params) + + expect(replay).toMatchObject({ ok: true, result: { ok: true, replayed: true } }) + expect(codex.live().calls.filter((entry) => entry.method === 'turn/start')).toHaveLength(1) + }) + it('joins an acquired attach through journal bind before draining final rows', async () => { const host = await ensureStructuredAgentSessionHost({ stateDirectory: root, diff --git a/src/main/runtime/structured-agent-session-runtime.test.ts b/src/main/runtime/structured-agent-session-runtime.test.ts index 49e2c444a04..516259a6c7c 100644 --- a/src/main/runtime/structured-agent-session-runtime.test.ts +++ b/src/main/runtime/structured-agent-session-runtime.test.ts @@ -7,8 +7,11 @@ import { createStructuredAgentSessionOwnerProbe } from './structured-agent-sessi const HOST_ID = 'local' -function record(ownerProcess: AgentSessionProcessIdentity | null): AgentSessionRecord { - return { sessionId: 'session-1', lease: { ownerProcess } } as AgentSessionRecord +function record( + ownerProcess: AgentSessionProcessIdentity | null, + processlessAt?: number | null +): AgentSessionRecord { + return { sessionId: 'session-1', lease: { ownerProcess, processlessAt } } as AgentSessionRecord } const OWNER: AgentSessionProcessIdentity = { @@ -47,4 +50,15 @@ describe('structured agent-session owner probe', () => { // would put a second writer on a live Codex thread. expect(result.outcome).toBe('indeterminate') }) + + it('releases only a reservation carrying durable pre-spawn proof', async () => { + const probe = vi.fn(async () => ({ outcome: 'pid-absent' }) as const) + const result = await createStructuredAgentSessionOwnerProbe( + HOST_ID, + probe + )(record(null, 1_800_000_000_000)) + + expect(probe).not.toHaveBeenCalled() + expect(result).toEqual({ outcome: 'reservation-unused' }) + }) }) diff --git a/src/main/runtime/structured-agent-session-runtime.ts b/src/main/runtime/structured-agent-session-runtime.ts index 66968c4a6fc..02cb7dffc00 100644 --- a/src/main/runtime/structured-agent-session-runtime.ts +++ b/src/main/runtime/structured-agent-session-runtime.ts @@ -116,6 +116,9 @@ export function createStructuredAgentSessionOwnerProbe( return async (record) => { const owner = record.lease.ownerProcess if (!owner) { + if (record.lease.processlessAt !== undefined && record.lease.processlessAt !== null) { + return { outcome: 'reservation-unused' } + } // Freeing a reservation needs positive proof that nothing spawned under // its token, and this host has no spawn-token process scan yet. return { diff --git a/src/shared/agent-session-record.ts b/src/shared/agent-session-record.ts index 761cde5ecb1..e7aa6c8cdee 100644 --- a/src/shared/agent-session-record.ts +++ b/src/shared/agent-session-record.ts @@ -85,6 +85,8 @@ export type AgentSessionLease = { ownerProcess: AgentSessionProcessIdentity | null /** Reserved before any process exists, then matched against the child's environment. */ reservedSpawnToken: string | null + /** Set only when acquisition failed before any spawn attempt. */ + processlessAt?: number | null leaseDeadlineAt: number lastRenewedAt: number handoffOperationId: string | null @@ -236,6 +238,9 @@ function isAgentSessionLease(value: unknown): value is AgentSessionLease { (lease.ownerProcess === null || isAgentSessionProcessIdentity(lease.ownerProcess)) && (lease.reservedSpawnToken === null || isBoundedString(lease.reservedSpawnToken, MAX_ID_LENGTH)) && + (lease.processlessAt === undefined || + lease.processlessAt === null || + (Number.isSafeInteger(lease.processlessAt) && (lease.processlessAt as number) >= 0)) && Number.isSafeInteger(lease.leaseDeadlineAt) && Number.isSafeInteger(lease.lastRenewedAt) && (lease.handoffOperationId === null ||