mirror of
https://github.com/stablyai/orca.git
synced 2026-10-22 00:03:47 +00:00
fix: close structured session data-loss gaps
This commit is contained in:
@@ -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. */
|
||||
|
||||
@@ -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()
|
||||
}
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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(
|
||||
{
|
||||
|
||||
@@ -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 }
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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 }))
|
||||
|
||||
@@ -269,6 +269,7 @@ export class StructuredAgentSessionHost {
|
||||
envelope: AgentSessionMutationEnvelope
|
||||
body: AgentJournalMessageItem
|
||||
retryUnknown?: true
|
||||
beforeRun?: () => void
|
||||
}
|
||||
): Promise<AgentSessionMutationResult<AgentSessionSendResult>> {
|
||||
return this.mutate(caller, params.envelope, sendPlan(params))
|
||||
|
||||
@@ -107,6 +107,7 @@ export async function admitAndRunAgentSessionMutation<TValue>(
|
||||
}
|
||||
}
|
||||
|
||||
plan.beforeRun?.()
|
||||
const outcome = await runSettledAgentSessionMutation({
|
||||
store: request.store,
|
||||
callerKey: request.callerKey,
|
||||
|
||||
@@ -27,6 +27,7 @@ import {
|
||||
export type MutationPlan<TValue> = {
|
||||
method: string
|
||||
fields: Record<string, unknown>
|
||||
beforeRun?: () => void
|
||||
run: (ctx: AgentSessionTurnContext) => Promise<TurnOutcome<TValue>>
|
||||
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<AgentSessionSendResult> {
|
||||
// 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
|
||||
|
||||
+106
@@ -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
|
||||
})
|
||||
})
|
||||
})
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
@@ -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<AgentSessionRecord> {
|
||||
async commitProcessIdentity(
|
||||
args: AgentSessionProcessIdentityCommit
|
||||
): Promise<AgentSessionRecord> {
|
||||
return this.mutate(args.sessionId, (record) =>
|
||||
commitAgentSessionProcessIdentity({ ...args, record })
|
||||
)
|
||||
}
|
||||
|
||||
markReservationProcessless = (
|
||||
args: AgentSessionReservationProcesslessProof
|
||||
): Promise<AgentSessionRecord> =>
|
||||
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<T>(apply: () => T): Promise<T> {
|
||||
return this.transactions.transact(apply)
|
||||
}
|
||||
private transact = <T>(apply: () => T): Promise<T> => this.transactions.transact(apply)
|
||||
}
|
||||
|
||||
@@ -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 () => {
|
||||
|
||||
@@ -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 }
|
||||
}
|
||||
|
||||
@@ -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 () => {
|
||||
|
||||
@@ -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({
|
||||
|
||||
@@ -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<string, { journal: AgentSessionJournal }> }
|
||||
).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,
|
||||
|
||||
@@ -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' })
|
||||
})
|
||||
})
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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 ||
|
||||
|
||||
Reference in New Issue
Block a user