diff --git a/mobile/src/session/mobile-native-chat-unsent-notices.test.ts b/mobile/src/session/mobile-native-chat-unsent-notices.test.ts index 74a2574046c..321c4177222 100644 --- a/mobile/src/session/mobile-native-chat-unsent-notices.test.ts +++ b/mobile/src/session/mobile-native-chat-unsent-notices.test.ts @@ -82,10 +82,18 @@ describe("the phone's line under a not-sent message", () => { it('says only that it was not sent when a loaded row already says why', () => { const notices = mobileNativeChatUnsentNotices( - { items: [startRow], submissions: [submission('lost', { rejection: startFailed })] }, + { + items: [startRow], + submissions: [ + submission('lost', { rejection: startFailed }), + submission('later', { dispatchState: 'pending', resolvedAt: null }) + ] + }, 'claude' ) - expect(notices.get(agentJournalSubmissionKey('lost'))).toBe('Your message was not sent.') + expect([...notices]).toEqual([ + [agentJournalSubmissionKey('lost'), 'Your message was not sent.'] + ]) }) it('gives one shared empty map when no send failed, a Stop taking one back included', () => { diff --git a/mobile/src/session/mobile-structured-session-status-feed.test.ts b/mobile/src/session/mobile-structured-session-status-feed.test.ts index ed0d680fb81..aa6a2256a94 100644 --- a/mobile/src/session/mobile-structured-session-status-feed.test.ts +++ b/mobile/src/session/mobile-structured-session-status-feed.test.ts @@ -83,6 +83,19 @@ describe("the phone's status stream", () => { expect(listener).toHaveBeenCalledTimes(2) }) + it('keeps a retrying host owned and starting until readiness, then revokes ownership on disconnect', () => { + mobileStructuredSessionStatusFeed(client).subscribe(() => {}) + const starting = { ...summary(false), hostExecutionPhase: 'starting' as const } + frames[0]?.({ type: 'snapshot', sessions: [starting] }) + frames[0]?.({ type: 'status', session: starting }) + expect(read()).toMatchObject({ hostExecutionOwned: true, hostExecutionPhase: 'starting' }) + frames[0]?.({ type: 'status', session: { ...starting, hostExecutionPhase: 'ready' } }) + expect(read()).toMatchObject({ hostExecutionOwned: true, hostExecutionPhase: 'ready' }) + connection('reconnecting') + expect(read()?.hostExecutionOwned).toBeUndefined() + expect(read()?.hostExecutionPhase).toBeUndefined() + }) + it('drops Stopping when the host ends the stream, and opens it again on the next connection', () => { mobileStructuredSessionStatusFeed(client).subscribe(() => {}) frames[0]?.({ type: 'snapshot', sessions: [summary(true)] }) diff --git a/src/main/agent-hooks/server-ingest-structured-status.test.ts b/src/main/agent-hooks/server-ingest-structured-status.test.ts index fb0ce6a3910..cce5340ea14 100644 --- a/src/main/agent-hooks/server-ingest-structured-status.test.ts +++ b/src/main/agent-hooks/server-ingest-structured-status.test.ts @@ -118,6 +118,20 @@ describe('AgentHookServer ingestStructuredStatus', () => { expect(server.getStatusSnapshot()[0]?.observation?.origin).toBe('structured') }) + it('keeps a logical start owned between physical children without changing the latest turn', () => { + const server = new AgentHookServer() + const starting = summary({ status: 'idle', hostExecutionPhase: 'starting' }) + server.ingestStructuredStatus(starting, SUBJECT) + server.ingestStructuredStatus(starting, SUBJECT) + expect(server.getStatusSnapshot()[0]).toMatchObject({ + structuredHost: 'owned', + state: 'done', + stateStartedAt: OBSERVED_AT + }) + server.ingestStructuredStatus({ ...starting, hostExecutionPhase: 'ready' }, SUBJECT) + expect(server.getStatusSnapshot()[0]).toMatchObject({ structuredHost: 'owned', state: 'done' }) + }) + it('maps attention to blocked and idle to done', () => { const server = new AgentHookServer() server.ingestStructuredStatus(summary({ status: 'attention' }), SUBJECT) diff --git a/src/main/claude/claude-structured-session-adapter.test.ts b/src/main/claude/claude-structured-session-adapter.test.ts index d3ce02a5c9b..54c3c0f062b 100644 --- a/src/main/claude/claude-structured-session-adapter.test.ts +++ b/src/main/claude/claude-structured-session-adapter.test.ts @@ -25,8 +25,34 @@ import { } from './claude-structured-session-test-support' import { invokeCanUseTool } from './claude-can-use-tool-test-support' import { claudeProviderHandle } from '../../shared/agent-session-provider-handle-encoding' +import { providerRequestTimeoutOf } from '../provider-process/provider-request-timeout' describe('ClaudeStructuredSessionAdapter.acquire', () => { + it('ends a stalled initialize at its 60-second deadline with the timed-out step', async () => { + vi.useFakeTimers() + try { + const claude = fakeClaude({ initDelayMs: 60_001 }) + const adapter = adapterFor(claude) + let settled = false + const result = adapter + .acquire({ identity: identityFor(), fence: 7, spawnToken: 'deadline' }) + .catch((error: unknown) => { + settled = true + return error + }) + await vi.advanceTimersByTimeAsync(59_999) + expect(settled).toBe(false) + await vi.advanceTimersByTimeAsync(1) + expect(providerRequestTimeoutOf(await result)).toEqual({ + method: 'initialize', + timeoutMs: 60_000 + }) + expect(claude.connections).toHaveLength(1) + } finally { + vi.useRealTimers() + } + }) + it('pins the account and proves init without treating the system-frame uuid as a chain leaf', async () => { const claude = fakeClaude() const events: ClaudeStructuredSessionEvent[] = [] @@ -562,6 +588,20 @@ describe('ClaudeStructuredSessionAdapter acquisition cleanup', () => { await expect(adapter.releaseAcquisition({ sessionId: 'session-1' })).resolves.toBe(true) }) + it('forgets the stored crash when the host acknowledges session release', async () => { + const claude = fakeClaude() + const adapter = await acquired(claude) + claude.connections[0].handlers.onExit?.(new Error('retained crash')) + await adapter.drainObservedExits() + expect(() => adapter.readOptions({ sessionId: 'session-1', fence: 7 })).toThrow( + 'retained crash' + ) + adapter.acknowledgeSessionRelease('session-1') + expect(() => adapter.readOptions({ sessionId: 'session-1', fence: 7 })).toThrow( + 'no live claude stream-json session' + ) + }) + it('keeps shutdown pending until a retained unexpected-exit proof settles', async () => { const claude = fakeClaude() const adapter = await acquired(claude) diff --git a/src/main/claude/claude-structured-session-adapter.ts b/src/main/claude/claude-structured-session-adapter.ts index 8afc7b7dff1..ae9592b29ad 100644 --- a/src/main/claude/claude-structured-session-adapter.ts +++ b/src/main/claude/claude-structured-session-adapter.ts @@ -241,6 +241,9 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda releaseAcquisition = (input: { sessionId: string }): Promise => this.afterClose(input.sessionId, () => this.releaseProviderSession(input.sessionId)) + acknowledgeSessionRelease = (sessionId: string): void => { + this.settledExitErrors.delete(sessionId) + } hasOwnedAcquisition = (sessionId: string): boolean => this.acquisitions.get(sessionId) !== undefined || this.sessions.has(sessionId) || diff --git a/src/main/native-chat/agent-session-wire/__fixtures__/startup-protocol-rejections.json b/src/main/native-chat/agent-session-wire/__fixtures__/startup-protocol-rejections.json new file mode 100644 index 00000000000..fcb217b12b3 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/__fixtures__/startup-protocol-rejections.json @@ -0,0 +1,149 @@ +{ + "provenance": { + "capturedAt": "2026-10-24T08:14:01Z", + "environment": "M4Air, isolated qa-with-lb protocol capture", + "scope": "Real representative command rejections, not startup-auth captures or Pi/ACP app lifecycle runs.", + "stdinEnd": "Input EOF after the error response; each CLI exited 0 (qa-report.md protocol section).", + "sanitization": "Only failing request, exact error response, invocation/version and exit retained. Working directories and spawn identities omitted; ACP request cwd replaced with .", + "source": "PR #33045 QA transfer, evidence/protocol-sanitized", + "provenanceSha256": "fe1a2e3ce30ec2ce980833c628fb392958310461d53cce83f375b22ccaa97da8" + }, + "claude": { + "sourceFile": "claude-control.jsonl", + "sourceSha256": "db149878a78961e5ac76197287eac33e22b30780c9ee88075b548936624cff38", + "command": [ + "/Users/m4air/.local/bin/claude", + "--print", + "--input-format", + "stream-json", + "--output-format", + "stream-json", + "--verbose" + ], + "version": "2.1.310 (Claude Code)", + "request": { + "type": "control_request", + "request_id": "qa-error", + "request": { + "subtype": "set_permission_mode", + "mode": "qa-invalid-33045" + } + }, + "response": { + "type": "control_response", + "response": { + "subtype": "error", + "request_id": "qa-error", + "error": "Cannot set permission mode: must be one of acceptEdits, auto, bypassPermissions, default, dontAsk, plan", + "error_code": "invalid_mode" + } + }, + "exit": { + "code": 0, + "signal": null, + "elapsedMs": 528, + "receivedFinal": true + } + }, + "codex": { + "sourceFile": "codex-rpc.jsonl", + "sourceSha256": "0f2dd5ac418bc63e1640df6aadd05bf9b1eed40deaecd71626a5ef8aa85a3531", + "command": ["/opt/homebrew/bin/codex", "app-server"], + "version": "codex-cli 0.163.0", + "request": { + "id": 2, + "method": "thread/read", + "params": { + "threadId": "qa-missing-33045", + "includeTurns": true + } + }, + "response": { + "error": { + "code": -32600, + "message": "invalid thread id: invalid character: expected an optional prefix of `urn:uuid:` followed by [0-9a-fA-F-], found `q` at 1" + }, + "id": 2 + }, + "exit": { + "code": 0, + "signal": null, + "elapsedMs": 90, + "receivedFinal": true + } + }, + "pi": { + "sourceFile": "pi-jsonl.jsonl", + "sourceSha256": "7a60aa4dda389cf68c2421d4ec3dfc7d5cb68daffc493a8a4fae24ce57b93b46", + "command": [ + "/opt/homebrew/bin/pi", + "--mode", + "rpc", + "--no-session", + "--provider", + "openai-codex" + ], + "version": "0.73.1", + "request": { + "id": "qa-pi-error", + "type": "set_model", + "provider": "openai", + "modelId": "qa-nonexistent-33045" + }, + "response": { + "id": "qa-pi-error", + "type": "response", + "command": "set_model", + "success": false, + "error": "Model not found: openai/qa-nonexistent-33045" + }, + "exit": { + "code": 0, + "signal": null, + "elapsedMs": 1009, + "receivedFinal": true + }, + "versionEvidence": { + "sourceFile": "pi-version-fixed-receipt.json", + "sourceSha256": "b8c747b9dccdfe9fd4738d359bb2d159159c49b78ff62e6734810586aaa684e9", + "captureCommand": ["qa-with-lb", "--", "node", "-e", "spawnSync pi --version"], + "command": ["/opt/homebrew/bin/pi", "--version"], + "stdout": "", + "stderr": "0.73.1\n", + "exit": 0 + } + }, + "acp": { + "sourceFile": "acp-rpc.jsonl", + "sourceSha256": "c026e934ce19ac919b42f75f6b9f350022a00dbd5a70e8b34d1e3a490df52b24", + "command": ["/opt/homebrew/bin/opencode", "acp"], + "version": "opencode v2.0.21", + "request": { + "jsonrpc": "2.0", + "id": 2, + "method": "session/load", + "params": { + "sessionId": "qa-missing-33045", + "cwd": "", + "mcpServers": [] + } + }, + "response": { + "jsonrpc": "2.0", + "id": 2, + "error": { + "code": -32603, + "message": "Internal error: Internal service failure", + "data": { + "errorName": "InvalidRequestError" + } + } + }, + "exit": { + "code": 0, + "signal": null, + "elapsedMs": 236, + "receivedFinal": true + } + } +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-accept-then-deliver.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-accept-then-deliver.test.ts index 5752b73a186..2b370d427b0 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-accept-then-deliver.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-accept-then-deliver.test.ts @@ -353,13 +353,14 @@ describe('a start the chat needed and did not get', () => { expect(JSON.stringify(await host.journalSnapshot(SESSION))).not.toContain('private') }) - it('writes one error row and rejects every queued message with it; the next send starts (W3)', async () => { + it('writes one error row for the requester and delivers a later queued message (W3)', async () => { await host.close(SESSION, 'evict') acquire.mockRejectedValueOnce(new Error('spawn codex ENOENT')) const first = await accept('first') const second = await accept('second') - await eventually(async () => expect((await submission(second))?.dispatchState).toBe('rejected')) + await eventually(async () => expect((await submission(first))?.dispatchState).toBe('rejected')) + await eventually(async () => expect((await submission(second))?.dispatchState).toBe('accepted')) const rows = await errorRows() expect(rows).toHaveLength(1) // Log Orca’s spawn error; absent observed exit must not blame the provider. @@ -369,31 +370,30 @@ describe('a start the chat needed and did not get', () => { refusal: { code: 'agent_session_operation_invalid', details: { ownerVerdict: 'exited' } } } expect(await errorFailures()).toEqual([failure]) - for (const id of [first, second]) { - expect(await submission(id)).toMatchObject({ - dispatchState: 'rejected', - reason: rows[0], - rejection: failure - }) - } + expect(await submission(first)).toMatchObject({ + dispatchState: 'rejected', + reason: rows[0], + rejection: failure + }) const next = await accept('after the fix') await eventually(async () => expect((await submission(next))?.dispatchState).toBe('accepted')) expect(await errorRows()).toHaveLength(1) }) - it('draws the messages it failed above the error row, since they were accepted first', async () => { + it('draws the failed requester before its error row and the later delivered message', async () => { await host.close(SESSION, 'evict') acquire.mockRejectedValueOnce(new Error('spawn codex ENOENT')) const first = await accept('first') + await eventually(async () => expect((await submission(first))?.dispatchState).toBe('rejected')) const second = await accept('second') - await eventually(async () => expect((await submission(second))?.dispatchState).toBe('rejected')) + await eventually(async () => expect((await submission(second))?.dispatchState).toBe('accepted')) const snapshot = await host.journalSnapshot(SESSION) const errorRow = snapshot.items.find( (item) => item.body.kind === 'status' && item.body.tone === 'error' )?.itemId - const shown = [agentJournalSubmissionKey(first), agentJournalSubmissionKey(second), errorRow] + const shown = [agentJournalSubmissionKey(first), errorRow, agentJournalSubmissionKey(second)] const drawn = hostTestDrawnRowIds(snapshot, [ { clientMessageId: first, text: 'first' }, { clientMessageId: second, text: 'second' } @@ -401,22 +401,23 @@ describe('a start the chat needed and did not get', () => { expect(drawn.filter((id) => shown.includes(id))).toEqual(shown) }) - it('notifies failed once for the queued messages one start failure refused', async () => { + it('notifies failed once for the requester whose start failed', async () => { await host.close(SESSION, 'evict') acquire.mockRejectedValueOnce(new Error('spawn codex ENOENT')) const completions: AgentSessionTurnCompletionEvent[] = [] host.subscribeTurnCompletions({ id: 'dot-1', emit: (event) => completions.push(event) }) - await accept('first') + const first = await accept('first') const second = await accept('second') - await eventually(async () => expect((await submission(second))?.dispatchState).toBe('rejected')) + await eventually(async () => expect((await submission(first))?.dispatchState).toBe('rejected')) + await eventually(async () => expect((await submission(second))?.dispatchState).toBe('accepted')) await host.flushAllStreamedEvents() expect(completions).toEqual([ { type: 'completion', completion: expect.objectContaining({ sessionId: SESSION, - turnId: agentJournalSubmissionKey(second), + turnId: agentJournalSubmissionKey(first), outcome: 'failure' }) } @@ -590,7 +591,7 @@ describe('what an earlier host process left behind', () => { }) describe('a child that exits before open completes', () => { - it('rejects the message with the start failure instead of starting another child (W24)', async () => { + it('rejects the requester once after five children exit before opening (W24)', async () => { await host.close(SESSION, 'evict') await startHost() acquire.mockImplementation(async (input) => { @@ -615,34 +616,42 @@ describe('a child that exits before open completes', () => { reason: 'Codex stopped before it finished starting. Send your message to try again.', rejection: { kind: 'providerStartFailed' } }) - expect(acquire).toHaveBeenCalledTimes(2) + expect(acquire).toHaveBeenCalledTimes(6) expect(dispatch).not.toHaveBeenCalled() }) }) describe('a start that fails while messages wait on it', () => { - it('keeps one row in the words its rejected messages carry', async () => { + it('keeps one final row for its requester while a joiner gets its own start', async () => { acquire.mockImplementation(spawnStartingChild) await host.close(SESSION, 'evict') await startHost() + const priorStarts = acquire.mock.calls.length const first = await accept('first') const second = await accept('second') - await eventually(() => expect(childPhase()).toBe('starting')) - await host.handleAdapterEvent({ - type: 'ended', - sessionId: SESSION, - fence: store.getRecord(SESSION)!.lease.runtimeFence, - acquisitionGeneration: `generation-${acquire.mock.calls.length}`, - reason: 'codex app-server exited with code 1', - failure: agentSessionFailureFact('providerStartFailed', { - detail: { text: 'codex: config.toml is invalid', audience: 'person' } - }), - cause: 'unexpected-exit' - }) - await host.flushStreamedEvents(SESSION) + for (let attempt = 1; attempt <= 5; attempt++) { + await eventually(() => expect(acquire).toHaveBeenCalledTimes(priorStarts + attempt)) + await eventually(() => expect(childPhase()).toBe('starting')) + expect(await errorRows()).toEqual([]) + expect((await submission(first))?.dispatchState).toBe('pending') + await host.handleAdapterEvent({ + type: 'ended', + sessionId: SESSION, + fence: store.getRecord(SESSION)!.lease.runtimeFence, + acquisitionGeneration: `generation-${acquire.mock.calls.length}`, + reason: 'codex app-server exited with code 1', + failure: agentSessionFailureFact('providerStartFailed', { + detail: { text: 'codex: config.toml is invalid', audience: 'person' } + }), + cause: 'unexpected-exit' + }) + await host.flushStreamedEvents(SESSION) + } - await eventually(async () => expect((await submission(second))?.dispatchState).toBe('rejected')) + await eventually(async () => expect((await submission(first))?.dispatchState).toBe('rejected')) + await eventually(() => expect(acquire).toHaveBeenCalledTimes(priorStarts + 6)) + expect((await submission(second))?.dispatchState).toBe('pending') const rows = (await host.journalSnapshot(SESSION)).items.filter((item) => item.itemId.includes('start-failure') ) @@ -651,13 +660,11 @@ describe('a start that fails while messages wait on it', () => { expect(row).toMatchObject({ failure: { kind: 'providerStartFailed', detail: { text: 'codex: config.toml is invalid' } } }) - for (const id of [first, second]) { - expect(await submission(id)).toMatchObject({ - dispatchState: 'rejected', - reason: row.kind === 'status' ? row.text : null, - rejection: row.kind === 'status' ? row.failure : null - }) - } + expect(await submission(first)).toMatchObject({ + dispatchState: 'rejected', + reason: row.kind === 'status' ? row.text : null, + rejection: row.kind === 'status' ? row.failure : null + }) expect(dispatch).not.toHaveBeenCalled() }) }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-acquire-aborts.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-acquire-aborts.ts index 97e2f9c2227..44cd322d073 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-acquire-aborts.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-acquire-aborts.ts @@ -1,7 +1,7 @@ export class StructuredAgentSessionAcquireAborts { private readonly inFlight = new Map< string, - Set<{ controller: AbortController; handover: boolean; goal: boolean }> + Set<{ controller: AbortController; handover: boolean; goal: boolean; opening: boolean }> >() /** Starts admitted after quit must begin aborted. */ private quitReason: Error | null = null @@ -10,7 +10,8 @@ export class StructuredAgentSessionAcquireAborts { begin( sessionId: string, handover = false, - goal = false + goal = false, + opening = false ): { signal: AbortSignal; abort: () => void; end: () => void } { const controller = new AbortController() if (this.quitReason) { @@ -18,8 +19,8 @@ export class StructuredAgentSessionAcquireAborts { } const waits = this.inFlight.get(sessionId) ?? - new Set<{ controller: AbortController; handover: boolean; goal: boolean }>() - const active = { controller, handover, goal } + new Set<{ controller: AbortController; handover: boolean; goal: boolean; opening: boolean }>() + const active = { controller, handover, goal, opening } waits.add(active) this.inFlight.set(sessionId, waits) return { @@ -55,6 +56,14 @@ export class StructuredAgentSessionAcquireAborts { return { reached, handedOver } } + abortOpen(sessionId: string): void { + for (const wait of this.inFlight.get(sessionId) ?? []) { + if (wait.opening) { + wait.controller.abort(new Error('provider exited while opening')) + } + } + } + /** Abort all waits, including queued attach-drain work. */ abortAll(reason: string): void { this.quitReason ??= new Error(reason) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.ts index f9c4e921859..6837149bc78 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.ts @@ -256,6 +256,9 @@ export class StructuredAgentSessionAdapterRouter implements StructuredAgentSessi acquisition?: StructuredAgentSessionAcquisitionReceipt ): void => { if (!acquisition || this.stoppedAcquisition(sessionId) === acquisition) { + this.routes + .get(sessionId) + ?.registration.adapter.acknowledgeSessionRelease?.(sessionId, acquisition) this.routes.delete(sessionId) } } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-agent-start.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-agent-start.ts index 7d70a2f626c..e9bb1c7216f 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-agent-start.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-agent-start.ts @@ -3,28 +3,29 @@ import { providerDiagnosticOf, + type SubmissionRejectionFact, type AgentSessionArgumentProblem, type ProviderDiagnostic } from '../../../shared/agent-session-failure' import { argumentProblemOf } from '../structured-agent-arguments-error' import { agentSessionRefusalFromReference, - readAgentSessionRefusalReference, - refuseUnclassified + readAgentSessionRefusalReference } from '../../../shared/agent-session-wire-refusals' -import { agentSessionWriteNoticeEnglish } from '../../../shared/agent-session-refusal-notice' import type { AgentSessionAttachResult, AgentSessionMutationResult, AgentSessionWireRefusal } from '../../../shared/agent-session-wire' import type { StructuredAgentSessionAttachContext } from './structured-agent-session-attach-context' -import { isQueuedAgentJournalSubmission } from '../../../shared/agent-session-queued-submission' -import { randomUUID } from 'node:crypto' import { attachStructuredAgentSessionForStart } from './structured-agent-session-attach-orchestration' import type { StructuredAgentSessionAttachLock } from './structured-agent-session-attach-lock' -import { structuredAgentSessionStartFlightLock } from './structured-agent-session-start-flight' -import { isStructuredAgentSessionStartupExpired } from './structured-agent-session-startup-attempt-contract' +import { + failedStructuredAgentSessionStart, + registerStructuredAgentSessionStart, + structuredAgentSessionStartStillQueued +} from './structured-agent-session-start-attempts' +import { retryableStructuredAgentSessionAcquisition } from './structured-agent-session-start-retry' import { failedCreateRefusal } from './structured-agent-session-failed-create-refusal' import { joinClosingStructuredAgentSessionChild, @@ -49,6 +50,9 @@ export type StructuredAgentSessionResumeOutcome = /** Startup silence/ceiling expiry means Orca stopped the start. */ expired?: true argumentProblem?: AgentSessionArgumentProblem + retry?: boolean + failure?: SubmissionRejectionFact + startKey?: string | null } const AGENT_START_CALLER_KEY = 'trusted-local:agent-start' @@ -65,24 +69,23 @@ export function claimStructuredAgentSessionAgent( ): StructuredAgentSessionStartClaim { const session = context.sessions.get(sessionId) const child = session?.child - if ( - startedFor && - !session?.journal - .submissions() - .some( - (submission) => - submission.clientMessageId === startedFor && isQueuedAgentJournalSubmission(submission) - ) - ) { - return { outcome: Promise.resolve({ ok: false, aborted: true, refusal: startFailedRefusal() }) } + if (!structuredAgentSessionStartStillQueued(session, startedFor)) { + return { outcome: Promise.resolve({ ...failedStructuredAgentSessionStart(), aborted: true }) } + } + const flight = context.runtimeState.startFlights.get(sessionId) + if (flight) { + return { outcome: flight.outcome } } if (child && !child.close) { return { outcome: Promise.resolve({ ok: true }) } } return { - outcome: - context.runtimeState.startFlights.get(sessionId)?.outcome ?? - registerStart(context, sessionId, startedFor) + outcome: registerStructuredAgentSessionStart( + context, + sessionId, + startedFor, + (lock, attemptId) => runStart(context, sessionId, startedFor, lock, attemptId) + ) } } @@ -118,56 +121,11 @@ export function claimOperationStart( sessionId, error }) - return { ok: false, refusal: startFailedRefusal() } + return failedStructuredAgentSessionStart() }) } } -/** Register cancellation, startup limits and quit drainage before the first wait. */ -function registerStart( - context: StructuredAgentSessionAttachContext, - sessionId: string, - startedFor: string | undefined -): Promise { - const { runtimeState } = context - const attemptId = randomUUID() - const acquire = runtimeState.acquireAborts.begin(sessionId) - runtimeState.startupAttempts.begin(sessionId, attemptId) - const lock = structuredAgentSessionStartFlightLock({ - sessionId, - attemptId, - signal: acquire.signal, - flights: runtimeState.startFlights, - serialize: context.serialize - }) - const outcome = context.tasks.trackAttach( - Promise.resolve() - .then(() => runStart(context, sessionId, startedFor, lock, attemptId)) - .catch((error: unknown) => { - if (!acquire.signal.aborted) { - throw error - } - const expired = isStructuredAgentSessionStartupExpired(acquire.signal.reason) - return { - ok: false as const, - refusal: startFailedRefusal(), - ...(expired ? { expired: true as const } : { aborted: true as const }) - } - }) - .finally(() => { - acquire.end() - runtimeState.startFlights.end(sessionId, attemptId) - runtimeState.startupAttempts.abandon(sessionId, attemptId) - }) - ) - runtimeState.startFlights.register(sessionId, { - attemptId, - outcome, - abort: acquire.abort - }) - return outcome -} - async function runStart( context: StructuredAgentSessionAttachContext, sessionId: string, @@ -246,13 +204,6 @@ async function runStart( ) } -function startFailedRefusal(): AgentSessionWireRefusal { - return refuseUnclassified( - 'agent_session_owner_restart_failed', - agentSessionWriteNoticeEnglish(['restartFailed']) - ) -} - function withDiagnostic( refusal: AgentSessionWireRefusal, error: unknown, @@ -266,7 +217,8 @@ function withDiagnostic( ...(diagnostic ? { diagnostic } : {}), ...(aborted ? { aborted: true as const } : {}), ...(expired ? { expired: true as const } : {}), - ...(argumentProblem ? { argumentProblem } : {}) + ...(argumentProblem ? { argumentProblem } : {}), + retry: retryableStructuredAgentSessionAcquisition(error) } } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-attach-publish.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-publish.ts index 7404cc61b85..80669924c26 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-attach-publish.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-publish.ts @@ -111,17 +111,20 @@ export async function publishAttachedChild( context.runtimeState.adoptEventSink(sessionId, candidate.sink) attempt.committed = candidate.sink === input.attemptSink indexProviderChild(conversation, candidate.child) + context.runtimeState.startFlights.published(sessionId, candidate.child) + if (attempt.startup) { + context.runtimeState.startupAttempts.published( + sessionId, + attempt.startup.attemptId, + candidate.child + ) + } // Apply exits observed during background startup after indexing the matching child. const exited = context.runtimeState.startFlights.takeExit(sessionId, candidate.child) const indexed = conversation.child if (exited && indexed) { await context.endExitedChild(sessionId, indexed, structuredAgentSessionEndedEventExit(exited)) } else if (attempt.startup) { - context.runtimeState.startupAttempts.published( - sessionId, - attempt.startup.attemptId, - candidate.child - ) if (candidate.child.phase === 'ready') { noteStructuredAgentSessionProviderStarted(context.deps, sessionId) } else { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-child-exit.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-child-exit.ts index 333c660a9c2..4c9d7ca8adc 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-child-exit.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-child-exit.ts @@ -68,7 +68,7 @@ export type StructuredAgentSessionChildExitContext< wakeDelivery?: (sessionId: string) => void startupAttempts?: Pick /** Retain exits between background-start spawn and publication. */ - startFlights?: Pick + startFlights?: Pick abortOpen?: (sessionId: string) => void serialize: (sessionId: string, task: () => Promise) => Promise now: () => number @@ -245,7 +245,12 @@ export async function endExitedStructuredAgentSessionChildUnderSerialize< )), ...(startFailure ? { exitFailure: startFailure } : {}), ...(startFailed && child.generation - ? { exitedDuringStartup: { generation: child.generation } } + ? { + exitedDuringStartup: { + generation: child.generation, + flightOwned: context.startFlights?.ownsChild(sessionId, child) + } + } : {}), exit: watched, ...(unrunRejection ? { unrunRejection } : {}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-client-delivery.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-client-delivery.ts index 35e41bbba8b..8fd442fe040 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-client-delivery.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-client-delivery.ts @@ -49,7 +49,8 @@ export class StructuredAgentSessionClientDelivery { onAgentStarted: (sessionId: string) => void, onChildWorkChanged: (sessionId: string) => void, // Current clients interpret an absent opening roster as no tasks. - readBackgroundTasks: NonNullable + readBackgroundTasks: NonNullable, + starting?: (sessionId: string) => boolean ) { this.requestFailures = new StructuredAgentSessionRequestFailures({ sessions, @@ -63,6 +64,7 @@ export class StructuredAgentSessionClientDelivery { }) this.statusFeed = createStructuredAgentSessionHostStatusFeed({ sessions, + starting, now, currentWork: (sessionId, journal) => this.currentWork(sessionId, journal), deps, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-lifetime-host.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-lifetime-host.ts new file mode 100644 index 00000000000..57b3e277325 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-lifetime-host.ts @@ -0,0 +1,16 @@ +import type { StructuredAgentSessionLifetimeContext } from './structured-agent-session-host-lifetime' +import type { StructuredAgentSessionConversations } from './structured-agent-session-conversations' +import type { StructuredAgentSessionHostSession } from './structured-agent-session-host-types' + +export type StructuredAgentSessionConversationLifetimeHost = { + /** Resolve lazily because host collaborators are assigned after construction. */ + context: () => StructuredAgentSessionLifetimeContext + sessions: StructuredAgentSessionConversations + serialize: (sessionId: string, task: () => Promise) => Promise + /** Caller already holds session serialization. */ + open: (sessionId: string) => Promise + deliveryActive: (sessionId: string) => boolean + /** Retain status rows while a tab still lists the chat. */ + closeStatus: (sessionId: string, options: { listed: boolean }) => void + childWorkOpen: (sessionId: string) => boolean +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-lifetime.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-lifetime.ts index 7f27303fdfa..0fe53dad12f 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-lifetime.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-lifetime.ts @@ -1,3 +1,5 @@ +import type { StructuredAgentSessionConversationLifetimeHost } from './structured-agent-session-conversation-lifetime-host' +import { retryableStructuredAgentSessionStart } from './structured-agent-session-start-retry' // Open conversations at rest; never start agents for reads. Acquire serialization once because the queue is nonreentrant. import { @@ -5,14 +7,12 @@ import { agentSessionRefusalError } from '../../../shared/agent-session-wire-refusals' import { createJournalOpenReadRefusals } from '../agent-session-journal/journal-open-failure' -import type { StructuredAgentSessionConversations } from './structured-agent-session-conversations' import { releaseLeaseOfEndedStructuredAgentSessionChild } from './structured-agent-session-child-close' import { holdClosedStructuredAgentSessionSends, closeStructuredAgentSessionConversationUnderSerialize, stopStructuredAgentSessionAgentUnderSerialize, type StructuredAgentSessionCloseCause, - type StructuredAgentSessionLifetimeContext, type StructuredAgentSessionStopEnding } from './structured-agent-session-host-lifetime' import type { @@ -34,25 +34,19 @@ import { providerStartupFailureFact, structuredAgentSessionStartFailure } from './structured-agent-session-failure-text' -import { recordStructuredAgentSessionStartFailure } from './structured-agent-session-start-failure-row' +import { + oldestQueuedSubmission, + recordStructuredAgentSessionStartFailure +} from './structured-agent-session-start-failure-row' import { structuredAgentSessionFailureWordsContext } from './structured-agent-session-send-preparation' export type StructuredAgentSessionConversationLifetime = ReturnType< typeof createStructuredAgentSessionConversationLifetime > -export function createStructuredAgentSessionConversationLifetime(host: { - /** Resolve lazily because host collaborators are assigned after construction. */ - context: () => StructuredAgentSessionLifetimeContext - sessions: StructuredAgentSessionConversations - serialize: (sessionId: string, task: () => Promise) => Promise - /** Caller already holds session serialization. */ - open: (sessionId: string) => Promise - deliveryActive: (sessionId: string) => boolean - /** Retain status rows while a tab still lists the chat. */ - closeStatus: (sessionId: string, options: { listed: boolean }) => void - childWorkOpen: (sessionId: string) => boolean -}) { +export function createStructuredAgentSessionConversationLifetime( + host: StructuredAgentSessionConversationLifetimeHost +) { let disposed = false const { sessions, serialize } = host const deps = () => host.context().deps @@ -246,30 +240,35 @@ export function createStructuredAgentSessionConversationLifetime(host: { openMs, error }) + const { runtimeState: runtime } = host.context() + const owned = runtime.startFlights.ownsChild(sessionId, child) + if (owned) { + const retry = retryableStructuredAgentSessionStart(error) + runtime.startupAttempts.childFailed(sessionId, child, failure, retry) + } try { - await recordStructuredAgentSessionStartFailure( - { journal: session.journal, fence: child.fence }, - { - startKey: child.generation, - ...structuredAgentSessionStartFailure( - { failure }, - structuredAgentSessionFailureWordsContext(deps().store.getRecord(sessionId)) - ) - } - ).catch((writeError: unknown) => - deps().logger.warn('recording a failed provider open failed', { - scope: 'provider-open-failure', - sessionId, - error: writeError - }) - ) - await stopAgent(sessionId, { - cause: 'host-stop', - reason: error instanceof Error ? error.message : String(error), - failure - }) + if (!owned) { + const record = deps().store.getRecord(sessionId) + const wording = structuredAgentSessionFailureWordsContext(record) + await recordStructuredAgentSessionStartFailure( + { journal: session.journal, fence: child.fence }, + { + startKey: child.generation, + ...structuredAgentSessionStartFailure({ failure }, wording) + }, + oldestQueuedSubmission(session)?.clientMessageId ?? null + ).catch((writeError: unknown) => + deps().logger.warn('recording a failed provider open failed', { + scope: 'provider-open-failure', + sessionId, + error: writeError + }) + ) + } + const reason = error instanceof Error ? error.message : String(error) + await stopAgent(sessionId, { cause: 'host-stop', reason, failure }) } finally { - host.context().runtimeState.startupAttempts.childEnded(sessionId, child) + runtime.startupAttempts.childEnded(sessionId, child) } }), dispose: (): void => { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.ts index a94f3006e56..0ef9c163052 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.ts @@ -60,7 +60,7 @@ export type StructuredAgentSessionDeadGenerationInput = { exitFailure?: SubmissionRejectionFact failureTextContext?: AgentSessionFailureWordsContext /** Share generation-keyed startup failure rows with delivery-loop settlement. */ - exitedDuringStartup?: { generation: string | null } + exitedDuringStartup?: { generation: string | null; flightOwned?: boolean } /** Refine translator-unverifiable rows after observed exit proof. */ exit?: StructuredAgentSessionWatchedExit unrunRejection?: SubmissionRejectionFact @@ -104,6 +104,7 @@ export function planStructuredAgentSessionDeadGeneration( // Delivery owns failure rows for sends; other startup effects need their own row. // Existing rejection wording is terminal and must remain unchanged. const recordedByDeliveryLoop = + input.exitedDuringStartup.flightOwned || input.journal.submissions?.().some(isQueuedAgentJournalSubmission) || hasStructuredAgentSessionStartFailureRow(allItems, startKey) if (!recordedByDeliveryLoop) { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-delivery-loop-deps.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-delivery-loop-deps.ts new file mode 100644 index 00000000000..1f2b73cfd10 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-delivery-loop-deps.ts @@ -0,0 +1,50 @@ +import type { StructuredAgentSessionStartFlights } from './structured-agent-session-start-flight' +import type { StructuredAgentSessionDeliveryOrder } from './structured-agent-session-delivery-order' +import type { AgentSessionRecord } from '../../../shared/agent-session-record' +import type { AgentChildWorkView } from '../../../shared/agent-status-child-work-view' +import type { AgentSessionFailureWordsContext } from '../../../shared/agent-session-failure-words' +import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter' +import type { StructuredAgentRegistry } from './structured-agent-registry' +import type { StructuredAgentSessionLogger } from './structured-agent-session-logger' +import type { StructuredAgentSessionCurrentWork } from './structured-agent-session-current-work' +import type { StructuredAgentSessionRanRecorder } from './structured-agent-session-ran-options' +import type { StructuredAgentSessionAcquireAborts } from './structured-agent-session-acquire-aborts' +import type { StructuredAgentSessionStartClaim } from './structured-agent-session-agent-start' +import type { StructuredAgentSessionStartReadiness } from './structured-agent-session-startup-attempt' +import type { + StructuredAgentSessionHostSession, + StructuredAgentSessionProviderChildIdentity +} from './structured-agent-session-host-types' + +export type StructuredAgentSessionDeliveryLoopDeps = { + startFlights?: Pick + deliveryOrder: StructuredAgentSessionDeliveryOrder + sessions: ReadonlyMap + adapter: StructuredAgentSessionAdapter + agents: StructuredAgentRegistry + serialize: (sessionId: string, task: () => Promise) => Promise + /** Track from enqueue so quit drains any child the start may create. */ + trackStart: (start: Promise) => Promise + startProviderChild: (sessionId: string, startedFor: string) => StructuredAgentSessionStartClaim + /** Retire mismatched login/permission children under serialization. */ + retireOutgrownChild: (sessionId: string) => Promise + readiness: ( + sessionId: string, + child: StructuredAgentSessionProviderChildIdentity + ) => StructuredAgentSessionStartReadiness | null + conversationFence: (sessionId: string) => number + /** Settle sends covered by partial close; later sends remain eligible. False means settlement failed. */ + closeWhatTheUserClosed: (sessionId: string) => Promise + failureTextContext: (sessionId: string) => AgentSessionFailureWordsContext + logger: StructuredAgentSessionLogger + record: (sessionId: string) => AgentSessionRecord | null + currentWork: (sessionId: string) => StructuredAgentSessionCurrentWork | null + /** Publish row-free changes in current work. */ + workChanged?: (sessionId: string) => void + readChildWork: (sessionId: string) => readonly AgentChildWorkView[] | undefined + childWorkOpen: (sessionId: string) => boolean + stopping: (sessionId: string) => boolean + now: () => number + acquireAborts: Pick + recordRanOptions?: StructuredAgentSessionRanRecorder +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-delivery-loop.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-delivery-loop.ts index 3e5e40e4519..f762d4aa4d4 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-delivery-loop.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-delivery-loop.ts @@ -1,26 +1,15 @@ +import type { StructuredAgentSessionDeliveryLoopDeps } from './structured-agent-session-delivery-loop-deps' // Accept sends separately, then derive each delivery step from journal/child state. Wait outside // serialization for startup/readiness/provider calls so acceptance and Stop stay responsive. // Recover earlier-process queued sends before any handover; running commands block input. -import type { StructuredAgentSessionDeliveryOrder } from './structured-agent-session-delivery-order' -import type { AgentJournalCursor } from '../../../shared/agent-session-journal-types' -import type { AgentSessionRecord } from '../../../shared/agent-session-record' -import type { AgentChildWorkView } from '../../../shared/agent-status-child-work-view' import { agentSessionFailureFact } from '../../../shared/agent-session-failure' -import type { AgentSessionFailureWordsContext } from '../../../shared/agent-session-failure-words' -import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter' -import type { StructuredAgentRegistry } from './structured-agent-registry' import { structuredAgentSessionStartFailure } from './structured-agent-session-failure-text' import type { StructuredAgentSessionStartClaim } from './structured-agent-session-agent-start' -import type { - StructuredAgentSessionHostSession, - StructuredAgentSessionProviderChildIdentity -} from './structured-agent-session-host-types' +import type { StructuredAgentSessionProviderChildIdentity } from './structured-agent-session-host-types' import { - acceptedNoLaterThan, oldestQueuedSubmission, recordStructuredAgentSessionStartFailure, - refusedStructuredAgentSessionStart, startThatFailedWhileQueued } from './structured-agent-session-start-failure-row' import { @@ -29,54 +18,22 @@ import { } from './structured-agent-session-handover' import { structuredAgentSessionNextHandover } from './structured-agent-session-opening-send' import { structuredAgentSessionCommandRunning } from './structured-agent-session-command-turn' -import type { StructuredAgentSessionLogger } from './structured-agent-session-logger' import type { StructuredAgentSessionStartReadiness } from './structured-agent-session-startup-attempt' import { holdRestartedStructuredAgentSessionSends } from './structured-agent-session-host-lifetime' -import type { StructuredAgentSessionCurrentWork } from './structured-agent-session-current-work' import { structuredAgentSessionEndedChildFailure } from './structured-agent-session-ended-child-failure' import { waitsForCodexTurnOptions } from './structured-agent-session-message-options' -import type { StructuredAgentSessionRanRecorder } from './structured-agent-session-ran-options' import { sameProviderChild } from './structured-agent-session-provider-child' -import type { StructuredAgentSessionAcquireAborts } from './structured-agent-session-acquire-aborts' -export type StructuredAgentSessionDeliveryLoopDeps = { - deliveryOrder: StructuredAgentSessionDeliveryOrder - sessions: ReadonlyMap - adapter: StructuredAgentSessionAdapter - agents: StructuredAgentRegistry - serialize: (sessionId: string, task: () => Promise) => Promise - /** Track from enqueue so quit drains any child the start may create. */ - trackStart: (start: Promise) => Promise - startProviderChild: (sessionId: string, startedFor: string) => StructuredAgentSessionStartClaim - /** Retire mismatched login/permission children under serialization. */ - retireOutgrownChild: (sessionId: string) => Promise - readiness: ( - sessionId: string, - child: StructuredAgentSessionProviderChildIdentity - ) => StructuredAgentSessionStartReadiness | null - conversationFence: (sessionId: string) => number - /** Settle sends covered by partial close; later sends remain eligible. False means settlement failed. */ - closeWhatTheUserClosed: (sessionId: string) => Promise - failureTextContext: (sessionId: string) => AgentSessionFailureWordsContext - logger: StructuredAgentSessionLogger - record: (sessionId: string) => AgentSessionRecord | null - currentWork: (sessionId: string) => StructuredAgentSessionCurrentWork | null - /** Publish row-free changes in current work. */ - workChanged?: (sessionId: string) => void - readChildWork: (sessionId: string) => readonly AgentChildWorkView[] | undefined - childWorkOpen: (sessionId: string) => boolean - stopping: (sessionId: string) => boolean - now: () => number - acquireAborts: Pick - recordRanOptions?: StructuredAgentSessionRanRecorder -} +export type { StructuredAgentSessionDeliveryLoopDeps } from './structured-agent-session-delivery-loop-deps' type Prepared = | 'stop' - | { start: StructuredAgentSessionStartClaim; acceptedBy: AgentJournalCursor } + | 'continue' + | { start: StructuredAgentSessionStartClaim } | { ok: true awaited: StructuredAgentSessionProviderChildIdentity + clientMessageId: string readiness: StructuredAgentSessionStartReadiness | null before: Promise } @@ -130,22 +87,35 @@ export class StructuredAgentSessionDeliveryLoop { if (prepared === 'stop') { return } + if (prepared === 'continue') { + continue + } if ('start' in prepared) { - // Start outside serialization so sends remain acceptable. - const started = await prepared.start.outcome - if (started.ok || started.aborted) { - // Intentional cancellation settled its work; re-derive so later sends get fresh starts and quit stops the loop. - continue - } - const failure = refusedStructuredAgentSessionStart(this.deps.record(sessionId), started) - if (started.refusal.details?.reason !== 'previousExitUnverifiable') { - await this.deps.serialize(sessionId, () => this.fail(sessionId, failure)) + // Outside the serialize: sends are accepted while the agent starts. + const completed = await prepared.start.outcome.then( + () => true, + (error: unknown) => { + this.deps.logger.warn('waiting for an agent start failed', { + scope: 'delivery-loop-start', + sessionId, + error + }) + return false + } + ) + if (!completed) { + await this.deps + .serialize(sessionId, async () => this.stop(sessionId)) + .catch((error: unknown) => + this.deps.logger.warn('parking delivery after a failed start failed', { + scope: 'delivery-loop-start', + sessionId, + error + }) + ) return } - // Unproven close applies only to work captured for this start; later sends may retry proof. - await this.deps.serialize(sessionId, () => - this.recordFailure(sessionId, failure, prepared.acceptedBy) - ) + // The flight settled its requester; joiners rederive their own work. continue } if (prepared.readiness) { @@ -153,7 +123,12 @@ export class StructuredAgentSessionDeliveryLoop { } await prepared.before const handed = await this.deps.serialize(sessionId, () => - this.handOver(sessionId, prepared.awaited, prepared.readiness !== null) + this.handOver( + sessionId, + prepared.awaited, + prepared.clientMessageId, + prepared.readiness !== null + ) ) if (handed === 'stop') { return @@ -176,10 +151,17 @@ export class StructuredAgentSessionDeliveryLoop { }) const cause = { hostFault: true } as const await this.deps - .serialize(sessionId, () => this.fail(sessionId, { startKey: null, cause })) + .serialize(sessionId, async () => { + try { + const session = this.deps.sessions.get(sessionId) + const selected = session && oldestQueuedSubmission(session) + await this.fail(sessionId, { startKey: null, cause }, selected?.clientMessageId ?? null) + } finally { + this.stop(sessionId) + } + }) .catch((failure: unknown) => { // Keep failed-settlement leftovers for the next accept/open recovery. - this.running.delete(sessionId) this.deps.logger.warn('recording a failed delivery failed', { scope: 'delivery-loop-fail', sessionId, @@ -210,21 +192,24 @@ export class StructuredAgentSessionDeliveryLoop { if (!oldest || (session.child && work && structuredAgentSessionCommandRunning(work))) { return this.stop(sessionId) } + if (this.deps.startFlights?.has(sessionId)) { + return { start: this.deps.startProviderChild(sessionId, oldest.clientMessageId) } + } const failedStart = startThatFailedWhileQueued(session, oldest) if (failedStart) { - return this.fail(sessionId, failedStart) + return this.fail(sessionId, failedStart, oldest.clientMessageId) } await this.deps.retireOutgrownChild(sessionId) const child = this.deps.sessions.get(sessionId)?.child if (!child || child.close) { return { - start: this.deps.startProviderChild(sessionId, oldest.clientMessageId), - acceptedBy: session.journal.cursor() + start: this.deps.startProviderChild(sessionId, oldest.clientMessageId) } } return { ok: true, awaited: { generation: child.generation, fence: child.fence }, + clientMessageId: oldest.clientMessageId, readiness: child.phase === 'starting' ? this.deps.readiness(sessionId, child) : null, before: this.deps.deliveryOrder.beforeSend(sessionId, oldest.clientMessageId) } @@ -233,6 +218,7 @@ export class StructuredAgentSessionDeliveryLoop { private async handOver( sessionId: string, awaited: StructuredAgentSessionProviderChildIdentity, + clientMessageId: string, waited: boolean ): Promise<'continue' | 'stop' | StructuredAgentSessionPendingHandover> { const session = this.deps.sessions.get(sessionId) @@ -249,12 +235,16 @@ export class StructuredAgentSessionDeliveryLoop { if (endedFailure === null) { return 'continue' } - return this.fail(sessionId, { - startKey: awaited.generation, - cause: endedFailure ?? - // Unknown disappearance is not an observed provider exit. - { failure: agentSessionFailureFact('startFailed') } - }) + return this.fail( + sessionId, + { + startKey: awaited.generation, + cause: endedFailure ?? + // Gone with no end observed: nothing says the provider stopped. + { failure: agentSessionFailureFact('startFailed') } + }, + clientMessageId + ) } // Stop may take the lane during readiness; recheck before steering into work it is ending. if (this.deps.stopping(sessionId) || awaitedChild.phase !== 'ready') { @@ -303,16 +293,11 @@ export class StructuredAgentSessionDeliveryLoop { return pending ?? 'continue' } - private async fail(sessionId: string, failure: StartFailure): Promise<'stop'> { - await this.recordFailure(sessionId, failure) - return this.stop(sessionId) - } - - private async recordFailure( + private async fail( sessionId: string, failure: StartFailure, - acceptedBy?: AgentJournalCursor - ): Promise { + clientMessageId: string | null + ): Promise<'continue'> { const session = this.deps.sessions.get(sessionId) if (session) { await recordStructuredAgentSessionStartFailure( @@ -324,9 +309,10 @@ export class StructuredAgentSessionDeliveryLoop { this.deps.failureTextContext(sessionId) ) }, - acceptedBy && acceptedNoLaterThan(session.journal, acceptedBy) + clientMessageId ) } + return 'continue' } /** Clear under serialization so the next accept can wake a new loop. */ diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-event-recovery.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-event-recovery.ts index 7403c75594a..f5c799f7bd4 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-event-recovery.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-event-recovery.ts @@ -75,8 +75,7 @@ export class StructuredAgentSessionEventRecovery { runtimeState, acknowledgeRelease: (sessionId) => deps.adapter.acknowledgeSessionRelease?.(sessionId) }, - abortOpen: (sessionId) => - runtimeState.acquireAborts.abort(sessionId, 'provider exited while opening') + abortOpen: (sessionId) => runtimeState.acquireAborts.abortOpen(sessionId) } } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-delivery.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-delivery.ts index 99c0669040a..02dfbeb9184 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-delivery.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-delivery.ts @@ -12,7 +12,10 @@ import { } from './structured-agent-session-conversation-open' import type { StructuredAgentSessionClientDelivery } from './structured-agent-session-client-delivery' import type { StructuredAgentSessionDeliveryOrder } from './structured-agent-session-delivery-order' -import { StructuredAgentSessionDeliveryLoop } from './structured-agent-session-delivery-loop' +import { + StructuredAgentSessionDeliveryLoop, + type StructuredAgentSessionDeliveryLoopDeps +} from './structured-agent-session-delivery-loop' import { structuredAgentSessionCommandRunning } from './structured-agent-session-command-turn' import type { StructuredAgentSessionStartClaim } from './structured-agent-session-agent-start' import type { StructuredAgentSessionStartReadiness } from './structured-agent-session-startup-attempt' @@ -55,6 +58,7 @@ export function createStructuredAgentSessionConversationDelivery(input: { trackStart: (start: Promise) => Promise startProviderChild: (sessionId: string, startedFor: string) => StructuredAgentSessionStartClaim deliveryOrder: StructuredAgentSessionDeliveryOrder + startFlights?: StructuredAgentSessionDeliveryLoopDeps['startFlights'] readiness: ( sessionId: string, child: StructuredAgentSessionProviderChildIdentity @@ -105,6 +109,7 @@ export function createStructuredAgentSessionConversationDelivery(input: { }) }, readiness: input.readiness, + startFlights: input.startFlights, deliveryOrder: input.deliveryOrder, conversationFence: (sessionId) => structuredAgentSessionConversationFence(deps.store, sessionId), diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-status-feed.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-status-feed.ts index 7dd0cb223c0..4ec556c74d9 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-status-feed.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-status-feed.ts @@ -18,6 +18,7 @@ import type { StructuredAgentSessionConversationNaming } from './structured-agen /** Resolve deps lazily because field initializers run before host constructor assignments. */ export function createStructuredAgentSessionHostStatusFeed(args: { sessions: StructuredAgentSessionStatusFeedDeps['sessions'] + starting?: (sessionId: string) => boolean now: () => number currentWork: StructuredAgentSessionSubagentRowsDeps['currentWork'] deps: () => { @@ -56,6 +57,7 @@ export function createStructuredAgentSessionHostStatusFeed(args: { }) const feed = new StructuredAgentSessionStatusFeed({ sessions: args.sessions, + starting: args.starting, getRecord: (sessionId) => args.deps().store.getRecord(sessionId), replacedRuntime: (sessionId) => args.deps().store.replacedRuntime(sessionId), now: args.now, 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 445b0df0064..abf5e003a87 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 @@ -28,10 +28,7 @@ import { StructuredAgentSessionEventRecovery } from './structured-agent-session- import { StructuredAgentSessionBackgroundTaskChannel } from './structured-agent-session-background-task-channel' import { StructuredAgentSessionClientDelivery } from './structured-agent-session-client-delivery' import { StructuredAgentSessionConversations } from './structured-agent-session-conversations' -import { - createStructuredAgentSessionRestartResume, - type StructuredAgentSessionRestartResume -} from './structured-agent-session-restart-resume-host' +import * as resume from './structured-agent-session-restart-resume-host' import { structuredAgentSessionRestartResumeSurfaces } from './structured-agent-session-restart-resume-wiring' import * as conversation from './structured-agent-session-host-delivery' import { wireStructuredAgentSessionQueuedMessages } from './structured-agent-session-queued-wiring' @@ -67,7 +64,8 @@ export class StructuredAgentSessionHost { this.queued, (sessionId) => this.restartResume.onAgentStarted(sessionId), (sessionId) => this.backgroundTasks.publish(sessionId), - (sessionId) => this.backgroundTasks.read(sessionId) + (sessionId) => this.backgroundTasks.read(sessionId), + (sessionId) => this.runtimeState?.startFlights.projectsStarting(sessionId) ?? false ) private readonly subscribers = this.clientDelivery.subscribers private readonly tasks = new StructuredAgentSessionTaskQueue() @@ -78,7 +76,7 @@ export class StructuredAgentSessionHost { private readonly conversationDelivery: conversation.StructuredAgentSessionConversationDelivery private readonly eventRecovery: StructuredAgentSessionEventRecovery private readonly backgroundTasks: StructuredAgentSessionBackgroundTaskChannel - readonly restartResume: StructuredAgentSessionRestartResume + readonly restartResume: resume.StructuredAgentSessionRestartResume constructor(readonly deps: HostTypes.StructuredAgentSessionHostDeps) { // Collaborators share a logger wrapper that cannot fail the operation it reports. @@ -114,6 +112,7 @@ export class StructuredAgentSessionHost { startProviderChild: (sessionId, startedFor) => agentStart.claimStructuredAgentSessionAgent(this.attachContext(), sessionId, startedFor), deliveryOrder: this.runtimeState.deliveryOrder, + startFlights: this.runtimeState.startFlights, readiness: (id, child) => this.runtimeState.startupAttempts.readiness(id, child), // Retire at rest; the next start refreshes login and permission mode. stopOutgrownAgent: (id) => this.lifetime.stopAgent(id, { cause: 'evict', resting: true }), @@ -144,7 +143,7 @@ export class StructuredAgentSessionHost { failStartingChild: (...args) => this.lifetime.failStartingChild(...args), trackOpen: (open) => this.tasks.trackAttach(open) }) - this.restartResume = createStructuredAgentSessionRestartResume(deps, this.sessions, { + this.restartResume = resume.createStructuredAgentSessionRestartResume(deps, this.sessions, { ...structuredAgentSessionRestartResumeSurfaces(this, this.now), readChildWork: this.clientDelivery.readChildWork, childWorkOpen: this.clientDelivery.childWorkOpen, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-operation-child.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-operation-child.ts index f8eca7c818d..1a0cf8175b6 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-operation-child.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-operation-child.ts @@ -50,7 +50,7 @@ export function selectStructuredAgentSessionOperationChild( readiness: StructuredAgentSessionStartReadiness | null } { const child = context.sessions.get(sessionId)?.child - if (!child || child.close) { + if (context.startFlights.get(sessionId) || !child || child.close) { return { ok: true, child: null, readiness: null } } const identity = { generation: child.generation, fence: child.fence } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-operation-start-refusal.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-operation-start-refusal.test.ts index 89fdf5aa887..4cf8446f76e 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-operation-start-refusal.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-operation-start-refusal.test.ts @@ -3,6 +3,7 @@ import type { AgentSessionRecord } from '../../../shared/agent-session-record' import type { StructuredAgentSessionAttachContext } from './structured-agent-session-attach-context' import { startStructuredAgentSessionAgentForOperation } from './structured-agent-session-agent-start' import { StructuredAgentSessionAcquireAborts } from './structured-agent-session-acquire-aborts' +import { StructuredAgentSessionStartupAttempts } from './structured-agent-session-startup-attempt' import { StructuredAgentSessionStartFlights } from './structured-agent-session-start-flight' import { recordStructuredAgentSessionOptionIntent } from './structured-agent-session-options-read' import { claudeAndCodexAgents } from './structured-agent-session-adapter-router-test-support' @@ -17,12 +18,13 @@ describe('an operation whose agent start throws', () => { const context = { deps: { logger: log.logger }, sessions: new Map(), + publishStatus: vi.fn(), serialize: (_sessionId: string, task: () => Promise) => task(), tasks: { trackAttach: (start: Promise) => start }, runtimeState: { acquireAborts: new StructuredAgentSessionAcquireAborts(), startFlights: new StructuredAgentSessionStartFlights(), - startupAttempts: { begin: () => {}, abandon: () => {} } + startupAttempts: new StructuredAgentSessionStartupAttempts({ expire: () => {} }) }, reconcileLeases: () => Promise.reject(cause) } @@ -35,13 +37,14 @@ describe('an operation whose agent start throws', () => { expect(refused).toEqual({ ok: false, + failure: { kind: 'hostFault' }, refusal: { code: 'agent_session_owner_restart_failed', message: "The agent couldn't restart." } }) expect(log.entries.map((entry) => entry.fields)).toEqual([ - { scope: 'operation-agent-start', sessionId: SESSION, error: cause } + { scope: 'agent-start', sessionId: SESSION, error: cause } ]) }) }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-provider-open.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-provider-open.ts index 74a0875f0ad..a0503d17be1 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-provider-open.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-provider-open.ts @@ -21,7 +21,7 @@ export async function openStructuredAgentSessionProviderChild( if (!context.runtimeState.startupAttempts.onClock(sessionId, child)) { return } - const attempt = context.runtimeState.acquireAborts.begin(sessionId) + const attempt = context.runtimeState.acquireAborts.begin(sessionId, false, false, true) const openedAt = context.now() try { attempt.signal.throwIfAborted() diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-ready-operation.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-ready-operation.ts index 1e36d5c0f11..4f5c3e39d72 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-ready-operation.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-ready-operation.ts @@ -254,8 +254,11 @@ export async function runWithReadyStructuredAgentSessionOperation( try { let { child, readiness } = selected if (!child) { + const start = selected.start?.outcome ?? Promise.resolve(undefined) + const ready = start.then(() => 'ready' as const) + const elapsed = waitForStructuredAgentSessionStart(ready, deadline - Date.now()) const started = await waitForPromiseWithSignal( - selected.start?.outcome ?? Promise.resolve(undefined), + elapsed.then((ready) => (ready === 'ready' ? start : undefined)), authority?.signal ).catch((error: unknown) => { if (authority?.signal.aborted) { @@ -264,7 +267,7 @@ export async function runWithReadyStructuredAgentSessionOperation( throw error }) if (!started?.ok) { - return started ?? operationDidNotRun() + return started ?? operationDidNotRun(!authority?.signal.aborted && Date.now() >= deadline) } const selection = selectStructuredAgentSessionOperationChild(context, envelope.sessionId) if (!selection.ok || !selection.child) { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-send-restarts-failed-start.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-send-restarts-failed-start.test.ts index 46cf3da6097..24882250113 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-send-restarts-failed-start.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-send-restarts-failed-start.test.ts @@ -199,13 +199,21 @@ describe('a send into a published session whose child ended before startup', () expect(dispatch).toHaveBeenCalledTimes(2) }) - it('retires the held message with the cause when the restarted child exits before proving its start', async () => { + it('retires the held message with the cause after five restarted children exit before proving their starts', async () => { const releasedFence = store.getRecord(SESSION)?.lease.runtimeFence ?? 0 const rowsBefore = (await journalStatuses()).length const held = await send('still not signed in', releasedFence) await eventually(() => expect(acquire).toHaveBeenCalledTimes(2)) - await exitBeforeProof() + for (let attempt = 2; attempt <= 6; attempt++) { + await eventually(() => expect(acquire).toHaveBeenCalledTimes(attempt)) + await eventually(() => + expect(host.collaboratorsForTests().sessions.get(SESSION)?.child?.phase).toBe('starting') + ) + expect((await journalStatuses()).length).toBe(rowsBefore) + expect((await submission(held))?.dispatchState).toBe('pending') + await exitBeforeProof() + } await eventually(async () => expect((await submission(held))?.dispatchState).toBe('rejected')) expect((await submission(held))?.handedOverAt).toBeUndefined() @@ -231,12 +239,12 @@ describe('a send into a published session whose child ended before startup', () ]).filter((id) => shown.includes(id)) ).toEqual(shown) // Acquisition and proven exit each advance the fence. - expect(store.getRecord(SESSION)?.lease.runtimeFence).toBe(releasedFence + 2) + expect(store.getRecord(SESSION)?.lease.runtimeFence).toBe(releasedFence + 10) expect(store.getRecord(SESSION)?.lease.claimStatus).toBe('released') - expect(acquire).toHaveBeenCalledTimes(2) + expect(acquire).toHaveBeenCalledTimes(6) const retry = await send('signed in now') - await eventually(() => expect(acquire).toHaveBeenCalledTimes(3)) + await eventually(() => expect(acquire).toHaveBeenCalledTimes(7)) await proveStarted() await handedOver(retry) expect(dispatch).toHaveBeenCalledOnce() diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-start-admission.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-start-admission.test.ts index f937ffc45f6..98be125ee57 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-start-admission.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-start-admission.test.ts @@ -428,10 +428,12 @@ describe('a start racing the host', () => { expect(replayed).toBe(false) expect(child()).toBeNull() release() - expect(await replay).toMatchObject({ ok: true, replayed: true }) + await eventually(() => expect(child()?.phase).toBe('starting')) + expect(replayed).toBe(false) expect(child()).toMatchObject({ phase: 'starting', generation: `generation-${spawned + 1}` }) expect(operationRan).toBe(false) opened.resolve() + expect(await replay).toMatchObject({ ok: true, replayed: true }) expect(await operation).toMatchObject({ ok: true }) expect(operationRan).toBe(true) await eventually(async () => diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-start-attempts.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-start-attempts.ts new file mode 100644 index 00000000000..dda212ad8e5 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-start-attempts.ts @@ -0,0 +1,206 @@ +import { randomUUID } from 'node:crypto' +import { setTimeout as delay } from 'node:timers/promises' +import { throwIfSignalAborted, waitForPromiseWithSignal } from '../../../shared/abort-signal-reason' +import { refuseUnclassified } from '../../../shared/agent-session-wire-refusals' +import { agentSessionWriteNoticeEnglish } from '../../../shared/agent-session-refusal-notice' +import { isQueuedAgentJournalSubmission } from '../../../shared/agent-session-queued-submission' +import { agentSessionFailureFact } from '../../../shared/agent-session-failure' +import type { StructuredAgentSessionAttachContext } from './structured-agent-session-attach-context' +import type { StructuredAgentSessionAttachLock } from './structured-agent-session-attach-lock' +import type { StructuredAgentSessionResumeOutcome } from './structured-agent-session-agent-start' +import type { StructuredAgentSessionHostSession } from './structured-agent-session-host-types' +import { structuredAgentSessionStartFlightLock } from './structured-agent-session-start-flight' +import { isStructuredAgentSessionStartupExpired } from './structured-agent-session-startup-attempt-contract' +import { refusedStructuredAgentSessionStart } from './structured-agent-session-start-eligibility' +import { + refusedStructuredAgentSessionStart as failedStart, + recordStructuredAgentSessionStartFailure +} from './structured-agent-session-start-failure-row' +import { + structuredAgentSessionStartFailure, + structuredAgentSessionStartFailureFact +} from './structured-agent-session-failure-text' +import { structuredAgentSessionFailureWordsContext } from './structured-agent-session-send-preparation' +import { sameProviderChild } from './structured-agent-session-provider-child' + +export function registerStructuredAgentSessionStart( + context: StructuredAgentSessionAttachContext, + sessionId: string, + startedFor: string | undefined, + run: ( + lock: StructuredAgentSessionAttachLock, + attemptId: string + ) => Promise +): Promise { + const { startFlights: flights, startupAttempts: attempts, acquireAborts } = context.runtimeState + const acquire = acquireAborts.begin(sessionId) + let attemptId = randomUUID() + let handle = attempts.begin(sessionId, attemptId) + const registeredAt = handle.registeredAt + const children: NonNullable>['children'] = [] + const register = () => + flights.register(sessionId, { + attemptId, + outcome, + abort: acquire.abort, + signal: acquire.signal, + children + }) + const outcome = context.tasks.trackAttach( + Promise.resolve() + .then(async (): Promise => { + let result: StructuredAgentSessionResumeOutcome = failedStructuredAgentSessionStart() + for (let number = 0; number < 5; number++) { + throwIfSignalAborted(acquire.signal) + if (attempts.pastCeiling(registeredAt)) { + result = { ...failedStructuredAgentSessionStart(), expired: true } + break + } + const lock = structuredAgentSessionStartFlightLock({ + sessionId, + attemptId, + signal: acquire.signal, + flights, + serialize: context.serialize + }) + const stillQueued = await lock.commit(async () => + structuredAgentSessionStartStillQueued(context.sessions.get(sessionId), startedFor) + ) + if (!stillQueued) { + return { ...failedStructuredAgentSessionStart(), aborted: true } + } + result = await run(lock, attemptId) + if (result.ok) { + const ready = await lock.wait(handle.ended) + throwIfSignalAborted(acquire.signal) + if (ready === 'ready') { + return result + } + result = await lock.commit(async () => { + const session = context.sessions.get(sessionId) + const ended = session?.lastEndedChild + const child = handle.result.child + if ( + !child || + !ended || + !sameProviderChild(ended, child) || + !ended.rootGone || + (session?.child && sameProviderChild(session.child, child)) + ) { + return refusedStructuredAgentSessionStart( + 'agent_session_ownership_unknown', + { reason: 'ownerUnproven' }, + 'The previous agent process could not be verified as stopped.' + ) + } + const failure = structuredAgentSessionStartFailureFact({ + exit: handle.result.failure ?? ended.failure + }) + return { + ...failedStructuredAgentSessionStart(), + startKey: ended.generation, + failure, + retry: + handle.result.retry ?? + (ended.cause === 'exit' && failure.kind === 'providerStartFailed') + } + }) + } + if (result.ok || result.aborted || result.expired || !result.retry || number === 4) { + break + } + attempts.abandon(sessionId, attemptId) + await waitForPromiseWithSignal(delay(100 * 2 ** number), acquire.signal) + throwIfSignalAborted(acquire.signal) + attemptId = randomUUID() + handle = attempts.begin(sessionId, attemptId, registeredAt) + register() + } + return result + }) + .catch((error: unknown): StructuredAgentSessionResumeOutcome => { + if (!acquire.signal.aborted) { + context.deps.logger.warn('starting the agent failed', { + scope: 'agent-start', + sessionId, + error + }) + return { + ...failedStructuredAgentSessionStart(), + failure: agentSessionFailureFact('hostFault') + } + } + return failedStructuredAgentSessionStart() + }) + .then(async (result): Promise => { + if (result.ok) { + return result + } + return context.serialize(sessionId, async () => { + const settled = acquire.signal.aborted + ? { + ...failedStructuredAgentSessionStart(), + ...(isStructuredAgentSessionStartupExpired(acquire.signal.reason) + ? { expired: true as const } + : { aborted: true as const }) + } + : result + if (!settled.aborted && (startedFor !== undefined || children.length > 0)) { + const session = context.sessions.get(sessionId) + if (session) { + const record = context.deps.store.getRecord(sessionId) + const failure = failedStart(record, settled) + await recordStructuredAgentSessionStartFailure( + { journal: session.journal, fence: record?.lease.runtimeFence ?? 0 }, + { + startKey: + children.at(-1)?.generation ?? failure.startKey ?? startedFor ?? attemptId, + ...structuredAgentSessionStartFailure( + failure.cause, + structuredAgentSessionFailureWordsContext(record) + ) + }, + startedFor ?? null + ) + } + } + return settled + }) + }) + .finally(() => { + acquire.end() + flights.end(sessionId, attemptId) + attempts.abandon(sessionId, attemptId) + context.publishStatus(sessionId) + }) + ) + register() + return outcome +} + +export function failedStructuredAgentSessionStart(): Extract< + StructuredAgentSessionResumeOutcome, + { ok: false } +> { + return { + ok: false, + refusal: refuseUnclassified( + 'agent_session_owner_restart_failed', + agentSessionWriteNoticeEnglish(['restartFailed']) + ) + } +} + +export function structuredAgentSessionStartStillQueued( + session: Pick | undefined, + startedFor: string | undefined +): boolean { + return ( + startedFor === undefined || + session?.journal + .submissions() + .some( + (entry) => entry.clientMessageId === startedFor && isQueuedAgentJournalSubmission(entry) + ) === true + ) +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-start-failure-row.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-start-failure-row.ts index 2d168fc3978..1f0b7f9c43b 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-start-failure-row.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-start-failure-row.ts @@ -18,6 +18,7 @@ import type { import { failedProviderChildStart } from './structured-agent-session-provider-child' import { structuredAgentSessionEndedChildFailure } from './structured-agent-session-ended-child-failure' +/** A failed start applies only to the queued message that requested it. */ export function startThatFailedWhileQueued( session: StructuredAgentSessionHostSession, oldest: AgentJournalSubmission @@ -25,6 +26,7 @@ export function startThatFailedWhileQueued( const ended = failedProviderChildStart(session) if ( !ended || + ended.startedFor !== oldest.clientMessageId || oldest.acceptedSequence === undefined || ended.endedAt.epoch !== session.journal.cursor().epoch || ended.endedAt.sequence < oldest.acceptedSequence @@ -41,14 +43,19 @@ export function refusedStructuredAgentSessionStart( refusal, diagnostic, argumentProblem, - expired + expired, + failure, + startKey = null }: Extract -): { startKey: null; cause: StructuredAgentSessionStartFailureCause } { +): { startKey: string | null; cause: StructuredAgentSessionStartFailureCause } { if (expired) { - return { startKey: null, cause: { failure: agentSessionFailureFact('hostStopped') } } + return { startKey, cause: { failure: agentSessionFailureFact('hostStopped') } } + } + if (failure) { + return { startKey, cause: { failure } } } return { - startKey: null, + startKey, cause: { refusal, ...(diagnostic ? { diagnostic } : {}), @@ -84,22 +91,22 @@ export function hasStructuredAgentSessionStartFailureRow( return items.some((item) => item.itemId === itemId) } -/** Reject affected queued messages before the failure row; owned-open failure may already have recorded cause. */ +/** + * A failed start settles its selected message, or only its row when no message requested it. + */ export async function recordStructuredAgentSessionStartFailure( session: Pick & { fence: number }, failure: StructuredAgentSessionStartFailure, - /** Exclude messages accepted after startup began. */ - which?: (submission: AgentJournalSubmission) => boolean + clientMessageId: string | null ): Promise { - const oldest = oldestQueuedSubmission(session) - const startKey = failure.startKey ?? oldest?.clientMessageId + const selected = session.journal + .submissions() + .find( + (entry) => entry.clientMessageId === clientMessageId && isQueuedAgentJournalSubmission(entry) + ) + const startKey = failure.startKey ?? clientMessageId const rejection = { reason: failure.reason, rejection: failure.rejection } - const chosen = which - ? session.journal - .submissions() - .filter((entry) => isQueuedAgentJournalSubmission(entry) && which(entry)) - : null - if (startKey === undefined || chosen?.length === 0) { + if (startKey === null || (clientMessageId !== null && !selected)) { return } await session.journal.appendLifecycleBatch({ @@ -107,19 +114,19 @@ export async function recordStructuredAgentSessionStartFailure( fence: session.fence, recovered: true, mutations: [structuredAgentSessionStartFailureRow(startKey, failure)], - ...(chosen + ...(selected ? { - dispatches: chosen.map(({ clientMessageId }) => ({ - clientMessageId, - fence: session.fence, - state: 'rejected' as const, - recovered: true as const, - ...rejection - })) + dispatches: [ + { + clientMessageId: selected.clientMessageId, + fence: session.fence, + state: 'rejected' as const, + recovered: true as const, + ...rejection + } + ] } - : oldest - ? { rejectsQueued: rejection } - : {}) + : {}) }) } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-start-failure-writer.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-start-failure-writer.test.ts index 1a591ddef09..44659fdff72 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-start-failure-writer.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-start-failure-writer.test.ts @@ -29,7 +29,7 @@ const EXIT_REASON = 'Claude Code is not signed in. Sign in with the Claude CLI' const EXIT_TEXT = 'Codex stopped before it finished starting. Send your message to try again.' const START_ROW = agentJournalItemKey({ provider: 'orca', - clientMessageId: 'start-failure:generation-2' + clientMessageId: 'start-failure:generation-6' }) function eventually(assertion: () => void | Promise): Promise { @@ -172,7 +172,15 @@ describe('a start that fails before it proves itself', () => { expect(host.collaboratorsForTests().sessions.get(SESSION)?.child?.phase).toBe('starting') ) - await exitBeforeProof() + for (let attempt = 2; attempt <= 6; attempt++) { + await eventually(() => expect(generation).toBe(attempt)) + await eventually(() => + expect(host.collaboratorsForTests().sessions.get(SESSION)?.child?.phase).toBe('starting') + ) + expect(await startRows()).toEqual([]) + expect(await submission(sent)).toMatchObject({ dispatchState: 'pending' }) + await exitBeforeProof() + } await eventually(async () => expect(await submission(sent)).toMatchObject({ dispatchState: 'rejected', reason: EXIT_TEXT }) ) @@ -180,10 +188,11 @@ describe('a start that fails before it proves itself', () => { expect(await startRows()).toEqual([EXIT_TEXT]) expect(publishedStartRows()).toEqual([EXIT_TEXT]) + expect(generation).toBe(6) expect(dispatch).not.toHaveBeenCalled() }) - it('leaves the row to the loop when the exit lands while the message still waits', async () => { + it('writes one final row when all five attempts exit while the message waits', async () => { exitOnStart = true const queued = await send('hello') @@ -192,6 +201,7 @@ describe('a start that fails before it proves itself', () => { ) await host.flushStreamedEvents(SESSION) + expect(generation).toBe(6) expect(dispatch).not.toHaveBeenCalled() expect(await startRows()).toEqual([EXIT_TEXT]) expect(publishedStartRows()).toEqual([EXIT_TEXT]) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-start-flight.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-start-flight.test.ts new file mode 100644 index 00000000000..a988a66c212 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-start-flight.test.ts @@ -0,0 +1,28 @@ +import { describe, expect, it } from 'vitest' +import { StructuredAgentSessionStartFlights } from './structured-agent-session-start-flight' + +describe('start flight child retention', () => { + it.each([10, 100])('retires a flight after %i duplicate publications', (count) => { + const flights = new StructuredAgentSessionStartFlights() + const flight = { + attemptId: 'start', + abort: () => {}, + outcome: Promise.resolve({ ok: true as const }) + } + flights.register('chat', flight) + const child = { generation: 'child', fence: 1 } + for (let n = 0; n < count; n++) { + flights.published('chat', { ...child }) + } + expect(flights.get('chat')?.children).toEqual([child]) + expect(flights.ownsChild('chat', child)).toBe(true) + const successor = { ...child, fence: 2 } + flights.published('chat', successor) + flights.published('chat', { ...successor }) + expect(flights.get('chat')?.children).toEqual([child, successor]) + flights.end('chat', flight.attemptId) + flights.published('chat', child) + expect(flights.has('chat')).toBe(false) + expect(flights.ownsChild('chat', child)).toBe(false) + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-start-flight.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-start-flight.ts index 2a66d4298d5..5b98a6ed014 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-start-flight.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-start-flight.ts @@ -4,11 +4,16 @@ import { throwIfSignalAborted, waitForPromiseWithSignal } from '../../../shared/abort-signal-reason' import type { StructuredAgentSessionAttachLock } from './structured-agent-session-attach-lock' import type { StructuredAgentSessionEndedEvent } from './structured-agent-session-adapter' +import { sameProviderChild } from './structured-agent-session-provider-child' +import type { StructuredAgentSessionProviderChildIdentity } from './structured-agent-session-host-types' import type { StructuredAgentSessionResumeOutcome } from './structured-agent-session-agent-start' export type StructuredAgentSessionStartFlight = { readonly abort: () => void readonly attemptId: string + signal?: AbortSignal + children?: StructuredAgentSessionProviderChildIdentity[] + /** Settles once the logical start is ready, refused, or aborted. */ readonly outcome: Promise } @@ -33,6 +38,22 @@ export class StructuredAgentSessionStartFlights { this.flying.set(sessionId, flight) } + published(sessionId: string, child: StructuredAgentSessionProviderChildIdentity): void { + const flight = this.get(sessionId) + if (flight && !this.ownsChild(sessionId, child)) { + ;(flight.children ??= []).push({ generation: child.generation, fence: child.fence }) + } + } + + ownsChild(sessionId: string, child: StructuredAgentSessionProviderChildIdentity): boolean { + return this.get(sessionId)?.children?.some((owned) => sameProviderChild(owned, child)) === true + } + + projectsStarting(sessionId: string): boolean { + const flight = this.get(sessionId) + return !!flight?.children?.length && !flight.signal?.aborted + } + end(sessionId: string, attemptId: string): void { if (this.isCurrent(sessionId, attemptId)) { this.flying.delete(sessionId) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-start-retry-recorded.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-start-retry-recorded.test.ts new file mode 100644 index 00000000000..2a7654fa03f --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-start-retry-recorded.test.ts @@ -0,0 +1,135 @@ +import { PassThrough } from 'node:stream' +import { describe, expect, it, vi } from 'vitest' +import { AcpAgentError } from '../../acp/acp-errors' +import { acpRpcEnvelopeSchema, settleAcpResponse } from '../../acp/acp-rpc-envelope' +import { + ClaudeControlRequestError, + runClaudeControl +} from '../../claude/claude-agent-sdk-control-requests' +import { createCodexAppServerRecordDispatcher } from '../../codex/codex-app-server-record-dispatch' +import { CodexAppServerRequestError } from '../../codex/app-server/codex-app-server-request-error' +import { JsonlRpcPeer, JsonlRpcResponseError } from '../../jsonl-rpc/peer' +import { withProviderRequestTimeout } from '../../provider-process/provider-request-timeout' +import { + AgentSessionAcquisitionExitProvenError, + AgentSessionAcquisitionExitUnprovenError, + AgentSessionAcquisitionRootExitObservedError +} from './structured-agent-session-acquisition-errors' +import { AgentSessionAcquisitionRefusal } from './structured-agent-session-adapter-errors' +import { + retryableStructuredAgentSessionAcquisition, + retryableStructuredAgentSessionStart +} from './structured-agent-session-start-retry' +import captures from './__fixtures__/startup-protocol-rejections.json' + +function expectExitEvidenceRequired(error: unknown) { + expect(retryableStructuredAgentSessionStart(error)).toBe(true) + expect(retryableStructuredAgentSessionAcquisition(error)).toBe(false) + expect( + retryableStructuredAgentSessionAcquisition(new AgentSessionAcquisitionExitProvenError(error)) + ).toBe(true) + expect( + retryableStructuredAgentSessionAcquisition( + new AgentSessionAcquisitionRootExitObservedError(error) + ) + ).toBe(true) + expect( + retryableStructuredAgentSessionAcquisition(new AgentSessionAcquisitionExitUnprovenError(error)) + ).toBe(false) +} + +describe('recorded command rejection wrappers at the startup classifier', () => { + it('wraps Claude rejection text at the SDK callback boundary', async () => { + const capture = captures.claude + // SDK rejection is supplied here; this does not exercise the SDK's wire parser. + const error = await runClaudeControl(capture.request.request.subtype, () => + Promise.reject(new Error(capture.response.response.error)) + ).catch((caught: unknown) => caught) + expect(error).toBeInstanceOf(ClaudeControlRequestError) + expect(error).toMatchObject({ + subtype: capture.request.request.subtype, + message: capture.response.response.error + }) + expectExitEvidenceRequired(error) + }) + + it('dispatches the recorded Codex reply with its method, code and provider diagnostic', async () => { + const capture = captures.codex + const pending = Promise.withResolvers() + const failed = pending.promise.catch((caught: unknown) => caught) + const dispatcher = createCodexAppServerRecordDispatcher({ + handlers: {}, + writeResponse: vi.fn(), + onProtocolFailure: vi.fn() + }) + dispatcher.addPending(capture.request.id, { + method: capture.request.method, + ...pending, + timer: undefined + }) + dispatcher.dispatch(capture.response) + const error = await failed + expect(error).toBeInstanceOf(CodexAppServerRequestError) + expect(error).toMatchObject({ + method: capture.request.method, + code: capture.response.error.code, + providerDiagnostic: { text: capture.response.error.message, audience: 'person' } + }) + expectExitEvidenceRequired(error) + }) + + it('parses the recorded Pi rejection through the JSONL peer', async () => { + const capture = captures.pi + const input = new PassThrough() + const output = new PassThrough() + const peer = new JsonlRpcPeer(input, output) + try { + const failed = peer + .request(capture.request.type, { + provider: capture.request.provider, + modelId: capture.request.modelId + }) + .catch((caught: unknown) => caught) + // Only correlation changes: the peer allocates orca-1 instead of the capture harness id. + input.write(`${JSON.stringify({ ...capture.response, id: 'orca-1' })}\n`) + const error = await failed + expect(error).toBeInstanceOf(JsonlRpcResponseError) + expect(error).toMatchObject({ + command: capture.response.command, + message: capture.response.error + }) + expectExitEvidenceRequired(error) + } finally { + peer.close() + input.destroy() + output.destroy() + } + }) + + it('parses the recorded ACP agent error without losing code or data', async () => { + const capture = captures.acp + const pending = Promise.withResolvers() + const failed = pending.promise.catch((caught: unknown) => caught) + const diagnose = vi.fn() + settleAcpResponse(acpRpcEnvelopeSchema.parse(capture.response), pending, diagnose) + const error = await failed + expect(error).toBeInstanceOf(AcpAgentError) + expect(error).toMatchObject(capture.response.error) + expect(diagnose).not.toHaveBeenCalled() + expectExitEvidenceRequired(error) + }) + + it.each([ + ['typed refusal', new AgentSessionAcquisitionRefusal('sign in', 'notSignedIn')], + [ + 'local request deadline', + withProviderRequestTimeout(new Error('deadline'), { method: 'initialize', timeoutMs: 250 }) + ], + ['host fault', new Error('journal write failed')] + ])('keeps %s final even with exit proof (local policy control)', (_name, error) => { + expect(retryableStructuredAgentSessionStart(error)).toBe(false) + expect( + retryableStructuredAgentSessionAcquisition(new AgentSessionAcquisitionExitProvenError(error)) + ).toBe(false) + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-start-retry.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-start-retry.ts new file mode 100644 index 00000000000..b664ce6bb9c --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-start-retry.ts @@ -0,0 +1,60 @@ +import { CodexAppServerRequestError } from '../../codex/app-server/codex-app-server-request-error' +import { ClaudeControlRequestError } from '../../claude/claude-agent-sdk-control-requests' +import { AcpRpcError } from '../../acp/acp-errors' +import { JsonlRpcResponseError } from '../../jsonl-rpc/peer' +import { findInErrorChain } from '../../provider-process/provider-error-chain' +import { providerExecutableMissing } from '../../provider-process/provider-executable-missing' +import { providerRequestTimeoutOf } from '../../provider-process/provider-request-timeout' +import { argumentProblemOf } from '../structured-agent-arguments-error' +import { agentCliInstallationProblemOf } from '../agent-cli-installation-error' +import { + AgentSessionAcquisitionExitProvenError, + AgentSessionAcquisitionRootExitObservedError, + AgentSessionAcquisitionExitUnprovenError, + AgentSessionAcquisitionRefusal, + isAgentSessionPreSpawnError +} from './structured-agent-session-adapter' +import { providerExitObserved } from './structured-agent-session-failure-text' + +/** Only positive provider evidence retries; an Orca fault or a typed refusal stays final. */ +export function retryableStructuredAgentSessionStart(error: unknown): boolean { + if ( + providerRequestTimeoutOf(error) || + argumentProblemOf(error) || + agentCliInstallationProblemOf(error) || + providerExecutableMissing(error) + ) { + return false + } + const final = findInErrorChain(error, (cause) => + cause instanceof AggregateError || + cause instanceof AgentSessionAcquisitionExitUnprovenError || + cause instanceof AgentSessionAcquisitionRefusal || + isAgentSessionPreSpawnError(cause) + ? true + : undefined + ) + if (final) { + return false + } + return ( + providerExitObserved(error) || + findInErrorChain(error, (cause) => + cause instanceof CodexAppServerRequestError || + cause instanceof ClaudeControlRequestError || + cause instanceof AcpRpcError || + cause instanceof JsonlRpcResponseError || + ('code' in cause && ['EAGAIN', 'EMFILE', 'ENFILE'].some((code) => code === cause.code)) + ? true + : undefined + ) === true + ) +} + +export function retryableStructuredAgentSessionAcquisition(error: unknown): boolean { + return ( + (error instanceof AgentSessionAcquisitionExitProvenError || + error instanceof AgentSessionAcquisitionRootExitObservedError) && + retryableStructuredAgentSessionStart(error) + ) +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-start-settlement-host.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-start-settlement-host.test.ts new file mode 100644 index 00000000000..61c02278448 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-start-settlement-host.test.ts @@ -0,0 +1,517 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type * as TimerPromises from 'node:timers/promises' +import { CodexAppServerRequestError } from '../../codex/app-server/codex-app-server-request-error' +import { withProviderRequestTimeout } from '../../provider-process/provider-request-timeout' +import { AgentSessionAcquisitionRefusal } from './structured-agent-session-adapter' +import { agentSessionFailureWords } from '../../../shared/agent-session-failure-words' +import { agentSessionFailureFact } from '../../../shared/agent-session-failure' +import { DISPATCH_REJECTED_CANCELLED } from '../../../shared/structured-agent-session-dispatch-rejection' +import * as startFlight from './structured-agent-session-start-flight' +import { STRUCTURED_AGENT_SESSION_START_WAIT_MS } from './structured-agent-session-send-settlement' +import { + CALLER, + attach, + envelope, + hostTestState +} from './structured-agent-session-host-test-harness' +import { + HOST_TEST_SESSION as SESSION, + hostTestAttachParams, + hostTestMessage +} from './structured-agent-session-host-test-data' + +const retryClock = vi.hoisted(() => ({ fake: false })) +vi.mock('node:timers/promises', async (importOriginal) => { + const actual = await importOriginal() + return { + ...actual, + setTimeout: (...args: Parameters) => { + if (!retryClock.fake) { + return actual.setTimeout(...args) + } + // ast-grep-ignore: test-no-real-time-wait -- routes the retry backoff onto the faked global timer + return new Promise((resolve) => setTimeout(() => resolve(args[1]), args[0])) + } + } +}) + +let host: ReturnType['host'] +let acquire: ReturnType['acquire'] +let dispatch: ReturnType['dispatch'] +const openGates = new Map void>() +const eventually = (assertion: () => void | Promise) => + vi.waitFor(assertion, { timeout: 10_000, interval: 0 }) + +beforeEach(async () => { + retryClock.fake = false + vi.useFakeTimers({ + toFake: ['Date', 'setTimeout', 'clearTimeout', 'setInterval', 'clearInterval'] + }) + ;({ host, acquire, dispatch } = hostTestState()) + await attach() + await host.close(SESSION, 'evict') + acquire.mockClear() + openGates.clear() + host.deps.adapter.open = ({ acquisitionGeneration }) => + new Promise((resolve) => { + openGates.set(acquisitionGeneration!, () => resolve({ reportedOptions: {} })) + }) +}) + +afterEach(() => vi.useRealTimers()) + +async function accept(text: string, options?: Readonly>): Promise { + const body = hostTestMessage(text) + const fields = { body, ...(options ? { options } : {}) } + const params = { envelope: envelope('agentSession.send', fields), ...fields } + expect(await host.send(CALLER, params)).toMatchObject({ ok: true }) + return params.envelope.clientOperationId +} + +async function submission(id: string) { + return (await host.journalSnapshot(SESSION)).submissions.find((row) => row.clientMessageId === id) +} + +async function heldBehindStart(text: string): Promise { + const id = await accept(text) + await eventually(() => + expect(host.collaboratorsForTests().sessions.get(SESSION)?.child).toBeTruthy() + ) + return id +} + +async function proveOpen(): Promise { + const generation = host.collaboratorsForTests().sessions.get(SESSION)?.child?.generation + if (!generation) { + throw new Error('expected a published child') + } + await eventually(() => expect(openGates.has(generation)).toBe(true)) + openGates.get(generation)!() + openGates.delete(generation) +} + +function stop() { + return host.cancel(CALLER, { envelope: envelope('agentSession.cancel', {}) }) +} + +function goal() { + const change = { kind: 'set' as const, objective: 'finish startup' } + return host.changeThreadGoal(CALLER, { + envelope: envelope('agentSession.threadGoal', { change }), + change + }) +} + +async function endAttempt() { + const session = host.collaboratorsForTests().sessions.get(SESSION)! + const child = session.child! + if (!child.generation) { + throw new Error('expected a published child') + } + await host.handleAdapterEvent({ + type: 'ended', + sessionId: SESSION, + fence: child.fence, + acquisitionGeneration: child.generation, + reason: 'exited while starting', + cause: 'unexpected-exit' + }) + await eventually(() => expect(session.child).toBeNull()) +} + +describe('final logical startup settlement', () => { + it.each([false, true])( + 'honors accepted Stop between final exit evidence and settlement (message owns start: %s)', + async (messageOwned) => { + host.deps.adapter.changeThreadGoal = vi.fn(async () => ({ ok: true as const })) + const operation = messageOwned ? undefined : goal() + const id = messageOwned ? await heldBehindStart('withdraw before settlement') : undefined + const { sessions, runtimeState, lifetime } = host.collaboratorsForTests() + await eventually(() => expect(sessions.get(SESSION)?.child).toBeTruthy()) + const flight = runtimeState.startFlights.get(SESSION)! + let stopped: ReturnType | undefined + const makeLock = startFlight.structuredAgentSessionStartFlightLock + const lockFactory = vi + .spyOn(startFlight, 'structuredAgentSessionStartFlightLock') + .mockImplementation((input) => { + const lock = makeLock(input) + return { + ...lock, + commit: (step) => { + const committed = lock.commit(step) + void committed.then( + (value) => { + if ( + value && + typeof value === 'object' && + 'retry' in value && + value.retry === false + ) { + // Queue real Stop after exit evidence, before the owner's await continuation. + stopped = stop() + } + }, + () => undefined + ) + return committed + } + } + }) + try { + // The current attempt already captured its lock; retry once to install the seam. + await endAttempt() + await eventually(() => expect(acquire).toHaveBeenCalledTimes(2)) + await eventually(() => expect(sessions.get(SESSION)?.child).toBeTruthy()) + await lifetime.failStartingChild( + SESSION, + sessions.get(SESSION)!.child!, + withProviderRequestTimeout(new CodexAppServerRequestError('initialize', -1, 'deadline'), { + method: 'initialize', + timeoutMs: 60_000 + }), + 60_000 + ) + await expect(flight.outcome).resolves.toMatchObject({ ok: false, aborted: true }) + expect(stopped).toBeDefined() + expect(await stopped).toMatchObject({ ok: true, value: { cancelled: true } }) + if (operation) { + expect(await operation).toMatchObject({ ok: false }) + } + if (id) { + expect(await submission(id)).toMatchObject({ + dispatchState: 'rejected', + reason: DISPATCH_REJECTED_CANCELLED + }) + } + const snapshot = await host.journalSnapshot(SESSION) + expect( + snapshot.items.filter((item) => item.body.kind === 'status' && item.body.tone === 'error') + ).toEqual([]) + expect(snapshot.items.filter((item) => item.body.kind === 'status')).toHaveLength( + messageOwned ? 1 : 0 + ) + expect(dispatch).not.toHaveBeenCalled() + expect(acquire).toHaveBeenCalledTimes(2) + expect(runtimeState.startFlights.has(SESSION)).toBe(false) + } finally { + lockFactory.mockRestore() + } + } + ) + + it.each(['deadline', 'stop'] as const)( + 'ends a joining operation on its own %s without running it on a replacement child', + async (ending) => { + const ran = vi.fn(async () => ({ ok: true as const })) + host.deps.adapter.changeThreadGoal = ran + const first = await heldBehindStart('message owns the start') + const operation = goal() + // Ensure the operation joined this flight before ending its first child. + await host.collaboratorsForTests().serialize(SESSION, async () => undefined) + await endAttempt() + if (ending === 'stop') { + expect(await stop()).toMatchObject({ ok: true, value: { cancelled: true } }) + } else { + await vi.advanceTimersByTimeAsync(STRUCTURED_AGENT_SESSION_START_WAIT_MS) + } + expect(await operation).toMatchObject({ + ok: false, + refusal: { + code: + ending === 'deadline' + ? 'agent_session_operation_expired' + : 'agent_session_owner_restart_failed' + } + }) + expect(ran).not.toHaveBeenCalled() + if (ending === 'deadline') { + expect(await submission(first)).toMatchObject({ dispatchState: 'pending' }) + await eventually(() => + expect(host.collaboratorsForTests().sessions.get(SESSION)?.child).toBeTruthy() + ) + await proveOpen() + await eventually(async () => + expect((await submission(first))?.dispatchState).toBe('accepted') + ) + } else { + expect(await submission(first)).toMatchObject({ + dispatchState: 'rejected', + reason: DISPATCH_REJECTED_CANCELLED + }) + } + expect(ran).not.toHaveBeenCalled() + } + ) + + it('recovers a message-owned failed final write on the next accept without transferring its failure', async () => { + const first = await heldBehindStart('original requester') + const later = await accept('later queued message') + const { sessions, lifetime, conversationDelivery } = host.collaboratorsForTests() + const journal = sessions.get(SESSION)!.journal + const append = journal.appendLifecycleBatch.bind(journal) + const write = vi.spyOn(journal, 'appendLifecycleBatch').mockImplementation((input) => { + if (input.settlementId.startsWith('start-failure:')) { + return Promise.reject(new Error('failure write unavailable')) + } + return append(input) + }) + await lifetime.failStartingChild( + SESSION, + sessions.get(SESSION)!.child!, + withProviderRequestTimeout(new CodexAppServerRequestError('initialize', -1, 'deadline'), { + method: 'initialize', + timeoutMs: 60_000 + }), + 60_000 + ) + await eventually(() => expect(conversationDelivery.loop.isRunning(SESSION)).toBe(false)) + expect(await submission(first)).toMatchObject({ dispatchState: 'pending' }) + expect(await submission(later)).toMatchObject({ dispatchState: 'pending' }) + expect(dispatch).not.toHaveBeenCalled() + write.mockRestore() + const wake = await accept('new wake after journal recovery') + await eventually(async () => expect((await submission(first))?.dispatchState).toBe('rejected')) + const failed = await host.journalSnapshot(SESSION) + expect(failed.submissions.find((row) => row.clientMessageId === first)).toMatchObject({ + dispatchState: 'rejected', + rejection: { + kind: 'startFailed', + requestTimeout: { method: 'initialize', timeoutMs: 60_000 } + } + }) + expect(failed.submissions.find((row) => row.clientMessageId === later)).toMatchObject({ + dispatchState: 'pending', + reason: null + }) + expect( + failed.items.filter((item) => item.body.kind === 'status' && item.body.tone === 'error') + ).toEqual([ + expect.objectContaining({ + body: expect.objectContaining({ failure: expect.objectContaining({ kind: 'startFailed' }) }) + }) + ]) + await eventually(() => expect(acquire).toHaveBeenCalledTimes(2)) + await proveOpen() + await eventually(() => + expect(dispatch.mock.calls.map(([input]) => input.clientMessageId)).toEqual([later, wake]) + ) + const snapshot = await host.journalSnapshot(SESSION) + expect( + snapshot.items.filter((item) => item.body.kind === 'status' && item.body.tone === 'error') + ).toHaveLength(1) + }) + + it('preserves pinned launch inputs and failure words through direct and flight-owned starts', async () => { + const options = { model: 'gpt-5', effort: 'high' } + await host.deps.store.replaceSessionOptions({ + sessionId: SESSION, + fence: host.deps.store.getRecord(SESSION)!.lease.runtimeFence, + options, + now: Date.now() + }) + const refuse = async () => { + throw new AgentSessionAcquisitionRefusal('sign in', 'notSignedIn') + } + acquire.mockImplementation(refuse) + const attached = await host.attach( + CALLER, + hostTestAttachParams(host.deps.store.getRecord(SESSION)!.lease.runtimeFence) + ) + expect(attached).toMatchObject({ + ok: false, + refusal: { code: 'agent_session_operation_invalid', details: { reason: 'notSignedIn' } } + }) + expect(acquire).toHaveBeenCalledTimes(1) + const direct = acquire.mock.calls.at(-1)![0] + expect(direct.options).toEqual(options) + acquire.mockClear() + const first = await accept('same launch through a message', options) + await eventually(async () => expect((await submission(first))?.dispatchState).toBe('rejected')) + const flight = acquire.mock.calls[0]![0] + expect({ + launch: flight.launch, + options: flight.options, + permissionMode: flight.permissionMode, + identity: flight.identity + }).toEqual({ + launch: direct.launch, + options: direct.options, + permissionMode: direct.permissionMode, + identity: direct.identity + }) + expect(await submission(first)).toMatchObject( + agentSessionFailureWords(agentSessionFailureFact('notSignedIn'), { + agentName: 'Codex', + surface: 'rejection' + }) + ) + expect(acquire).toHaveBeenCalledTimes(1) + expect(dispatch).not.toHaveBeenCalled() + }) + + // Leave room for five starts per requester under the saved provider-handle cap. + it.each([4, 40])( + 'bounds physical work and retires starts for %i accepted requesters', + async (count) => { + retryClock.fake = true + const { sessions, runtimeState, conversationDelivery, serialize } = + host.collaboratorsForTests() + const retained = hostTestAttachParams(host.deps.store.getRecord(SESSION)!.lease.runtimeFence) + expect(await host.attach(CALLER, retained)).toMatchObject({ ok: true }) + const child = sessions.get(SESSION)!.child + expect(child).toBeTruthy() + for (let n = 0; n < count; n++) { + expect(await host.attach(CALLER, retained)).toMatchObject({ ok: true, replayed: true }) + expect(sessions.get(SESSION)!.child).toBe(child) + } + expect(acquire).toHaveBeenCalledTimes(1) + await host.close(SESSION, 'evict') + expect(runtimeState.currentEventSink(SESSION)).toBeUndefined() + acquire.mockClear() + openGates.clear() + const other = 'unrelated-chat' + const unrelated = hostTestAttachParams(null, { + providerHandle: undefined, + envelope: envelope( + 'agentSession.attach', + {}, + { sessionId: other, expectedRuntimeFence: null } + ) + }) + expect(await host.create(CALLER, unrelated)).toMatchObject({ ok: true }) + await serialize(other, async () => {}) + expect(conversationDelivery.loop.isRunning(other)).toBe(false) + const otherJournal = sessions.get(other)!.journal + const frames = Array.from({ length: count }, () => 0) + const unsubscribe = frames.map((_, index) => + host.subscribeStatus({ + id: `count-${index}`, + emit: (event) => { + if (event.type === 'status' && event.session.sessionId === SESSION) { + frames[index]!++ + } + } + }) + ) + frames.fill(0) + const reads = vi.spyOn(otherJournal, 'submissions') + const writes = vi.spyOn(otherJournal, 'appendLifecycleBatch') + const probe = vi.fn>(async () => ({ + outcome: 'pid-absent' + })) + host.deps.probeOwner = probe + const baselineTimers = vi.getTimerCount() + const active = new Set() + const begin = runtimeState.acquireAborts.begin.bind(runtimeState.acquireAborts) + const handles = vi + .spyOn(runtimeState.acquireAborts, 'begin') + .mockImplementation((...args) => { + const handle = begin(...args) + active.add(handle.signal) + return { + ...handle, + end: () => { + active.delete(handle.signal) + handle.end() + } + } + }) + try { + const ids: string[] = [] + for (let n = 0; n < count; n++) { + ids.push(await accept(`failed requester ${n}`)) + } + const successor = [await accept('successor first'), await accept('successor second')] + await eventually(() => expect(sessions.get(SESSION)?.child).toBeTruthy()) + expect(acquire).toHaveBeenCalledTimes(1) + expect(runtimeState.startFlights.get(SESSION)!.children).toHaveLength(1) + const settlements = vi.spyOn(sessions.get(SESSION)!.journal, 'appendLifecycleBatch') + for (let n = 0; n < count; n++) { + await eventually(() => expect(acquire).toHaveBeenCalledTimes(n * 5 + 1)) + const flight = runtimeState.startFlights.get(SESSION)! + for (let attempt = 0; attempt < 5; attempt++) { + await eventually(() => expect(acquire).toHaveBeenCalledTimes(n * 5 + attempt + 1)) + await eventually(() => expect(sessions.get(SESSION)?.child).toBeTruthy()) + await endAttempt() + if (attempt < 4) { + await vi.advanceTimersByTimeAsync(100 * 2 ** attempt) + } + } + await expect(flight.outcome).resolves.toMatchObject({ ok: false }) + expect(flight.children).toHaveLength(5) + } + await eventually(() => expect(acquire).toHaveBeenCalledTimes(5 * count + 1)) + await proveOpen() + await eventually(() => + expect(dispatch.mock.calls.map(([input]) => input.clientMessageId)).toEqual(successor) + ) + const snapshot = await host.journalSnapshot(SESSION) + const rejected = snapshot.submissions.filter((row) => row.dispatchState === 'rejected') + expect(rejected.map((row) => row.clientMessageId)).toEqual(ids) + const finalWrites = settlements.mock.calls.filter(([batch]) => + batch.settlementId.startsWith('start-failure:') + ) + expect(finalWrites).toHaveLength(count) + expect( + finalWrites.flatMap( + ([batch]) => batch.dispatches?.map((row) => row.clientMessageId) ?? [] + ) + ).toEqual(ids) + settlements.mockRestore() + expect(snapshot.items.filter((item) => item.body.kind === 'status')).toHaveLength(count) + expect( + snapshot.items.filter((item) => item.body.kind === 'status' && item.body.tone === 'error') + ).toHaveLength(count) + expect(runtimeState.startFlights.has(SESSION)).toBe(false) + expect(runtimeState.startupAttempts.isOpen(SESSION)).toBe(false) + await host.close(SESSION, 'evict') + const stopped: string[] = [] + for (let n = 0; n < count; n++) { + stopped.push(await accept(`withdraw requester ${n}`)) + } + for (let attempt = 0; attempt < 4; attempt++) { + await eventually(() => expect(sessions.get(SESSION)?.child).toBeTruthy()) + await endAttempt() + await vi.advanceTimersByTimeAsync(100 * 2 ** attempt) + } + await eventually(() => expect(acquire).toHaveBeenCalledTimes(5 * count + 6)) + await eventually(() => expect(sessions.get(SESSION)?.child).toBeTruthy()) + expect(await stop()).toMatchObject({ ok: true, value: { cancelled: true } }) + await vi.advanceTimersByTimeAsync(2000) + await eventually(() => expect(conversationDelivery.loop.isRunning(SESSION)).toBe(false)) + expect(acquire).toHaveBeenCalledTimes(5 * count + 6) + expect(dispatch.mock.calls.map(([input]) => input.clientMessageId)).toEqual(successor) + const final = await host.journalSnapshot(SESSION) + expect(final.submissions.filter((row) => stopped.includes(row.clientMessageId))).toEqual( + stopped.map((clientMessageId) => + expect.objectContaining({ + clientMessageId, + dispatchState: 'rejected', + reason: DISPATCH_REJECTED_CANCELLED + }) + ) + ) + expect(runtimeState.startFlights.has(SESSION)).toBe(false) + expect(runtimeState.startupAttempts.isOpen(SESSION)).toBe(false) + expect(runtimeState.currentEventSink(SESSION)).toBeUndefined() + expect(active.size).toBe(0) + expect(vi.getTimerCount()).toBeLessThanOrEqual(baselineTimers) + const place = runtimeState.deliveryOrder.reserve(SESSION) + await place.before + place.release() + expect(reads).not.toHaveBeenCalled() + expect(writes).not.toHaveBeenCalled() + expect(probe.mock.calls.every(([record]) => record.sessionId !== other)).toBe(true) + expect(frames[0]).toBeGreaterThan(0) + expect(frames[0]).toBeLessThanOrEqual(20 * (5 * count + 6) + 8 * count) + expect(frames.every((value) => value === frames[0])).toBe(true) + } finally { + handles.mockRestore() + reads.mockRestore() + writes.mockRestore() + unsubscribe.forEach((off) => off()) + } + }, + 120_000 + ) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-startup-attempt-host.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-startup-attempt-host.test.ts index 5c177b8e654..c5aba3c82fd 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-startup-attempt-host.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-startup-attempt-host.test.ts @@ -2,9 +2,13 @@ import { mkdtemp, rm } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' import { afterEach, beforeEach, describe, expect, it, vi, type Mock } from 'vitest' +import { CodexAppServerRequestError } from '../../codex/app-server/codex-app-server-request-error' +import { withProviderRequestTimeout } from '../../provider-process/provider-request-timeout' +import { AgentSessionAcquisitionRefusal } from './structured-agent-session-adapter' import { agentSessionFailureFact } from '../../../shared/agent-session-failure' import { agentSessionFailureWords } from '../../../shared/agent-session-failure-words' import type { AgentJournalSubmission } from '../../../shared/agent-session-journal-types' +import { AGENT_JOURNAL_THREAD_SCOPE } from '../../../shared/agent-session-journal-types' import { computeAgentSessionPayloadFingerprint } from '../../../shared/agent-session-mutation-envelope' import { codexProviderHandle } from '../../../shared/agent-session-provider-handle-encoding' import { DISPATCH_REJECTED_CANCELLED } from '../../../shared/structured-agent-session-dispatch-rejection' @@ -18,7 +22,10 @@ import type { StructuredAgentSessionAcquireInput, StructuredAgentSessionAdapter } from './structured-agent-session-adapter' -import { NO_STRUCTURED_AGENTS } from './structured-agent-session-adapter-router-test-support' +import { + claudeAndCodexDeclared, + NO_STRUCTURED_AGENTS +} from './structured-agent-session-adapter-router-test-support' import { StructuredAgentSessionHost } from './structured-agent-session-host' import { HOST_TEST_NOW as NOW, @@ -45,6 +52,7 @@ let dispatch: Mock let closeSession: Mock> let setOption: Mock let startupLimits: Partial | undefined +let supportsGoals = false const providerStarted = vi.fn() const recordLiveListing = vi.fn() const openGates = new Map void>() @@ -79,7 +87,7 @@ const publishFirst: StructuredAgentSessionAdapter['acquire'] = async (input) => async function startHost(): Promise { host = new StructuredAgentSessionHost({ - agents: NO_STRUCTURED_AGENTS, + agents: supportsGoals ? claudeAndCodexDeclared() : NO_STRUCTURED_AGENTS, logger: createStructuredAgentSessionLogger(), store, adapter: { @@ -119,6 +127,7 @@ async function restartWith(next: StructuredAgentSessionAdapter['acquire']): Prom acquire.mockImplementation(next) closeSession.mockClear() await startHost() + acquire.mockClear() } beforeEach(async () => { @@ -128,6 +137,7 @@ beforeEach(async () => { root = await mkdtemp(join(tmpdir(), 'orca-startup-attempt-')) resetHostTestOperationIds() startupLimits = undefined + supportsGoals = false providerStarted.mockReset() recordLiveListing.mockReset() openGates.clear() @@ -261,24 +271,55 @@ describe('a send while the agent starts', () => { }) describe('a start that ends before it proves itself', () => { - it('settles what it held as never sent when the agent exits during startup', async () => { - await restartWith(publishFirst) - const id = await heldBehindStart('hello') + it.each([false, true])( + 'settles only its own message after startup failure (acquisition refusal: %s)', + async (refused) => { + await restartWith(publishFirst) + const id = await heldBehindStart('hello') + let later: string | undefined - await host.handleAdapterEvent({ - type: 'ended', - sessionId: SESSION, - fence: store.getRecord(SESSION)!.lease.runtimeFence, - acquisitionGeneration: `generation-${acquire.mock.calls.length}`, - reason: 'exited with code 1', - cause: 'unexpected-exit' - }) + const physicalAttempts = refused ? 1 : 5 + for (let attempt = 1; attempt <= physicalAttempts; attempt++) { + await eventually(() => expect(acquire).toHaveBeenCalledTimes(attempt)) + await eventually(() => + expect(host.collaboratorsForTests().sessions.get(SESSION)?.child).toBeTruthy() + ) + if (attempt === physicalAttempts) { + later = await accept('a later message') + if (refused) { + acquire.mockRejectedValueOnce( + new AgentSessionAcquisitionRefusal('sign in', 'notSignedIn') + ) + } + } + await host.handleAdapterEvent({ + type: 'ended', + sessionId: SESSION, + fence: store.getRecord(SESSION)!.lease.runtimeFence, + acquisitionGeneration: `generation-${acquire.mock.calls.length}`, + reason: 'exited with code 1', + cause: 'unexpected-exit' + }) + } - await eventually(async () => expect((await submission(id))?.dispatchState).toBe('rejected')) - expect((await submission(id))?.handedOverAt).toBeUndefined() - expect(dispatch).not.toHaveBeenCalled() - expect(startupAttemptOpen()).toBe(false) - }) + await eventually(async () => expect((await submission(id))?.dispatchState).toBe('rejected')) + expect(await submission(id)).toMatchObject( + agentSessionFailureWords( + agentSessionFailureFact(refused ? 'notSignedIn' : 'providerStartFailed'), + { agentName: 'Codex', surface: 'rejection' } + ) + ) + expect((await submission(id))?.handedOverAt).toBeUndefined() + expect(dispatch).not.toHaveBeenCalled() + await eventually(() => expect(acquire).toHaveBeenCalledTimes(refused ? 3 : 6)) + await proveOpen() + await eventually(async () => + expect((await submission(later!))?.dispatchState).toBe('accepted') + ) + expect(dispatch.mock.calls.map(([input]) => input.clientMessageId)).toEqual([later]) + expect(startupAttemptOpen()).toBe(false) + } + ) it('withdraws what it held on a Stop, without waiting on the handshake', async () => { await restartWith(publishFirst) @@ -488,3 +529,304 @@ describe('a chat after startup permission backfill', () => { } ) }) + +describe('one logical start across physical attempts', () => { + beforeEach(() => { + supportsGoals = true + }) + async function endAttempt() { + const session = host.collaboratorsForTests().sessions.get(SESSION)! + const child = session.child! + if (child.generation === null) { + throw new Error('expected a published child generation') + } + const event = { + type: 'ended' as const, + sessionId: SESSION, + fence: child.fence, + acquisitionGeneration: child.generation, + reason: 'exited while starting', + cause: 'unexpected-exit' as const + } + await host.handleAdapterEvent(event) + await eventually(() => expect(session.child).toBeNull()) + return event + } + + function goal() { + const change = { kind: 'set' as const, objective: 'finish startup' } + return host.changeThreadGoal(CALLER, { + envelope: { + sessionId: SESSION, + clientOperationId: hostTestOperationId(), + expectedRuntimeFence: null, + payloadFingerprint: computeAgentSessionPayloadFingerprint({ + method: 'agentSession.threadGoal', + sessionId: SESSION, + fields: { change } + }) + }, + change + }) + } + + it('holds both sends and a joining operation across retry without status or transcript churn', async () => { + await restartWith(publishFirst) + const phases: (string | undefined)[] = [] + host.subscribeStatus({ + id: 'retry-observer', + emit: (event) => { + if (event.type === 'status') { + phases.push(event.session.hostExecutionPhase) + } + } + }) + const ran = vi.fn>(async () => ({ + ok: true as const + })) + host.deps.adapter.changeThreadGoal = ran + const artifacts: unknown[] = [] + await host.subscribe({ + id: 'retry-transcript', + sessionId: SESSION, + emit: (frame) => { + if (frame.type === 'batch') { + artifacts.push(...frame.batch.items.filter((item) => item.body.kind === 'status')) + } + } + }) + const first = await heldBehindStart('first') + phases.length = 0 + const operation = goal() + const stale = await endAttempt() + const second = await accept('second') + expect(await submission(first)).toMatchObject({ dispatchState: 'pending' }) + expect((await host.readStatusSummary(SESSION))?.hostExecutionPhase).toBe('starting') + const reconnect: (string | undefined)[] = [] + host.subscribeStatus({ + id: 'during-backoff', + emit: (event) => { + if (event.type === 'snapshot') { + reconnect.push(...event.sessions.map((row) => row.hostExecutionPhase)) + } + } + }) + expect(reconnect).toContain('starting') + const during = await host.journalSnapshot(SESSION) + expect(during.items.filter((item) => item.body.kind === 'status')).toEqual([]) + expect( + host.collaboratorsForTests().sessions.get(SESSION)?.journal.stopMarks.latest() + ).toBeNull() + await eventually(() => expect(acquire).toHaveBeenCalledTimes(2)) + await proveOpen() + await host.handleAdapterEvent(stale) + await eventually(async () => expect((await submission(second))?.dispatchState).toBe('accepted')) + expect(dispatch.mock.calls.map(([input]) => input.clientMessageId)).toEqual([first, second]) + expect(phases.filter((phase) => phase !== 'starting' && phase !== 'ready')).toEqual([]) + expect(artifacts).toEqual([]) + expect((await host.readStatusSummary(SESSION))?.hostExecutionPhase).toBe('ready') + expect(await operation).toMatchObject({ ok: true }) + expect(ran).toHaveBeenCalledTimes(1) + expect(ran.mock.calls[0]?.[0]).toMatchObject({ + fence: host.collaboratorsForTests().sessions.get(SESSION)?.child?.fence + }) + expect(acquire).toHaveBeenCalledTimes(2) + }) + + it('keeps a typed refusal or an unproven close final without respawning', async () => { + await restartWith(async () => { + throw new AgentSessionAcquisitionRefusal('sign in', 'notSignedIn') + }) + const first = await accept('needs login') + await eventually(async () => expect((await submission(first))?.dispatchState).toBe('rejected')) + expect(acquire).toHaveBeenCalledTimes(1) + acquire.mockClear() + await restartWith(publishFirst) + closeSession.mockResolvedValue(false) + const second = await heldBehindStart('unproven process') + const child = host.collaboratorsForTests().sessions.get(SESSION)!.child! + const { lifetime } = host.collaboratorsForTests() + await expect( + lifetime.failStartingChild( + SESSION, + child, + new CodexAppServerRequestError('thread/start', -1, 'refused'), + 1 + ) + ).rejects.toThrow('eviction failed') + await eventually(async () => expect((await submission(second))?.dispatchState).toBe('rejected')) + expect(host.collaboratorsForTests().sessions.get(SESSION)?.child).toBe(child) + expect(acquire).toHaveBeenCalledTimes(1) + closeSession.mockResolvedValue(true) + }) + + it('keeps a starting child journal sink fault final for its requester without respawning', async () => { + await restartWith(publishFirst) + const first = await heldBehindStart('journal fails while starting') + const { sessions, runtimeState } = host.collaboratorsForTests() + const session = sessions.get(SESSION)! + const sink = runtimeState.currentEventSink(SESSION)! + const writes = vi.spyOn(session.journal, 'appendLifecycleBatch') + vi.spyOn(session.journal, 'appendItem').mockRejectedValueOnce(new Error('journal unavailable')) + sink.sink.appendItem( + { provider: 'codex', threadId: THREAD, turnId: 'turn-starting', ordinal: 1 }, + { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text: 'starting' }] }, + { turnScope: AGENT_JOURNAL_THREAD_SCOPE } + ) + await eventually(async () => expect((await submission(first))?.dispatchState).toBe('rejected')) + expect(session.lastEndedChild).toMatchObject({ + cause: 'exit', + duringStartup: true, + rootGone: true, + failure: { kind: 'hostFault' } + }) + const failures = writes.mock.calls.filter(([input]) => + input.settlementId.startsWith('start-failure:') + ) + expect(failures).toHaveLength(1) + expect(failures[0]![0].dispatches).toEqual([ + expect.objectContaining({ + clientMessageId: first, + state: 'rejected', + rejection: { kind: 'hostFault' } + }) + ]) + const snapshot = await host.journalSnapshot(SESSION) + expect(snapshot.items.filter((item) => item.body.kind === 'status')).toEqual([ + expect.objectContaining({ body: expect.objectContaining({ failure: { kind: 'hostFault' } }) }) + ]) + expect(closeSession).toHaveBeenCalledTimes(1) + expect(acquire).toHaveBeenCalledTimes(1) + expect(dispatch).not.toHaveBeenCalled() + }) + + it('ends a backoff on Stop and keeps the original ceiling across attempts', async () => { + startupLimits = { ceilingMs: 1000 } + await restartWith(publishFirst) + const first = await heldBehindStart('stop in backoff') + await endAttempt() + expect(await stop()).toMatchObject({ ok: true }) + await vi.advanceTimersByTimeAsync(500) + expect(acquire).toHaveBeenCalledTimes(1) + expect((await host.readStatusSummary(SESSION))?.hostExecutionPhase).toBeUndefined() + expect(await submission(first)).toMatchObject({ dispatchState: 'rejected' }) + acquire.mockClear() + const second = await heldBehindStart('one clock') + await vi.advanceTimersByTimeAsync(400) + await endAttempt() + await eventually(() => expect(acquire).toHaveBeenCalledTimes(2)) + await vi.advanceTimersByTimeAsync(500) + await eventually( + async () => expect((await submission(second))?.dispatchState).toBe('rejected'), + 50 + ) + expect(acquire).toHaveBeenCalledTimes(2) + expect(await submission(second)).toMatchObject(HOST_STOPPED) + }) + + it('keeps an elapsed provider-step deadline final and preserves its words', async () => { + await restartWith(publishFirst) + const first = await heldBehindStart('initialize deadline') + const { sessions, lifetime } = host.collaboratorsForTests() + const requestTimeout = { method: 'initialize', timeoutMs: 60_000 } + await vi.advanceTimersByTimeAsync(60_000) + await lifetime.failStartingChild( + SESSION, + sessions.get(SESSION)!.child!, + withProviderRequestTimeout( + new CodexAppServerRequestError('initialize', -1, 'deadline'), + requestTimeout + ), + 60_000 + ) + await eventually(async () => expect((await submission(first))?.dispatchState).toBe('rejected')) + expect(await submission(first)).toMatchObject( + agentSessionFailureWords(agentSessionFailureFact('startFailed', { requestTimeout }), { + agentName: 'Codex', + surface: 'rejection' + }) + ) + expect(acquire).toHaveBeenCalledTimes(1) + }) + + it('delivers a later message after a delivery step throws', async () => { + await restartWith(publishFirst) + const first = await heldBehindStart('failed preparation') + vi.spyOn( + host.collaboratorsForTests().runtimeState.deliveryOrder, + 'beforeSend' + ).mockImplementationOnce(() => { + throw new Error('delivery preparation failed') + }) + await proveOpen() + await eventually(async () => expect((await submission(first))?.dispatchState).toBe('rejected')) + const later = await accept('after preparation failed') + await eventually(async () => expect((await submission(later))?.dispatchState).toBe('accepted')) + expect(dispatch.mock.calls.map(([input]) => input.clientMessageId)).toEqual([later]) + }) + + it.each([ + { joins: false, failsWrite: false }, + { joins: true, failsWrite: false }, + { joins: true, failsWrite: true } + ])( + 'settles an operation start without rejecting a later message (joins: $joins, write fails: $failsWrite)', + async ({ joins, failsWrite }) => { + await restartWith(publishFirst) + host.deps.adapter.changeThreadGoal = vi.fn(async () => ({ ok: true as const })) + const operation = goal() + let later: string | undefined + let restoreWrite: (() => void) | undefined + for (let number = 1; number <= 5; number++) { + await eventually(() => expect(acquire).toHaveBeenCalledTimes(number)) + await eventually(() => + expect(host.collaboratorsForTests().sessions.get(SESSION)?.child).toBeTruthy() + ) + if (number === 5 && failsWrite) { + const journal = host.collaboratorsForTests().sessions.get(SESSION)!.journal + const append = journal.appendLifecycleBatch.bind(journal) + const write = vi.spyOn(journal, 'appendLifecycleBatch').mockImplementation((input) => { + if (input.settlementId.startsWith('start-failure:')) { + return Promise.reject(new Error('failure write unavailable')) + } + return append(input) + }) + restoreWrite = () => write.mockRestore() + } + await endAttempt() + if (number === 1 && joins) { + later = await accept('my own start') + } + } + expect(await operation).toMatchObject({ ok: false }) + const snapshot = await host.journalSnapshot(SESSION) + expect( + snapshot.items.filter((item) => item.body.kind === 'status' && item.body.tone === 'error') + ).toHaveLength(failsWrite ? 0 : 1) + if (later) { + const expectedMessages = [later] + expect(await submission(later)).toMatchObject({ dispatchState: 'pending' }) + if (failsWrite) { + await eventually(() => + expect(host.collaboratorsForTests().conversationDelivery.loop.isRunning(SESSION)).toBe( + false + ) + ) + expect(await submission(later)).toMatchObject({ dispatchState: 'pending' }) + restoreWrite?.() + expectedMessages.push(await accept('after the journal recovered')) + } + await eventually(() => expect(acquire).toHaveBeenCalledTimes(6)) + await proveOpen() + await eventually(async () => + expect((await submission(later))?.dispatchState).toBe('accepted') + ) + await eventually(() => + expect(dispatch.mock.calls.map(([input]) => input.clientMessageId)).toEqual( + expectedMessages + ) + ) + } + } + ) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-startup-attempt.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-startup-attempt.ts index e35198b2955..ff938b33886 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-startup-attempt.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-startup-attempt.ts @@ -1,4 +1,5 @@ import { randomUUID } from 'node:crypto' +import type { SubmissionRejectionFact } from '../../../shared/agent-session-failure' import { waitForStructuredAgentSessionStart, type StructuredAgentSessionStartReadiness @@ -72,6 +73,8 @@ type Tracked = { expired: boolean /** Defer expiry reporting until aborted spawn settles; successful readiness overrides it. */ expiry: StructuredAgentSessionStartupSettled | null + failure?: SubmissionRejectionFact + retry?: boolean readiness: PromiseWithResolvers<'ready' | 'ended'> } @@ -95,17 +98,35 @@ export class StructuredAgentSessionStartupAttempts { this.now = deps.now ?? Date.now } - begin(sessionId: string, attemptId: string): void { - this.register(sessionId, attemptId) + /** A start registered before its first wait: its ceiling runs from now. */ + begin(sessionId: string, attemptId: string, registeredAt = this.now()) { + const result = this.register(sessionId, attemptId, registeredAt) + return { ended: result.readiness.promise, result, registeredAt } } - private register(sessionId: string, attemptId: string): Tracked { + pastCeiling(registeredAt: number): boolean { + return this.now() - registeredAt >= this.limits.ceilingMs + } + + childFailed( + sessionId: string, + child: StructuredAgentSessionProviderChildIdentity, + failure: SubmissionRejectionFact, + retry: boolean + ): void { + const tracked = this.open.get(sessionId) + if (tracked?.child && sameProviderChild(tracked.child, child)) { + Object.assign(tracked, { failure, retry }) + } + } + + private register(sessionId: string, attemptId: string, registeredAt = this.now()): Tracked { this.end(sessionId) const tracked: Tracked = { attemptId, attempt: null, child: null, - registeredAt: this.now(), + registeredAt, spawnedAt: null, timer: null, expired: false, @@ -163,11 +184,11 @@ export class StructuredAgentSessionStartupAttempts { if (tracked?.attemptId !== attemptId) { return } + tracked.child = { generation: child.generation, fence: child.fence } if (child.phase === 'ready') { this.settle(sessionId, tracked, 'ready') return } - tracked.child = { generation: child.generation, fence: child.fence } if (tracked.expired) { this.expire(sessionId, tracked, null) return diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed-deps.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed-deps.ts index 76289775bf4..f59fde3446f 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed-deps.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed-deps.ts @@ -25,6 +25,7 @@ export type StatusFeedSession = { export type StructuredAgentSessionStatusFeedDeps = { sessions: ReadonlyMap + starting?: (sessionId: string) => boolean getRecord: (sessionId: string) => AgentSessionRecord | null /** Absent means no replaced runtime. */ replacedRuntime?: (sessionId: string) => { fence: number } | undefined diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.ts index 99e7821300f..5f92331c8e7 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.ts @@ -179,6 +179,7 @@ export class StructuredAgentSessionStatusFeed { const summary = structuredAgentSessionStatusSummary({ sessionId, session, + starting: this.deps.starting?.(sessionId), journal: source, record, state: projection.state, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-status-summary.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-status-summary.ts index 8164c2d3ba6..2d3dd29d1aa 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-status-summary.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-status-summary.ts @@ -21,6 +21,7 @@ export function structuredAgentSessionStatusSummary({ record, state, stopping, + starting, childWork, namingFailure, now @@ -34,6 +35,7 @@ export function structuredAgentSessionStatusSummary({ journal: AgentSessionJournal record: AgentSessionRecord | null state: StructuredAgentSessionStatusState + starting?: boolean stopping: boolean childWork: Pick namingFailure: StructuredChatNamingFailure | undefined @@ -49,10 +51,10 @@ export function structuredAgentSessionStatusSummary({ sessionId, workspaceId: session.params.location.workspaceId, agent: session.params.provider, - ...(session.child + ...(session.child || starting ? { hostExecutionOwned: true as const, - hostExecutionPhase: session.child.phase + hostExecutionPhase: session.child?.phase ?? 'starting' } : {}), ...(session.restartResume ? { restartResume: { phase: session.restartResume.phase } } : {}), diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-view-start-after-failed-start.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-view-start-after-failed-start.test.ts index 81d8d3e7626..9fd22829433 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-view-start-after-failed-start.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-view-start-after-failed-start.test.ts @@ -145,7 +145,7 @@ describe('a fresh chat whose Claude start fails', () => { ['after the create already died', false], ['while the create is still starting', true] ] as const)( - 'starts once for the open and once for a send, one row each, when the view binds %s', + 'starts once for the open and retries a send five times, one final row each, when the view binds %s', async (_when, createStillStarting) => { // Bind views before releasing create, independent of runner speed. let releaseCreate = (): void => {} @@ -190,13 +190,13 @@ describe('a fresh chat whose Claude start fails', () => { ).toMatchObject({ dispatchState: 'rejected', reason: startFailure }) ) await settleExits() - expect(claude.connections).toHaveLength(2) + expect(claude.connections).toHaveLength(6) expect(await timeline()).toEqual([startFailure, 'message', startFailure]) unsubscribe() await view(SURFACE) await settleExits() - expect(claude.connections).toHaveLength(2) + expect(claude.connections).toHaveLength(6) expect(await timeline()).toEqual([startFailure, 'message', startFailure]) } ) diff --git a/src/main/runtime/claude-structured-resumed-start-failure.test.ts b/src/main/runtime/claude-structured-resumed-start-failure.test.ts index 13cf2b557a0..db5f9474665 100644 --- a/src/main/runtime/claude-structured-resumed-start-failure.test.ts +++ b/src/main/runtime/claude-structured-resumed-start-failure.test.ts @@ -39,7 +39,8 @@ describe('a reopened Claude chat whose CLI dies before initialize', () => { await waitForStructuredAgentSessionRecovery() await host.close(SESSION, 'evict') - // Opening starts nothing; the next send spawns the child whose init hangs and tree is unproven. + // Opening starts nothing; the send retries children whose root exits are observed, + // even when descendant cleanup cannot be confirmed. claude.behave(SESSION, { initHangs: true, closeUnproven: true }) const events: AgentSessionSubscribeEvent[] = [] await host.subscribe({ id: 'sub-1', sessionId: SESSION, emit: (event) => events.push(event) }) @@ -61,8 +62,18 @@ describe('a reopened Claude chat whose CLI dies before initialize', () => { expect(sent).toMatchObject({ ok: true, value: { submission: { dispatchState: 'pending' } } }) await vi.waitFor(() => expect(claude.children(SESSION)).toHaveLength(2)) - claude.child(SESSION).exit(scriptedClaudeExitError(DIAGNOSTIC)) - await waitForStructuredAgentSessionRecovery() + for (let count = 2; count <= 6; count++) { + await vi.waitFor(() => expect(claude.children(SESSION)).toHaveLength(count), { + timeout: 10_000 + }) + await vi.waitFor(() => expect(claude.child(SESSION).initializing).toBe(true)) + expect(statusTexts(events)).not.toContain(STARTUP_TEXT) + claude.child(SESSION).exit(scriptedClaudeExitError(DIAGNOSTIC)) + await waitForStructuredAgentSessionRecovery() + await vi.waitFor(() => + expect(host.deps.store.getRecord(SESSION)?.lease.claimStatus).toBe('released') + ) + } await vi.waitFor(() => expect(statusTexts(events)).toContainEqual(STARTUP_TEXT)) const submission = (await host.journalSnapshot(SESSION)).submissions.find( diff --git a/src/main/runtime/claude-structured-send-during-startup.test.ts b/src/main/runtime/claude-structured-send-during-startup.test.ts index 365c702cb2d..bad7b45a207 100644 --- a/src/main/runtime/claude-structured-send-during-startup.test.ts +++ b/src/main/runtime/claude-structured-send-during-startup.test.ts @@ -79,7 +79,7 @@ async function failLatestStart(host: StructuredAgentSessionHost, count: number): } describe('a send into a Claude chat whose CLI keeps failing at startup', () => { - it('restarts once, rejects the message held for it with the diagnostic when that start dies too, then delivers once the CLI is healthy', async () => { + it('exhausts five starts and rejects the held message with the diagnostic, then delivers once the CLI is healthy', async () => { claude.behave(SESSION, { initHangs: true }) const host = await claude.install() await expect(host.attach(CALLER, claude.attachParams(SESSION, null))).resolves.toMatchObject({ @@ -95,7 +95,10 @@ describe('a send into a Claude chat whose CLI keeps failing at startup', () => { expect(host.deps.store.getRecord(SESSION)?.lease.claimStatus).toBe('live') ) expect(await written(host)).toBe(false) - await failLatestStart(host, 2) + for (let count = 2; count <= 6; count++) { + expect(await statusRows(host)).toEqual([STARTUP_TEXT]) + await failLatestStart(host, count) + } await vi.waitFor(async () => expect(await submission(host, held)).toMatchObject({ @@ -105,14 +108,14 @@ describe('a send into a Claude chat whose CLI keeps failing at startup', () => { }) ) expect(await statusRows(host)).toEqual([STARTUP_TEXT, STARTUP_TEXT]) - // Restart acquisition and exit release each advance the fence. - expect(fence(host)).toBe(releasedFence + 2) - expect(claude.children(SESSION)).toHaveLength(2) + // Each of the five attempts acquires and then releases its fence. + expect(fence(host)).toBe(releasedFence + 10) + expect(claude.children(SESSION)).toHaveLength(6) claude.behave(SESSION, {}) await send(host, 'hello again') - await vi.waitFor(() => expect(claude.children(SESSION)).toHaveLength(3)) - await vi.waitFor(() => expect(fence(host)).toBe(releasedFence + 3)) + await vi.waitFor(() => expect(claude.children(SESSION)).toHaveLength(7)) + await vi.waitFor(() => expect(fence(host)).toBe(releasedFence + 11)) await vi.waitFor(() => expect(claude.child(SESSION).calls).toContain('send')) await vi.waitFor(() => expect(host.deps.store.getRecord(SESSION)?.lease.claimStatus).toBe('live') diff --git a/src/main/runtime/claude-structured-send-restart-dies-before-dispatch.test.ts b/src/main/runtime/claude-structured-send-restart-dies-before-dispatch.test.ts index 51457523cb1..f5ed2d9cbfe 100644 --- a/src/main/runtime/claude-structured-send-restart-dies-before-dispatch.test.ts +++ b/src/main/runtime/claude-structured-send-restart-dies-before-dispatch.test.ts @@ -122,7 +122,7 @@ function received(events: AgentSessionSubscribeEvent[]) { } describe('a send whose restarted Claude child dies before it proves its start', () => { - it('settles rejected with the diagnostic, keeps one failure row, and a Retry is one new attempt', async () => { + it('settles rejected with the diagnostic, keeps one failure row, and a new send can start after exhaustion', async () => { claude.behave(SESSION, { initHangs: true }) const host = await claude.install() await expect(host.attach(CALLER, claude.attachParams(SESSION, null))).resolves.toMatchObject({ @@ -137,7 +137,10 @@ describe('a send whose restarted Claude child dies before it proves its start', await eventually(() => expect(claude.child(SESSION).initializing).toBe(true)) expect(await submission(host, sent)).toMatchObject({ dispatchState: 'pending' }) expect(claude.child(SESSION).calls).not.toContain('send') - await failLatestStart(host, 2) + for (let count = 2; count <= 6; count++) { + expect(await statusRows(host)).toEqual([STARTUP_TEXT]) + await failLatestStart(host, count) + } await eventually(async () => expect(await submission(host, sent)).toMatchObject({ @@ -147,12 +150,12 @@ describe('a send whose restarted Claude child dies before it proves its start', }) ) expect(await statusRows(host)).toEqual([STARTUP_TEXT, STARTUP_TEXT]) - expect(fence(host)).toBe(releasedFence + 2) - expect(claude.children(SESSION)).toHaveLength(2) + expect(fence(host)).toBe(releasedFence + 10) + expect(claude.children(SESSION)).toHaveLength(6) claude.behave(SESSION, {}) await send(host, 'hello again') - await eventually(() => expect(claude.children(SESSION)).toHaveLength(3)) + await eventually(() => expect(claude.children(SESSION)).toHaveLength(7)) await eventually(() => expect(claude.child(SESSION).calls).toContain('send')) expect(claude.child(SESSION).calls.filter((call) => call === 'send')).toHaveLength(1) expect(await statusRows(host)).toHaveLength(2) @@ -200,7 +203,10 @@ describe('a send whose restarted Claude child dies before it proves its start', try { await failLatestStart(host, 1) const sent = await send(host, 'hello?') - await failLatestStart(host, 2) + for (let count = 2; count <= 6; count++) { + expect(await statusRows(host)).toEqual([STARTUP_TEXT]) + await failLatestStart(host, count) + } await eventually(() => { const seen = received(events) @@ -234,7 +240,7 @@ describe('a chat whose Claude CLI keeps failing to start, seen by a subscriber o return rows } - it('shows one row naming the cause per failed attempt, however the start died, and none once the CLI is fixed', async () => { + it('shows one row naming the cause per exhausted start, however the start died, and none once the CLI is fixed', async () => { claude.behave(SESSION, { initHangs: true }) const host = await claude.install() const events: AgentSessionSubscribeEvent[] = [] @@ -253,7 +259,10 @@ describe('a chat whose Claude CLI keeps failing to start, seen by a subscriber o await eventually(() => expect([...shownRows(events).values()]).toEqual([STARTUP_FAILURE])) const sent = await send(host, 'hello?') - await failLatestStart(host, 2) + for (let count = 2; count <= 6; count++) { + expect(await statusRows(host)).toEqual([STARTUP_TEXT]) + await failLatestStart(host, count) + } await eventually(async () => expect(await submission(host, sent)).toMatchObject({ dispatchState: 'rejected', diff --git a/src/main/runtime/orchestration/chat-worker-request-failures.test.ts b/src/main/runtime/orchestration/chat-worker-request-failures.test.ts index f49c55ac1ff..90df39bd314 100644 --- a/src/main/runtime/orchestration/chat-worker-request-failures.test.ts +++ b/src/main/runtime/orchestration/chat-worker-request-failures.test.ts @@ -326,7 +326,8 @@ it('keeps each request’s own saved error when another start fails before the d ) await recordStructuredAgentSessionStartFailure( { journal: r.journal, fence: 0 }, - { startKey: 'B-start', ...refused } + { startKey: 'B-start', ...refused }, + 'B' ) await tick() expect(r.mail().map((message) => JSON.parse(message.payload ?? '{}'))).toMatchObject([ diff --git a/src/main/runtime/structured-agent-session-adapter-start-failure.test.ts b/src/main/runtime/structured-agent-session-adapter-start-failure.test.ts index fd2a38dab5d..8807734636b 100644 --- a/src/main/runtime/structured-agent-session-adapter-start-failure.test.ts +++ b/src/main/runtime/structured-agent-session-adapter-start-failure.test.ts @@ -67,7 +67,7 @@ async function failedStart(agent: 'pi' | 'codex' | 'grok', script: ScriptedAgent } ) params.envelope.clientOperationId = operationId() - const attached = await host.attach(caller, params) + const attached = await host.create(caller, params) if (!attached.ok) { throw new Error(JSON.stringify(attached.refusal)) } @@ -89,7 +89,10 @@ async function failedStart(agent: 'pi' | 'codex' | 'grok', script: ScriptedAgent throw new Error(JSON.stringify(sent.refusal)) } await vi.waitFor(() => expect(script.spawns()).toBe(1)) - script.failHandshake?.('provider config is invalid: bad key') + for (let attempt = 1; attempt <= 5; attempt++) { + await vi.waitFor(() => expect(script.spawns()).toBe(attempt)) + script.failHandshake('provider config is invalid: bad key') + } const submission = async () => { await host.flushStreamedEvents(SESSION) return (await host.journalSnapshot(SESSION)).submissions.find( @@ -114,7 +117,7 @@ async function failedStart(agent: 'pi' | 'codex' | 'grok', script: ScriptedAgent ) ).toHaveLength(1) expect(script.prompts()).toEqual([]) - expect(script.spawns()).toBe(1) + expect(script.spawns()).toBe(5) } it('Pi records one failure without dispatch after a failed start', async () => { diff --git a/src/renderer/src/components/native-chat/structured-agent-session-delivery-notices.test.ts b/src/renderer/src/components/native-chat/structured-agent-session-delivery-notices.test.ts index b4fc9204113..080faa3ca2f 100644 --- a/src/renderer/src/components/native-chat/structured-agent-session-delivery-notices.test.ts +++ b/src/renderer/src/components/native-chat/structured-agent-session-delivery-notices.test.ts @@ -138,7 +138,14 @@ describe('the line under each of the chat own messages', () => { const facts = structuredAgentSessionStartFailureFacts([statusRow(startRowKey, startFailed)]) const recorded = (id: string) => row(id, 'rejected', { reason: 'Written by the host.', rejection: startFailed }) - expect(texts(notices({ submissions: [recorded('first')], startFailures: facts }))).toEqual({ + expect( + texts( + notices({ + submissions: [recorded('first'), row('later', 'pending')], + startFailures: facts + }) + ) + ).toEqual({ [agentJournalSubmissionKey('first')]: 'Your message was not sent.' }) expect(texts(notices({ submissions: [recorded('first')] }))).toEqual({