diff --git a/src/main/runtime/claude-structured-session-integration.test.ts b/src/main/runtime/claude-structured-session-integration.test.ts index d464d87e8f7b..2e161f496754 100644 --- a/src/main/runtime/claude-structured-session-integration.test.ts +++ b/src/main/runtime/claude-structured-session-integration.test.ts @@ -27,6 +27,7 @@ import type { OrcaRuntimeService } from './orca-runtime' import type { RpcRequest, RpcResponse } from './rpc/core' import type { ClaudeStructuredAuthPolicy } from '../claude-accounts/claude-structured-auth-policy' import { RpcDispatcher } from './rpc/dispatcher' +import { expectStructuredWorkerAdmission } from './structured-worker-admission.test-support' import { STRUCTURED_AGENT_SESSION_METHODS } from './rpc/methods/structured-agent-session' import { ensureStructuredAgentSessionHost, @@ -532,6 +533,15 @@ describe('a structured Claude session over agentSession.*', () => { expect(leaseOf(SESSION)).toMatchObject({ claimStatus: 'released', handoffStage: null }) }) + it('admits a fresh worker preamble before the provider echo without resending', async () => { + await ok('agentSession.create', createIntentParams()) + claude.live().send = async (message) => { + claude.live().sent.push(message) + } + await expectStructuredWorkerAdmission(SESSION) + expect(claude.live().sent.filter((message) => message.type === 'user')).toHaveLength(1) + }) + it('creates, sends, streams, approves, interrupts, and resumes from the chain head', async () => { vi.stubEnv('ANTHROPIC_API_KEY', 'sk-ant-SHELL-LEAK') const created = await ok<{ fence: number }>('agentSession.create', createIntentParams()) diff --git a/src/main/runtime/orchestration/structured-mailbox-pointer-delivery.test.ts b/src/main/runtime/orchestration/structured-mailbox-pointer-delivery.test.ts index a181b89aebc0..c9f342345db1 100644 --- a/src/main/runtime/orchestration/structured-mailbox-pointer-delivery.test.ts +++ b/src/main/runtime/orchestration/structured-mailbox-pointer-delivery.test.ts @@ -74,7 +74,7 @@ function attentionJournal(): AgentJournalRenderItem[] { function harness(options: { journal: AgentJournalRenderItem[] | null - dispatchState?: 'accepted' | 'rejected' | 'unknown' + dispatchState?: 'pending' | 'accepted' | 'rejected' | 'unknown' refusal?: AgentSessionPtyWriteRefusal /** The coordinator of this worker's Run is mid-batch: it checked and has not acked yet. */ outstandingRunDelivery?: boolean @@ -269,6 +269,27 @@ describe('structured mailbox pointer delivery', () => { expect(send).not.toHaveBeenCalled() }) + it('retains admitted mail and replays its operation until provider acceptance', async () => { + const { delivery, send, markAsDelivered, stored } = harness({ + journal: idleJournal(), + dispatchState: 'pending' + }) + delivery.deliverForHandle('dispatch:d1') + await flush() + expect(markAsDelivered).not.toHaveBeenCalled() + const first = send.mock.calls[0]![0].operationId + delivery.onJournalActivity('session-1') + await flush() + expect(send.mock.calls[1]![0].operationId).toBe(first) + expect(markAsDelivered).not.toHaveBeenCalled() + send.mockResolvedValueOnce({ kind: 'sent', state: 'accepted' }) + delivery.onJournalActivity('session-1') + await flush() + expect(send.mock.calls[2]![0].operationId).toBe(first) + expect(markAsDelivered).toHaveBeenCalledExactlyOnceWith(['m1']) + expect(stored.size).toBe(0) + }) + it('retries a rejected nudge on the next journal edge', async () => { // A rejection consumes no mail and nothing else redrives this mailbox, so leaving it unparked // stranded the worker until unrelated mail happened to arrive. diff --git a/src/main/runtime/orchestration/structured-mailbox-pointer-host.test.ts b/src/main/runtime/orchestration/structured-mailbox-pointer-host.test.ts index 91bf1158e077..f49e03255c3a 100644 --- a/src/main/runtime/orchestration/structured-mailbox-pointer-host.test.ts +++ b/src/main/runtime/orchestration/structured-mailbox-pointer-host.test.ts @@ -80,9 +80,7 @@ describe('structured mailbox pointer host', () => { it.each([ ['accepted', 'accepted'], ['rejected', 'rejected'], - // Neither is an acknowledgement, and only `accepted` may consume mail: both have to reach the - // caller as `unknown` so the pointer is retained for the next journal edge. - ['pending', 'unknown'], + ['pending', 'pending'], ['unknown', 'unknown'] ])('maps a %s submission to %s', async (dispatchState, expected) => { const send = vi.fn( diff --git a/src/main/runtime/orchestration/structured-mailbox-pointer-host.ts b/src/main/runtime/orchestration/structured-mailbox-pointer-host.ts index b722952df92f..144dd368e00b 100644 --- a/src/main/runtime/orchestration/structured-mailbox-pointer-host.ts +++ b/src/main/runtime/orchestration/structured-mailbox-pointer-host.ts @@ -94,11 +94,12 @@ export function createStructuredMailboxPointerHost(): StructuredMailboxPointerHo ? { kind: 'unattached' } : { kind: 'sent', state: 'rejected' } } - // `pending` is not yet an acknowledgement; only `accepted` may consume mail. + // Mail consumption still requires provider acceptance, beyond transport admission. const state = result.value.submission.dispatchState return { kind: 'sent', - state: state === 'accepted' ? 'accepted' : state === 'rejected' ? 'rejected' : 'unknown' + state: + state === 'pending' || state === 'accepted' || state === 'rejected' ? state : 'unknown' } } } diff --git a/src/main/runtime/orchestration/structured-session-pointer-delivery.test.ts b/src/main/runtime/orchestration/structured-session-pointer-delivery.test.ts index 09e5010a1045..96182f106438 100644 --- a/src/main/runtime/orchestration/structured-session-pointer-delivery.test.ts +++ b/src/main/runtime/orchestration/structured-session-pointer-delivery.test.ts @@ -161,6 +161,7 @@ describe('dispatch outcome classification', () => { it('marks mail delivered only on an accepted dispatch', () => { expect(structuredDispatchDelivered('accepted')).toBe(true) expect(structuredDispatchDelivered('rejected')).toBe(false) + expect(structuredDispatchDelivered('pending')).toBe(false) }) it('does not treat unknown as delivered, because a dead child settles unknown', () => { @@ -168,6 +169,7 @@ describe('dispatch outcome classification', () => { }) it('names the retain reason for each non-accepted dispatch', () => { + expect(retainReasonForDispatch('pending')).toBe('dispatch-pending') expect(retainReasonForDispatch('rejected')).toBe('dispatch-rejected') expect(retainReasonForDispatch('unknown')).toBe('dispatch-unknown') }) diff --git a/src/main/runtime/orchestration/structured-session-pointer-delivery.ts b/src/main/runtime/orchestration/structured-session-pointer-delivery.ts index dd8ccf64f719..ec62b1b63594 100644 --- a/src/main/runtime/orchestration/structured-session-pointer-delivery.ts +++ b/src/main/runtime/orchestration/structured-session-pointer-delivery.ts @@ -23,6 +23,7 @@ export type StructuredPointerRetainReason = | 'session-not-attached' | 'turn-unsettled' | 'awaiting-human' + | 'dispatch-pending' | 'dispatch-rejected' | 'dispatch-unknown' @@ -31,7 +32,7 @@ export type StructuredPointerDecision = | { deliver: false; retain: StructuredPointerRetainReason } /** The dispatch states both provider adapters converge on. */ -export type StructuredDispatchState = 'accepted' | 'rejected' | 'unknown' +export type StructuredDispatchState = 'pending' | 'accepted' | 'rejected' | 'unknown' /** * A refusal names an owner this pointer may be redirected to only when that @@ -128,9 +129,8 @@ export function decideStructuredSessionPointerDelivery(input: { /** * Only an accepted dispatch may mark mail delivered. * - * `unknown` covers a dead provider child and a slow acknowledgement alike — the - * adapters cannot tell them apart — so it must retain. Treating it as delivered - * would drop mail whenever a child died mid-send. + * Pending admission awaits provider acceptance; unknown delivery may never settle. + * Both retain the mail and its operation ID for journal-driven reconciliation. */ export function structuredDispatchDelivered(state: StructuredDispatchState): boolean { return state === 'accepted' @@ -139,7 +139,11 @@ export function structuredDispatchDelivered(state: StructuredDispatchState): boo export function retainReasonForDispatch( state: Exclude ): StructuredPointerRetainReason { - return state === 'rejected' ? 'dispatch-rejected' : 'dispatch-unknown' + return state === 'pending' + ? 'dispatch-pending' + : state === 'rejected' + ? 'dispatch-rejected' + : 'dispatch-unknown' } /** diff --git a/src/main/runtime/rpc/methods/orchestration-structured-worker-session.test.ts b/src/main/runtime/rpc/methods/orchestration-structured-worker-session.test.ts index 493f9d092c36..3ccbe7174df1 100644 --- a/src/main/runtime/rpc/methods/orchestration-structured-worker-session.test.ts +++ b/src/main/runtime/rpc/methods/orchestration-structured-worker-session.test.ts @@ -234,10 +234,8 @@ describe('structured worker dispatch preamble', () => { const send = (host: never) => sendStructuredWorkerPreamble({ host, sessionId: 's1', dispatchId: 'd1', preamble: 'spec' }) - it('reports the preamble delivered only on an accepted submission', async () => { - await expect( - send(hostWithSubmission({ dispatchState: 'accepted', reason: null })) - ).resolves.toBeUndefined() + it.each(['pending', 'accepted'])('admits a %s preamble', async (dispatchState) => { + await expect(send(hostWithSubmission({ dispatchState, reason: null }))).resolves.toBeUndefined() }) it('never claims delivery for a submission the provider never acknowledged', async () => { @@ -245,7 +243,7 @@ describe('structured worker dispatch preamble', () => { // into `unknown`, and `performSend` still returns ok. Reporting that as `dispatch_input: // accepted` marks the worker ready with no task, and the coordinator blocks in // `check --wait --types worker_done` until it times out. - for (const dispatchState of ['unknown', 'pending'] as const) { + for (const dispatchState of ['unknown', 'future-state'] as const) { const error = await send( hostWithSubmission({ dispatchState, reason: 'provider child exited' }) ).catch((thrown: unknown) => thrown) diff --git a/src/main/runtime/rpc/methods/orchestration-structured-worker-session.ts b/src/main/runtime/rpc/methods/orchestration-structured-worker-session.ts index 5bde461a0595..f97465740e32 100644 --- a/src/main/runtime/rpc/methods/orchestration-structured-worker-session.ts +++ b/src/main/runtime/rpc/methods/orchestration-structured-worker-session.ts @@ -245,7 +245,8 @@ export async function sendStructuredWorkerPreamble(args: { throw new Error(`The dispatch preamble was refused: ${result.refusal.message}`) } const submission = result.value.submission - if (submission.dispatchState === 'accepted') { + // The host returns pending after transport admission; provider identity arrives through the journal. + if (submission.dispatchState === 'pending' || submission.dispatchState === 'accepted') { return } if (submission.dispatchState === 'rejected') { @@ -259,10 +260,7 @@ export async function sendStructuredWorkerPreamble(args: { `The dispatch preamble was not delivered: ${submission.reason ?? 'no reason given'}.` ) } - // Only `accepted` is an acknowledgement — the same rule the mail lane already applies. A thrown - // adapter call settles as `unknown`, which is indistinguishable from a lost reply, so the start - // may claim neither delivery nor failure: `operation_unknown` is what turns this into the - // `outcome_unknown` receipt whose nextCommands send the coordinator to look. + // A lost reply cannot prove admission or non-delivery; never retry the preamble here. throw new OrchestrationError( 'operation_unknown', `The dispatch preamble was submitted but not acknowledged (${submission.dispatchState}): ${submission.reason ?? 'no reason given'}.` diff --git a/src/main/runtime/rpc/methods/orchestration-structured-worker-start-failure.test.ts b/src/main/runtime/rpc/methods/orchestration-structured-worker-start-failure.test.ts index 1f29dfb49044..8f30513f4a20 100644 --- a/src/main/runtime/rpc/methods/orchestration-structured-worker-start-failure.test.ts +++ b/src/main/runtime/rpc/methods/orchestration-structured-worker-start-failure.test.ts @@ -10,6 +10,8 @@ import { beforeEach, describe, expect, it, vi } from 'vitest' import { setStructuredAgentSessionHost } from '../../../native-chat/agent-session-wire/structured-agent-session-registry' import type { OrcaRuntimeService } from '../../orca-runtime' +import { releaseStructuredWorkerSession } from './orchestration-structured-worker-session' +import { isUnknownWorkerStartOutcome } from './orchestration/worker/worker-topology' import type { OrchestrationDb } from '../../orchestration/db' import { structuredWorkerIdentities } from '../../structured-worker-identity' @@ -19,16 +21,9 @@ vi.mock('./structured-agent-session-create', () => ({ value: { sessionId: args.envelope.sessionId } }) })) -vi.mock('./orchestration-structured-worker-session', async (importOriginal) => ({ - ...(await importOriginal>()), - // The realistic post-create failure: the session is live, the preamble turn is not acknowledged. - sendStructuredWorkerPreamble: async () => { - throw new Error('The dispatch preamble was rejected: no capacity') - } -})) vi.mock('./orchestration/worker/worker-start-validation', () => ({ - prepareLocalWorkerStart: () => ({ - agent: 'claude', + prepareLocalWorkerStart: (args: { params: { agent: string } }) => ({ + agent: args.params.agent, launch: { receipt: { requested: null, effective: null }, preferences: undefined } }) })) @@ -37,9 +32,10 @@ vi.mock('./orchestration/worker/worker-setup-gate', () => ({ persistWorkerReadinessStage: () => {}, persistWorkerSetupWaitOutcome: () => {} })) -vi.mock('./orchestration/worker/worker-start-receipt', () => ({ - failWorkerStartWithReceipt: (args: { failedStage: string }) => ({ - state: 'failed', +vi.mock('./orchestration/worker/worker-start-receipt', async (importOriginal) => ({ + ...(await importOriginal>()), + failWorkerStartWithReceipt: (args: { failedStage: string; error: unknown }) => ({ + state: isUnknownWorkerStartOutcome(args.error, args.failedStage) ? 'outcome_unknown' : 'failed', stage: args.failedStage }) })) @@ -52,10 +48,17 @@ const { startLocalWorker } = await import('./orchestration/worker/local-worker-s const WORKTREE = 'wt_1' -function installHost() { +function installHost(dispatchState = 'rejected') { const closed: string[] = [] const visibility: [string, boolean][] = [] - setStructuredAgentSessionHost({ + const send = vi.fn(async () => ({ + ok: true, + value: { submission: { dispatchState, reason: 'provider outcome' } } + })) + const release = vi.fn() + const unsubscribe = vi.fn() + const host = { + send, setSessionTabVisibility: async (sessionId: string, visible: boolean) => { visibility.push([sessionId, visible]) }, @@ -64,8 +67,8 @@ function installHost() { }, hasSession: () => true, hold: async () => {}, - release: () => {}, - subscribe: () => () => {}, + release, + subscribe: () => unsubscribe, deps: { store: { getRecord: () => ({ @@ -79,8 +82,10 @@ function installHost() { }) } } - } as never) - return { closed, visibility } + } + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: This fixture implements every host member used by structured worker creation, dispatch and cleanup. + setStructuredAgentSessionHost(host as never) + return { closed, visibility, send, release, unsubscribe } } function fakes() { @@ -104,14 +109,18 @@ function fakes() { getTerminalPaneKey: vi.fn(() => 'pane_1'), retireStructuredAgentSessionTabFromSnapshot } as unknown as OrcaRuntimeService - const db = { + const dbFixture = { createStartingWorkerDispatch: () => ({ dispatch: { id: 'd_fail', depth: 0 }, task: { id: 't1', spec: 'do the thing' } }), recordWorkerStage: () => {}, - prepareStartingWorkerAuthority: () => 'capability' - } as unknown as OrchestrationDb + prepareStartingWorkerAuthority: () => 'capability', + getWorkerDispatch: () => ({ state: 'starting' }), + markWorkerDispatchReady: () => ({ state: 'ready', stage: 'ready' }) + } + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: This fixture implements the database calls reached by local worker start; receipt persistence is mocked. + const db = dbFixture as unknown as OrchestrationDb return { runtime, db, retireStructuredAgentSessionTabFromSnapshot } } @@ -149,3 +158,53 @@ describe('a structured worker-start that fails after the session exists', () => expect(retireStructuredAgentSessionTabFromSnapshot).toHaveBeenCalledWith(sessionId) }) }) + +describe.each(['codex', 'claude'] as const)('%s structured worker admission', (agent) => { + it.each(['pending', 'accepted', 'rejected', 'unknown'])( + 'preserves readiness and cleanup for %s', + async (dispatchState) => { + const host = installHost(dispatchState) + const { runtime, db, retireStructuredAgentSessionTabFromSnapshot } = fakes() + try { + const receipt = await startLocalWorker({ + params: { from: 'term_c', timeoutMs: 1_000, agent }, + mode: { + mode: 'structured', + preferred: 'structured', + reason: 'user_default', + detail: 'structured by default' + }, + runtime, + db, + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Only id is read by this start fixture. + run: { id: 'run_1' } as never, + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Only id and spec are read by this start fixture. + existingTask: { id: 't1', spec: 'do the thing' } as never, + coordinatorPane: null, + orchestrationMutation: undefined + }) + expect(host.send).toHaveBeenCalledTimes(1) + if (dispatchState === 'pending' || dispatchState === 'accepted') { + expect(receipt).toMatchObject({ state: 'ready', turnStart: 'unsupported' }) + expect(host.closed).toHaveLength(0) + expect(host.release).not.toHaveBeenCalled() + expect(host.unsubscribe).not.toHaveBeenCalled() + expect(retireStructuredAgentSessionTabFromSnapshot).not.toHaveBeenCalled() + } else { + expect(receipt).toMatchObject({ + state: dispatchState === 'unknown' ? 'outcome_unknown' : 'failed', + stage: 'dispatch_input' + }) + expect(host.closed).toHaveLength(1) + expect(host.release).toHaveBeenCalledTimes(1) + expect(host.unsubscribe).toHaveBeenCalledTimes(1) + expect(retireStructuredAgentSessionTabFromSnapshot).toHaveBeenCalledTimes(1) + } + } finally { + releaseStructuredWorkerSession('d_fail', runtime) + } + expect(host.release).toHaveBeenCalledTimes(1) + expect(host.unsubscribe).toHaveBeenCalledTimes(1) + } + ) +}) diff --git a/src/main/runtime/rpc/methods/orchestration/worker/deliver-worker-dispatch-preamble.ts b/src/main/runtime/rpc/methods/orchestration/worker/deliver-worker-dispatch-preamble.ts index 3da57f9c530e..292a37c35455 100644 --- a/src/main/runtime/rpc/methods/orchestration/worker/deliver-worker-dispatch-preamble.ts +++ b/src/main/runtime/rpc/methods/orchestration/worker/deliver-worker-dispatch-preamble.ts @@ -11,7 +11,7 @@ type StructuredSession = Awaited { }) }) + it('admits a fresh worker preamble before the provider echo without resending', async () => { + await ok('agentSession.create', createIntentParams()) + await expectStructuredWorkerAdmission(SESSION) + expect(codex.live().calls.filter((call) => call.method === 'turn/start')).toHaveLength(1) + }) + it('dispatches and streams a plain first send from a fresh session', async () => { const created = await ok<{ fence: number; page: { items: unknown[] } }>( 'agentSession.create', diff --git a/src/main/runtime/structured-worker-admission.test-support.ts b/src/main/runtime/structured-worker-admission.test-support.ts new file mode 100644 index 000000000000..1e30683f2779 --- /dev/null +++ b/src/main/runtime/structured-worker-admission.test-support.ts @@ -0,0 +1,24 @@ +import { expect } from 'vitest' +import { getStructuredAgentSessionHost } from '../native-chat/agent-session-wire/structured-agent-session-registry' +import { sendStructuredWorkerPreamble } from './rpc/methods/orchestration-structured-worker-session' + +export async function expectStructuredWorkerAdmission(sessionId: string): Promise { + const host = getStructuredAgentSessionHost() + if (!host) { + throw new Error('structured host missing') + } + const preamble = 'Complete the dispatched task once.' + await expect( + sendStructuredWorkerPreamble({ + host, + sessionId, + dispatchId: 'worker-admission', + preamble + }) + ).resolves.toBeUndefined() + const snapshot = host.journalSnapshot(sessionId) + expect(snapshot.submissions).toMatchObject([{ dispatchState: 'pending' }]) + expect(snapshot.items.filter((item) => item.body.kind === 'message')).toMatchObject([ + { body: { role: 'user', blocks: [{ type: 'text', text: preamble }] } } + ]) +} diff --git a/src/shared/agent-status-legacy-adapter.ts b/src/shared/agent-status-legacy-adapter.ts index cbed5778a78f..a80d7ec6f2b2 100644 --- a/src/shared/agent-status-legacy-adapter.ts +++ b/src/shared/agent-status-legacy-adapter.ts @@ -92,7 +92,8 @@ function freezeRecursively(value: unknown, seen: WeakSet): void { } seen.add(value) for (const key of Reflect.ownKeys(value)) { - freezeRecursively(Reflect.get(value, key), seen) + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: `value` is an object, so indexed access preserves prototype lookup and getter receiver semantics. + freezeRecursively((value as Record)[key], seen) } Object.freeze(value) }