diff --git a/config/scripts/vitest-sqlite-runtime-files.mjs b/config/scripts/vitest-sqlite-runtime-files.mjs index a1cad1a8e123..ba34a82c73b2 100644 --- a/config/scripts/vitest-sqlite-runtime-files.mjs +++ b/config/scripts/vitest-sqlite-runtime-files.mjs @@ -2,6 +2,8 @@ import { SQLITE_DATABASE_RUNTIME_INCLUDE } from './vitest-sqlite-database-runtim // Real SQLite fixtures keep Node publication and close semantics. export const SQLITE_RUNTIME_INCLUDE = [ + 'src/main/pi/rpc-session-adapter.test.ts', + 'src/main/pi/rpc-session-startup.test.ts', 'src/main/acp/acp-structured-publish-at-spawn.test.ts', 'tests/e2e/native-chat-startup-option-pickers.unit.test.ts', ...SQLITE_DATABASE_RUNTIME_INCLUDE, @@ -255,6 +257,10 @@ export const SQLITE_RUNTIME_INCLUDE = [ 'src/main/runtime/runtime-client-agent-launch-settings-durability.test.ts', 'src/main/runtime/structured-agent-session-codex-scripted-child.test.ts', 'src/main/runtime/structured-agent-session-codex-start-deadline.test.ts', + 'src/main/runtime/structured-agent-session-pi-startup.test.ts', + 'src/main/runtime/structured-agent-session-pi-signed-out-start.test.ts', + 'src/main/runtime/structured-agent-session-adapter-start-failure.test.ts', + 'src/main/runtime/structured-agent-session-provider-operations-while-starting.test.ts', 'src/main/runtime/structured-agent-session-codex-settlement-order.test.ts', 'src/main/runtime/structured-agent-session-integration-replay.test.ts', 'src/main/runtime/structured-agent-session-integration.test.ts', diff --git a/src/main/jsonl-rpc/agent-connection.ts b/src/main/jsonl-rpc/agent-connection.ts index 551beb40dec6..a194660f7b70 100644 --- a/src/main/jsonl-rpc/agent-connection.ts +++ b/src/main/jsonl-rpc/agent-connection.ts @@ -1,5 +1,6 @@ import type { PipedProcessSpawner } from '@orca/process-host/process-spec' import { spawnProcess } from '@orca/process-host' +import { providerDiagnostic, withProviderDiagnostic } from '../../shared/agent-session-failure' import { spawnManagedProviderProcess, type ManagedProviderProcess, @@ -81,10 +82,14 @@ export class JsonlRpcAgentConnection { managed.child.stderr.on('error', this.onError) managed.onExit((exit) => { clearTimeout(this.streamExitTimer) - const error = new Error( - exit.processless - ? 'Agent process could not start' - : `Agent process exited (code ${exit.code ?? 'none'}, signal ${exit.signal ?? 'none'})` + // The agent's last stderr lines are what a person can act on when it dies. + const error = withProviderDiagnostic( + new Error( + exit.processless + ? 'Agent process could not start' + : `Agent process exited (code ${exit.code ?? 'none'}, signal ${exit.signal ?? 'none'})` + ), + providerDiagnostic(managed.stderrTail(), 'person') ) this.peer.finishOnInputEnd(error) const stdout = managed.child.stdout 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 a5ec23fccc3a..c64908eaeb0f 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 @@ -136,9 +136,6 @@ export class StructuredAgentSessionAdapterRouter implements StructuredAgentSessi this.liveOwnerOrNull(sessionId)?.holdsLiveProviderProcess?.(sessionId, acquisitionGeneration) ?? false - startUnavailable: NonNullable = (sessionId) => - this.liveOwnerOrNull(sessionId)?.startUnavailable?.(sessionId) - stopEndsSession = (sessionId: string): boolean => this.liveOwnerOrNull(sessionId)?.stopEndsSession?.(sessionId) ?? false diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts index ae36ca266bd8..331d9951c689 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts @@ -1,7 +1,4 @@ -import type { - AgentSessionAccountKind, - AgentSessionUnavailable -} from '../../../shared/agent-session-availability' +import type { AgentSessionAccountKind } from '../../../shared/agent-session-availability' import type { AgentSessionRewindReason, AgentSessionRewindSupport @@ -368,9 +365,6 @@ export type StructuredAgentSessionAdapter = StructuredAgentSessionAdapterStop & /** The adapter's own child for this exact acquisition has a pid and its root exit has not been * seen: first-hand proof of life for lease renewal. Absent or false falls back to a PID probe. */ holdsLiveProviderProcess?(sessionId: string, acquisitionGeneration: string): boolean - /** Why the session's running child can take no turn, as its own start found; undefined when - * unknown or no child runs. Lives and dies with that child. */ - startUnavailable?(sessionId: string): AgentSessionUnavailable | undefined /** The `/` surface the running provider reports for itself. Undefined when the * provider never reports one, which is what keeps the client on its catalog. */ readCommands?(sessionId: string): AgentSessionSlashCommand[] | undefined 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 dd932c95676b..efb568dbc13f 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 @@ -94,7 +94,6 @@ export function createStructuredAgentSessionConversationDelivery(input: { }, providerHoldsDispatch: () => deps.adapter.holdsDispatch?.(sessionId) === true }, - startUnavailable: () => deps.adapter.startUnavailable?.(sessionId), launchPermission: async () => { const record = deps.store.getRecord(sessionId) return record ? structuredAgentSessionLaunchPermission(deps, record) : null @@ -205,7 +204,7 @@ export function createStructuredAgentSessionConversationDelivery(input: { * A rewind found prepared when the conversation opens was started under a child this process no * longer has — the open runs only when none is indexed — so nothing will finish it, and left alone * it refuses every send until a view attaches. Settled here instead of by a start inside - * acceptance. A Codex rewind only its provider can prove stays for the attach. + * acceptance. A Codex rewind only its provider can prove stays for the next proven start. */ async function settleInterruptedCommands( deps: StructuredAgentSessionHostDeps, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-outgrown-child.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-outgrown-child.test.ts index 470e740a043c..216c6fa72371 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-outgrown-child.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-outgrown-child.test.ts @@ -28,7 +28,6 @@ import type { AgentJournalTurnLifecycle } from '../../../shared/agent-session-journal-types' import type { AgentChildWorkView } from '../../../shared/agent-status-child-work-view' -import type { AgentSessionUnavailable } from '../../../shared/agent-session-availability' import type { JournalItemLinkageVisitor } from '../agent-session-journal/journal-store-contracts' import type { StructuredAgentSessionChildWorkReads } from './structured-agent-session-idle-sweep' @@ -69,17 +68,6 @@ describe('a send after the agent said it is not signed in', () => { expect(rig.adapter.acquire).toHaveBeenCalledTimes(2) }) - // A Pi that started with no model keeps a placeholder: a sign-in made since reaches a new Pi only. - it('goes to a new agent when the running one started signed out', async () => { - await foundRestTestChat(rig) - rig.adapter.startUnavailable.mockReturnValueOnce({ reason: 'notSignedIn' }) - rig.adapter.closeSession.mockClear() - expect((await rig.host.send(CALLER, restTestSend('signed in since', fence()))).ok).toBe(true) - await vi.waitFor(() => expect(rig.adapter.dispatch).toHaveBeenCalledTimes(2)) - expect(rig.adapter.closeSession).toHaveBeenCalledWith(SESSION) - expect(rig.adapter.acquire).toHaveBeenCalledTimes(2) - }) - it('keeps the agent after any other rejection', async () => { await foundRestTestChat(rig) rig.adapter.dispatch.mockImplementationOnce(async () => ({ @@ -180,14 +168,6 @@ describe('structuredAgentSessionChildReportedSignedOut', () => { expect(reported({ items, child: childAt({ phase: 'starting' }) })).toBe(false) expect(reported({ items, child: null })).toBe(false) }) - - it("reads the running child's own start, and only a signed-out one", () => { - const fromStart = (startUnavailable: AgentSessionUnavailable, child = childAt({})) => - structuredAgentSessionChildReportedSignedOut(conversation({ child }), startUnavailable) - expect(fromStart({ reason: 'notSignedIn' })).toBe(true) - expect(fromStart({ reason: 'cliMissing' })).toBe(false) - expect(fromStart({ reason: 'notSignedIn' }, childAt({ phase: 'starting' }))).toBe(false) - }) }) describe('retireOutgrownStructuredAgentSessionChild', () => { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-outgrown-child.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-outgrown-child.ts index 77b90764aabd..03f3a32e0c70 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-outgrown-child.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-outgrown-child.ts @@ -1,7 +1,6 @@ // A child the chat outgrew (signed out, or launched with another permission mode) is replaced before // the next send: agents read their login and permission mode only when they start. -import type { AgentSessionUnavailable } from '../../../shared/agent-session-availability' import { readWholeAgentSessionFailureFact } from '../../../shared/agent-session-failure' import { isRootAgentJournalItem } from '../../../shared/agent-session-journal-producer' import type { AgentJournalSubmission } from '../../../shared/agent-session-journal-types' @@ -28,20 +27,15 @@ type OutgrownChildReading = Pick & { const signedOut = (failure: unknown): boolean => readWholeAgentSessionFailureFact(failure)?.kind === 'notSignedIn' -/** The running child itself said it is not signed in: its start, a send it rejected, or a row it - * wrote. Each child holds its own fence, so matching it scopes both to this child; a new one has - * nothing. `startUnavailable` is the adapter's read of that child's start. */ +/** The running child itself said it is not signed in through a rejected send or a row it wrote. + * Its fence scopes both to this child, so a new one has nothing. */ export function structuredAgentSessionChildReportedSignedOut( - session: OutgrownChildReading, - startUnavailable?: AgentSessionUnavailable + session: OutgrownChildReading ): boolean { const { child, journal } = session if (!child || child.phase === 'starting' || child.close) { return false } - if (startUnavailable?.reason === 'notSignedIn') { - return true - } if ( journal .submissions() @@ -99,7 +93,6 @@ export async function retireOutgrownStructuredAgentSessionChild( session: OutgrownChildReading | undefined, deps: { work: StructuredAgentSessionChildWorkReads - startUnavailable?: () => AgentSessionUnavailable | undefined /** The mode a start of the chat would launch with now. */ launchPermission?: () => Promise stopAgent: (sessionId: string) => Promise @@ -109,7 +102,7 @@ export async function retireOutgrownStructuredAgentSessionChild( if ( !session || !( - structuredAgentSessionChildReportedSignedOut(session, deps.startUnavailable?.()) || + structuredAgentSessionChildReportedSignedOut(session) || (await launchedWithOtherPermissionMode(session.child, deps.launchPermission)) ) || structuredAgentSessionChildHasOpenWork(session.journal, deps.work) || diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-rest-test-rig.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-rest-test-rig.ts index 21e35eebb9a8..37e13a2c71e4 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-rest-test-rig.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-rest-test-rig.ts @@ -50,7 +50,6 @@ export type RestTestAdapter = { NonNullable > holdsDispatch: Mock> - startUnavailable: Mock> readOptions: Mock> } @@ -166,7 +165,6 @@ export async function createRestTestRig( dispatch: vi.fn(async () => acceptedDispatch()), acknowledgeSessionRelease: vi.fn(), holdsDispatch: vi.fn(() => false), - startUnavailable: vi.fn(() => undefined), readOptions: vi.fn(async () => ({ models: [], current: { model: 'gpt-live' } })) } const starts = structuredAgentSessionHeldStarts(() => rig.host) diff --git a/src/main/pi/pi-scripted-child.test-fixture.ts b/src/main/pi/pi-scripted-child.test-fixture.ts new file mode 100644 index 000000000000..5c6c4d38dec1 --- /dev/null +++ b/src/main/pi/pi-scripted-child.test-fixture.ts @@ -0,0 +1,266 @@ +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { providerDiagnostic, withProviderDiagnostic } from '../../shared/agent-session-failure' +import type { JsonlRpcAgentConnectionOptions } from '../jsonl-rpc/agent-connection' +import { JsonlRpcResponseError, type JsonlRpcRecord } from '../jsonl-rpc/peer' +import type { ProviderProcessLaunch } from '../provider-process/provider-process-launch' +import type { ScriptedAgentChild } from '../runtime/structured-agent-scripted-child.test-fixture' +import type { PiRpcConnection } from './rpc-session' + +class ScriptedPiConnection implements PiRpcConnection { + closed = false + rootVerdict: PiRpcConnection['rootVerdict'] = 'live' + processless = false + lastCloseResult: PiRpcConnection['lastCloseResult'] = null + readonly pid: number + readonly file: string + private readonly handshake = Promise.withResolvers() + private readonly exitListeners = new Set<() => void>() + private firstStateRead = true + private streaming = false + private model = { provider: 'scripted', id: 'model-1', reasoning: true } + private effort = 'medium' + + constructor( + launch: ProviderProcessLaunch, + private readonly handlers: JsonlRpcAgentConnectionOptions, + private readonly script: ScriptedAgentChild, + private readonly received: string[], + private readonly wire: string[], + private readonly onClosed: () => void, + number: number, + /** A Pi with no login lists no model. */ + private readonly signedOut = false + ) { + this.pid = 42_000 + number + const sessionIndex = launch.args.indexOf('--session') + this.file = + sessionIndex !== -1 + ? launch.args[sessionIndex + 1] + : join(tmpdir(), 'orca-scripted-pi', `session-${number}.jsonl`) + void this.handshake.promise.catch(() => undefined) + } + + async request(command: string, params: Record = {}): Promise { + if (this.closed) { + throw new Error('Scripted Pi connection closed') + } + this.wire.push(command) + if (command === 'get_state') { + if (this.firstStateRead) { + this.firstStateRead = false + if (!this.script.holdHandshakes) { + this.releaseHandshake() + } + return this.handshake.promise + } + return this.state() + } + if (command === 'get_available_models') { + return { models: this.signedOut ? [] : [this.model] } + } + if (command === 'get_commands') { + return { commands: [] } + } + if ( + command === 'set_model' && + typeof params.provider === 'string' && + typeof params.modelId === 'string' + ) { + this.model = { ...this.model, provider: params.provider, id: params.modelId } + return this.model + } + if (command === 'set_thinking_level' && typeof params.level === 'string') { + this.effort = params.level + return {} + } + if (command === 'abort') { + this.endTurn('aborted') + return {} + } + throw new Error(`Unscripted Pi request: ${command}`) + } + + async send(frame: JsonlRpcRecord): Promise { + if (this.closed || frame.type !== 'prompt' || typeof frame.message !== 'string') { + throw new Error('Scripted Pi requires a live prompt') + } + this.received.push(frame.message) + this.wire.push(`prompt:${frame.message}`) + if (this.signedOut) { + this.emit({ + type: 'response', + command: 'prompt', + success: false, + error: 'No API key found for scripted' + }) + return + } + if (!this.streaming) { + this.streaming = true + this.emit({ type: 'agent_start' }) + } + this.emit({ type: 'message_end', message: { role: 'user', content: frame.message } }) + this.emit({ + type: 'response', + command: 'prompt', + success: true, + data: { disposition: 'started' } + }) + if (this.script.completeTurns) { + this.endTurn('stop') + } + } + + releaseHandshake(): void { + if (!this.closed) { + this.handshake.resolve(this.state()) + } + } + + failHandshake(message: string): void { + const error = withProviderDiagnostic( + new JsonlRpcResponseError('get_state', message), + providerDiagnostic(message, 'person') + ) + this.handshake.reject(error) + this.exit(error) + } + + async close(): Promise>> { + if (!this.closed) { + this.onClosed() + const error = new Error('Scripted Pi closed') + this.handshake.reject(error) + this.exit(error) + } + return { root: 'exited', tree: 'exited' } + } + + onExit(listener: () => void): void { + if (this.rootVerdict === 'exited') { + listener() + } else { + this.exitListeners.add(listener) + } + } + + pauseReading(): void {} + resumeReading(): void {} + + private exit(error: Error): void { + if (this.closed) { + return + } + this.closed = true + this.rootVerdict = 'exited' + this.lastCloseResult = { root: 'exited', tree: 'exited' } + this.exitListeners.forEach((listener) => listener()) + this.exitListeners.clear() + this.handlers.onExit?.(error, { + expected: false, + exit: { code: 0, signal: null, processless: false } + }) + } + + private endTurn(stopReason: string): void { + if (!this.streaming) { + return + } + this.streaming = false + this.emit({ + type: 'message_end', + message: { role: 'assistant', content: 'Scripted Pi reply', stopReason } + }) + this.emit({ type: 'agent_end' }) + this.emit({ type: 'agent_settled' }) + } + + private emit(frame: JsonlRpcRecord): void { + this.handlers.onRecord?.(frame) + } + + private state() { + return { + sessionFile: this.file, + model: this.signedOut ? null : this.model, + thinkingLevel: this.effort, + isStreaming: this.streaming, + isCompacting: false, + pendingMessageCount: 0 + } + } +} + +export type ScriptedPiChild = ScriptedAgentChild & { + /** Every request command and prompt any spawn received, in order. */ + wire(): readonly string[] + signIn(): void +} + +export const piScriptedChild = (options: { signedOut?: boolean } = {}): ScriptedPiChild => { + const children: ScriptedPiConnection[] = [] + const received: string[] = [] + const wire: string[] = [] + let resumed = 0 + let closed = 0 + let signedOut = options.signedOut ?? false + const script: ScriptedPiChild = { + holdHandshakes: true, + completeTurns: true, + deps: { + resolvePiLaunch: async ({ identity }) => { + const handle = identity.providerHandle + if (handle && (handle.transport !== 'jsonl-rpc' || handle.agent !== 'pi')) { + throw new Error('Scripted Pi received another provider handle') + } + return { + command: join(tmpdir(), 'orca-scripted-pi', 'fake-pi'), + cwd: tmpdir(), + fullAccess: true, + previous: handle + ? { + linkId: 'scripted-previous', + handle, + origin: 'created', + mintedAtFence: 0, + observedAt: Date.now() + } + : null, + ...(handle ? { sessionFile: handle.nativeId } : {}) + } + }, + openPiConnection: (launch, handlers) => { + if (launch.args.includes('--session')) { + resumed += 1 + } + const child = new ScriptedPiConnection( + launch, + handlers, + script, + received, + wire, + () => { + closed += 1 + }, + children.length + 1, + signedOut + ) + children.push(child) + return child + }, + readProcessStartTime: async () => 1_700_000_000_000 + }, + releaseHandshake: () => children.at(-1)?.releaseHandshake(), + failHandshake: (message) => children.at(-1)?.failHandshake(message), + prompts: () => received, + spawns: () => children.length, + resumes: () => resumed, + closes: () => closed, + wire: () => wire, + signIn: () => { + signedOut = false + } + } + return script +} diff --git a/src/main/pi/rpc-configured-default.test.ts b/src/main/pi/rpc-configured-default.test.ts index 467417c8eadd..3091f3096f7b 100644 --- a/src/main/pi/rpc-configured-default.test.ts +++ b/src/main/pi/rpc-configured-default.test.ts @@ -210,10 +210,21 @@ async function piHost() { options: opts.saved ?? {}, events: rig.eventSink }) - if (!acquired.catalogListing) { + const started = await adapter + .open({ + sessionId, + fence: 1, + acquisitionGeneration: acquired.acquisitionGeneration ?? null, + signal: new AbortController().signal + }) + .catch(async (error: unknown) => { + await adapter.closeSession(sessionId) + throw error + }) + if (!started.catalogListing) { throw new Error('a Pi start hands the host its listing') } - catalog.recordLiveListing(sessionId, acquired.catalogListing) + catalog.recordLiveListing(sessionId, started.catalogListing) if (opts.afterStart && child) { await opts.afterStart({ child, adapter, sessionId }) } @@ -224,7 +235,7 @@ async function piHost() { } } await read() - return { listing: acquired.catalogListing, read } + return { listing: started.catalogListing, read } } return { catalog, chat, store, clock } } @@ -301,12 +312,12 @@ describe('Pi’s default comes from what a chat with no pick runs', () => { const { catalog, chat } = await piHost() const clean = workspace('clean') await catalog.read({ agent: 'pi', waitForListing: true }) - expect((await chat(clean, { available: [] })).listing).not.toHaveProperty('configuredDefault') + await expect(chat(clean, { available: [] })).rejects.toMatchObject({ reason: 'notSignedIn' }) expect(firstFrame(await catalog.read({ agent: 'pi', workspacePath: clean }))).toEqual(NOTHING) // Nor does it forget one a signed-in chat taught. await chat(clean) - await chat(clean, { available: [] }) + await expect(chat(clean, { available: [] })).rejects.toMatchObject({ reason: 'notSignedIn' }) expect(firstFrame(await catalog.read({ agent: 'pi', workspacePath: clean }))).toEqual({ model: 'openai/gpt-6', effort: 'high' diff --git a/src/main/pi/rpc-options.test.ts b/src/main/pi/rpc-options.test.ts index 5beaefb54ca1..d15c602cf550 100644 --- a/src/main/pi/rpc-options.test.ts +++ b/src/main/pi/rpc-options.test.ts @@ -70,8 +70,8 @@ describe('Pi RPC session options', () => { effort: 'high' }) expect(rpc.request.mock.calls).toEqual([ - ['set_model', { provider: 'openai-codex', modelId: 'gpt-6.1-sol' }], - ['set_thinking_level', { level: 'high' }] + ['set_model', { provider: 'openai-codex', modelId: 'gpt-6.1-sol' }, {}], + ['set_thinking_level', { level: 'high' }, {}] ]) expect(() => parsePiModelOptionId('openai-codex')).toThrow() await expect(applyPiRpcSessionOption(rpc, selected, 'model', '/gpt-6.1-sol')).rejects.toThrow() diff --git a/src/main/pi/rpc-options.ts b/src/main/pi/rpc-options.ts index 4f75e064c6e1..b881d31f1729 100644 --- a/src/main/pi/rpc-options.ts +++ b/src/main/pi/rpc-options.ts @@ -92,17 +92,18 @@ export async function applyPiRpcSessionOption( rpc: PiRpcRequester, selected: Map, key: string, - value: string + value: string, + options: { timeoutMs?: number | null } = {} ): Promise>> { if (key === 'model') { const model = parsePiModelOptionId(value) modelSchema.parse({ provider: model.provider, id: model.modelId }) - await rpc.request('set_model', model) + await rpc.request('set_model', model, options) } else if (key === 'effort') { if (!THINKING_LEVELS.some((level) => level === value)) { throw new Error(`Unsupported Pi thinking level: ${value}`) } - await rpc.request('set_thinking_level', { level: value }) + await rpc.request('set_thinking_level', { level: value }, options) } else { throw new Error(`Unsupported Pi option: ${key}`) } diff --git a/src/main/pi/rpc-runtime-registration.ts b/src/main/pi/rpc-runtime-registration.ts index a8952bcb417d..b273e580078f 100644 --- a/src/main/pi/rpc-runtime-registration.ts +++ b/src/main/pi/rpc-runtime-registration.ts @@ -25,11 +25,13 @@ import { PI_SESSION_HISTORY } from './pi-session-history' function createPiRpcAdapter(context: StructuredAgentAdapterContext): StructuredAgentRuntimeAdapter { const { deps } = context return new PiRpcSessionAdapter({ - resolveLaunch: composeStructuredLaunch( - PI_RPC_AGENT, - structuredAgentLaunchSources(context), - piRpcLaunchPart({}) - ), + resolveLaunch: + deps.resolvePiLaunch ?? + composeStructuredLaunch( + PI_RPC_AGENT, + structuredAgentLaunchSources(context), + piRpcLaunchPart({}) + ), ...(deps.openPiConnection ? { openConnection: deps.openPiConnection } : {}), ...(deps.readProcessStartTime ? { readProcessStartTime: deps.readProcessStartTime } : {}), onLifecycle: context.deliverLifecycle, diff --git a/src/main/pi/rpc-session-adapter.test-fixture.ts b/src/main/pi/rpc-session-adapter.test-fixture.ts new file mode 100644 index 000000000000..10810ec4abf0 --- /dev/null +++ b/src/main/pi/rpc-session-adapter.test-fixture.ts @@ -0,0 +1,224 @@ +import { expect, vi, type Mock } from 'vitest' +import type { AgentSessionProcessIdentity } from '../../shared/agent-session-record' +import type { StructuredAgentSessionAcquireInput } from '../native-chat/agent-session-wire/structured-agent-session-adapter' +import { + closeProviderTimelineRigs, + openProviderTimelineRig +} from '../native-chat/agent-session-timeline/provider-timeline-assembler-test-support' +import type { JsonlRpcAgentConnectionOptions } from '../jsonl-rpc/agent-connection' +import type { JsonlRpcRecord } from '../jsonl-rpc/peer' +import { JsonlRpcResponseError } from '../jsonl-rpc/peer' +import type { PiRpcConnection } from './rpc-session' +import type { ProviderProcessLaunch } from '../provider-process/provider-process-launch' +import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' +import { PiRpcSessionAdapter, type PiRpcSessionAdapterDeps } from './rpc-session-adapter' +import type { PiRpcResolvedLaunch } from './rpc-launch-resolution' + +export const sessionId = 'session-timeline' +export const file = '/host/account/sessions/session.jsonl' +export const state = { + sessionFile: file, + isStreaming: false, + isCompacting: false, + model: { provider: 'anthropic', id: 'model-1' } +} +const cleanups: (() => Promise)[] = [] +export async function closePiRpcAdapterFixtures(): Promise { + for (const cleanup of cleanups.splice(0)) { + await cleanup() + } + await closeProviderTimelineRigs() +} + +export class FakeConnection implements PiRpcConnection { + pid = 4123 + closed = false + rootVerdict: 'live' | 'unverifiable' | 'exited' = 'live' + processless = false + deferExit = false + closeError?: Error + lastCloseResult: PiRpcConnection['lastCloseResult'] = null + closeResult: Awaited> = { root: 'exited', tree: 'exited' } + readonly sent: JsonlRpcRecord[] = [] + readonly requests: string[] = [] + private readonly exitListeners: (() => void)[] = [] + requestOverride?: (command: string) => Promise | undefined + abortRequest?: () => void + constructor(readonly handlers: JsonlRpcAgentConnectionOptions) {} + async request(command: string): Promise { + this.requests.push(command) + const override = this.requestOverride?.(command) + if (override) { + return override + } + if (command === 'get_state') { + return state + } + if (command === 'get_available_models') { + return { models: [state.model] } + } + if (command === 'get_commands') { + return { commands: [{ name: 'help' }] } + } + if (command === 'set_thinking_level') { + throw new JsonlRpcResponseError(command, 'saved effort unavailable') + } + return {} + } + async send(frame: JsonlRpcRecord): Promise { + this.sent.push(frame) + } + async close(): Promise>> { + this.abortRequest?.() + this.closed = true + this.rootVerdict = this.closeResult.root + this.lastCloseResult = this.closeResult + if (this.closeResult.root === 'exited' && !this.deferExit) { + this.exit(this.closeError) + } + return this.closeResult + } + pauseReading(): void {} + resumeReading(): void {} + onExit(listener: () => void): void { + if (this.rootVerdict === 'exited') { + listener() + } else { + this.exitListeners.push(listener) + } + } + receive(frame: JsonlRpcRecord): void { + this.handlers.onRecord?.(frame) + } + exit(error = new Error('child exited')): void { + this.closed = true + this.rootVerdict = 'exited' + for (const listener of this.exitListeners.splice(0)) { + listener() + } + this.handlers.onExit?.(error, { + expected: false, + exit: { code: 1, signal: null, processless: false } + }) + } +} + +type PiLifecycleEvent = Parameters[0] + +/** The first `type` event the adapter reported; a missing one fails the test. */ +export function reportedEvent( + lifecycle: Mock, + type: T +): Extract { + const isType = (event: PiLifecycleEvent): event is Extract => + event.type === type + const found = lifecycle.mock.calls.map(([event]) => event).find(isType) + if (!found) { + throw new Error(`no ${type} event was reported`) + } + return found +} + +export async function setup( + options: Readonly> = {}, + eventSink?: (sink: StructuredAgentSessionEventSink) => StructuredAgentSessionEventSink +) { + const rig = await openProviderTimelineRig({ agent: 'pi', sessionId }) + const connections: FakeConnection[] = [] + const lifecycle = vi.fn() + const settled = vi.fn() + const idle = vi.fn() + const onSpawned = vi.fn(async (_process: AgentSessionProcessIdentity) => { + expect(connections.at(-1)?.requests).toEqual([]) + }) + const input: StructuredAgentSessionAcquireInput = { + identity: { + sessionId, + workspaceId: 'folder-1', + hostId: 'local', + agent: 'pi', + providerHandle: null + }, + fence: 7, + spawnToken: 'spawn-token', + options, + events: eventSink?.(rig.eventSink) ?? rig.eventSink, + onSpawned + } + const resolveLaunch = vi.fn(async (): Promise => ({ + command: '/host/bin/pi', + cwd: '/host/folder', + fullAccess: true, + previous: null + })) + const openConnection = vi.fn( + (_launch: ProviderProcessLaunch, handlers: JsonlRpcAgentConnectionOptions) => { + const connection = new FakeConnection(handlers) + connections.push(connection) + return connection + } + ) + const adapter = new PiRpcSessionAdapter({ + resolveLaunch, + readProcessStartTime: async () => 12345, + openConnection, + onLifecycle: lifecycle, + onSettled: settled, + onIdle: idle, + logger: { warn: vi.fn(), error: vi.fn() } + }) + const acquired = await adapter.acquire(input) + const started = await adapter.open({ + sessionId, + fence: 7, + acquisitionGeneration: acquired.acquisitionGeneration ?? null, + signal: new AbortController().signal + }) + lifecycle.mockClear() + const connection = connections[0] + if (!connection) { + throw new Error('connection missing') + } + cleanups.push(async () => { + await adapter.closeAll().catch(() => {}) + await adapter.drainObservedExits() + adapter.acknowledgeSessionRelease(sessionId) + }) + return { + adapter, + connection, + rig, + acquired, + started, + onSpawned, + lifecycle, + settled, + idle, + input, + resolveLaunch, + openConnection, + connections + } +} + +export async function restart( + h: Awaited>, + configure: (connection: FakeConnection) => void, + input: Partial = {} +) { + await h.adapter.closeSession(sessionId) + await h.adapter.drainObservedExits() + h.lifecycle.mockClear() + h.openConnection.mockImplementationOnce((_launch, handlers) => { + const connection = new FakeConnection(handlers) + configure(connection) + h.connections.push(connection) + return connection + }) + const acquired = await h.adapter.acquire({ ...h.input, fence: 8, ...input }) + const child = h.connections.at(-1) + if (!child) { + throw new Error('child missing') + } + return { acquired, child } +} diff --git a/src/main/pi/rpc-session-adapter.test.ts b/src/main/pi/rpc-session-adapter.test.ts index 8c74e65eea1c..004a21b5e6be 100644 --- a/src/main/pi/rpc-session-adapter.test.ts +++ b/src/main/pi/rpc-session-adapter.test.ts @@ -1,185 +1,17 @@ import { afterEach, describe, expect, it, vi } from 'vitest' import { agentJournalItemKey } from '../../shared/agent-session-journal-item-key' -import type { AgentSessionProcessIdentity } from '../../shared/agent-session-record' import { agentJournalTurnBody } from '../../shared/agent-session-turn-record' -import { - AgentSessionAcquisitionRootExitObservedError, - type StructuredAgentSessionAcquireInput -} from '../native-chat/agent-session-wire/structured-agent-session-adapter' -import { - closeProviderTimelineRigs, - openProviderTimelineRig -} from '../native-chat/agent-session-timeline/provider-timeline-assembler-test-support' -import type { JsonlRpcAgentConnectionOptions } from '../jsonl-rpc/agent-connection' -import type { JsonlRpcRecord } from '../jsonl-rpc/peer' +import { AgentSessionAcquisitionRootExitObservedError } from '../native-chat/agent-session-wire/structured-agent-session-adapter' import { JsonlRpcResponseError } from '../jsonl-rpc/peer' -import type { PiRpcConnection } from './rpc-session' -import type { ProviderProcessLaunch } from '../provider-process/provider-process-launch' -import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' +import { + closePiRpcAdapterFixtures, + file, + sessionId, + setup +} from './rpc-session-adapter.test-fixture' import { agentSessionLaunchFolderMissing } from '../runtime/agent-session-launch-directory' -import { PiRpcSessionAdapter } from './rpc-session-adapter' - -const sessionId = 'session-timeline' -const file = '/host/account/sessions/session.jsonl' -const state = { - sessionFile: file, - isStreaming: false, - isCompacting: false, - model: { provider: 'anthropic', id: 'model-1' } -} -const cleanups: (() => Promise)[] = [] -afterEach(async () => { - for (const cleanup of cleanups.splice(0)) { - await cleanup() - } - await closeProviderTimelineRigs() -}) -class FakeConnection implements PiRpcConnection { - pid = 4123 - closed = false - rootVerdict: 'live' | 'unverifiable' | 'exited' = 'live' - processless = false - deferExit = false - lastCloseResult: PiRpcConnection['lastCloseResult'] = null - closeResult: Awaited> = { root: 'exited', tree: 'exited' } - readonly sent: JsonlRpcRecord[] = [] - readonly requests: string[] = [] - private readonly exitListeners: (() => void)[] = [] - requestOverride?: (command: string) => Promise | undefined - abortRequest?: () => void - constructor(readonly handlers: JsonlRpcAgentConnectionOptions) {} - async request(command: string): Promise { - this.requests.push(command) - const override = this.requestOverride?.(command) - if (override) { - return override - } - if (command === 'get_state') { - return state - } - if (command === 'get_available_models') { - return { models: [state.model] } - } - if (command === 'get_commands') { - return { commands: [{ name: 'help' }] } - } - if (command === 'set_thinking_level') { - throw new Error('saved effort unavailable') - } - return {} - } - async send(frame: JsonlRpcRecord): Promise { - this.sent.push(frame) - } - async close(): Promise>> { - this.abortRequest?.() - this.closed = true - this.rootVerdict = this.closeResult.root - this.lastCloseResult = this.closeResult - if (this.closeResult.root === 'exited' && !this.deferExit) { - this.exit() - } - return this.closeResult - } - pauseReading(): void {} - resumeReading(): void {} - onExit(listener: () => void): void { - if (this.rootVerdict === 'exited') { - listener() - } else { - this.exitListeners.push(listener) - } - } - receive(frame: JsonlRpcRecord): void { - this.handlers.onRecord?.(frame) - } - exit(error = new Error('child exited')): void { - this.closed = true - this.rootVerdict = 'exited' - for (const listener of this.exitListeners.splice(0)) { - listener() - } - this.handlers.onExit?.(error, { - expected: false, - exit: { code: 1, signal: null, processless: false } - }) - } -} - -async function setup( - options: Readonly> = {}, - eventSink?: (sink: StructuredAgentSessionEventSink) => StructuredAgentSessionEventSink -) { - const rig = await openProviderTimelineRig({ agent: 'pi', sessionId }) - const connections: FakeConnection[] = [] - const lifecycle = vi.fn(), - settled = vi.fn(), - idle = vi.fn() - const onSpawned = vi.fn(async (_process: AgentSessionProcessIdentity) => { - expect(connections.at(-1)?.requests).toEqual([]) - }) - const input: StructuredAgentSessionAcquireInput = { - identity: { - sessionId, - workspaceId: 'folder-1', - hostId: 'local', - agent: 'pi', - providerHandle: null - }, - fence: 7, - spawnToken: 'spawn-token', - options, - events: eventSink?.(rig.eventSink) ?? rig.eventSink, - onSpawned - } - const resolveLaunch = vi.fn(async () => ({ - command: '/host/bin/pi', - cwd: '/host/folder', - fullAccess: true, - previous: null - })) - const openConnection = vi.fn( - (_launch: ProviderProcessLaunch, handlers: JsonlRpcAgentConnectionOptions) => { - const connection = new FakeConnection(handlers) - connections.push(connection) - return connection - } - ) - const adapter = new PiRpcSessionAdapter({ - resolveLaunch, - readProcessStartTime: async () => 12345, - openConnection, - onLifecycle: lifecycle, - onSettled: settled, - onIdle: idle, - logger: { warn: vi.fn(), error: vi.fn() } - }) - const acquired = await adapter.acquire(input) - const connection = connections[0] - if (!connection) { - throw new Error('connection missing') - } - cleanups.push(async () => { - await adapter.closeAll().catch(() => {}) - await adapter.drainObservedExits() - adapter.acknowledgeSessionRelease(sessionId) - }) - return { - adapter, - connection, - rig, - acquired, - onSpawned, - lifecycle, - settled, - idle, - input, - resolveLaunch, - openConnection, - connections - } -} +afterEach(closePiRpcAdapterFixtures) describe('Pi RPC session ownership and delivery', () => { it('refuses a floating chat whose folder is gone with the reason the person reads', async () => { @@ -206,36 +38,13 @@ describe('Pi RPC session ownership and delivery', () => { await expect(started).rejects.toThrow('closed while starting') expect(h.connections).toHaveLength(1) await expect(h.adapter.acquire({ ...h.input, fence: 9 })).resolves.toMatchObject({ - link: { origin: 'created' } + process: { pid: 4123 } }) resolving.resolve(launch!) await Promise.resolve() expect(h.connections).toHaveLength(2) }) - it('kills the child and rejects a stalled startup when the host aborts it', async () => { - const h = await setup() - await h.adapter.closeSession(sessionId) - const controller = new AbortController() - const opened = Promise.withResolvers() - const reply = Promise.withResolvers() - h.openConnection.mockImplementationOnce((_launch, handlers) => { - const connection = new FakeConnection(handlers) - connection.requestOverride = (command) => - command === 'get_state' ? reply.promise : undefined - connection.abortRequest = () => reply.reject(new Error('closed while starting')) - h.connections.push(connection) - opened.resolve(connection) - return connection - }) - const started = h.adapter.acquire({ ...h.input, fence: 8, signal: controller.signal }) - const child = await opened.promise - await Promise.resolve() - controller.abort(new Error('Pi closed while starting')) - await expect(started).rejects.toThrow('closed while starting') - expect(child.rootVerdict).toBe('exited') - }) - it('retains final events when exit publication times out under journal backpressure', async () => { let blocked = true const h = await setup({}, (sink) => ({ @@ -278,7 +87,7 @@ describe('Pi RPC session ownership and delivery', () => { await h.adapter.closeSession(sessionId) h.adapter.acknowledgeSessionRelease(sessionId) await expect(h.adapter.acquire({ ...h.input, fence: 8 })).resolves.toMatchObject({ - link: { origin: 'created' } + process: { pid: 4123 } }) let drained = false const delivery = h.adapter.drainObservedExits().then(() => { @@ -310,12 +119,13 @@ describe('Pi RPC session ownership and delivery', () => { spawnToken: 'spawn-token' }) expect(h.acquired.process).toEqual(h.onSpawned.mock.calls[0]?.[0]) - expect(h.acquired.link).toMatchObject({ + expect(h.acquired.link).toBeUndefined() + expect(h.started.link).toMatchObject({ handle: { transport: 'jsonl-rpc', agent: 'pi', nativeId: file }, origin: 'created', mintedAtFence: 7 }) - expect(h.adapter.readOptionRestoreFailures(sessionId)).toEqual(['effort', 'unknown']) + expect(h.adapter.readOptionRestoreFailures(sessionId)).toEqual(['unknown', 'effort']) expect(h.connection.requests).toContain('get_commands') }) @@ -572,29 +382,3 @@ describe('Pi RPC session ownership and delivery', () => { expect(await other.adapter.closeSession(sessionId)).toBe(false) }) }) - -describe('Pi RPC start without a model', () => { - it('reports a start that listed no model as signed out, for that child only', async () => { - const h = await setup() - expect(h.adapter.startUnavailable(sessionId)).toBeUndefined() - await h.adapter.closeSession(sessionId) - h.openConnection.mockImplementationOnce((_launch, handlers) => { - const connection = new FakeConnection(handlers) - // What a signed-out Pi lists (captured from Pi 1.0.4 in __fixtures__/signed-out.jsonl). - connection.requestOverride = (command) => - command === 'get_available_models' ? Promise.resolve({ models: [] }) : undefined - h.connections.push(connection) - return connection - }) - await h.adapter.acquire({ ...h.input, fence: 8 }) - expect(h.adapter.startUnavailable(sessionId)).toEqual({ reason: 'notSignedIn' }) - // Its root gone while its output still drains: that child says nothing more. - const child = h.connections.at(-1)! - child.rootVerdict = 'exited' - expect(child.closed).toBe(false) - expect(h.adapter.startUnavailable(sessionId)).toBeUndefined() - child.rootVerdict = 'live' - await h.adapter.closeSession(sessionId) - expect(h.adapter.startUnavailable(sessionId)).toBeUndefined() - }) -}) diff --git a/src/main/pi/rpc-session-adapter.ts b/src/main/pi/rpc-session-adapter.ts index 139268fe0a78..6dc725194c29 100644 --- a/src/main/pi/rpc-session-adapter.ts +++ b/src/main/pi/rpc-session-adapter.ts @@ -1,6 +1,5 @@ import { randomUUID } from 'node:crypto' import { waitForPromiseWithSignal } from '../../shared/abort-signal-reason' -import type { AgentSessionUnavailable } from '../../shared/agent-session-availability' import { agentSessionFailureFact } from '../../shared/agent-session-failure' import { agentSessionFailureWords } from '../../shared/agent-session-failure-words' import { @@ -16,7 +15,7 @@ import { providerSpawnedProcessIdentity } from '../provider-process/provider-spa import { ProviderAcquisitionStarts } from '../provider-process/provider-acquisition-starts' import { compactPiRpcSession } from './rpc-compaction' import { buildPiRpcLaunch } from './rpc-launch' -import { piRpcProviderLink, type PiRpcResolvedLaunch } from './rpc-launch-resolution' +import type { PiRpcResolvedLaunch } from './rpc-launch-resolution' import { PiRpcSession, type PiRpcSessionDeps, type PiRpcConnection } from './rpc-session' import { PiRpcPromptError, preparePiRpcPrompt } from './rpc-prompt' import { applyPiRpcSessionOption } from './rpc-options' @@ -81,7 +80,7 @@ export class PiRpcSessionAdapter implements StructuredAgentSessionAdapter { } const spec = buildPiRpcLaunch(launch) const fresh = !launch.sessionFile && !launch.forkFile - session = new PiRpcSession(input, randomUUID(), spec, this.deps, fresh) + session = new PiRpcSession(input, randomUUID(), spec, this.deps, launch, fresh) this.sessions.set(id, session) this.starts.track(attempt, session.connection) const spawned = providerSpawnedProcessIdentity( @@ -93,15 +92,13 @@ export class PiRpcSessionAdapter implements StructuredAgentSessionAdapter { await spawned.onSpawned(session.connection.pid) } const process = await spawned.read(session.connection.pid) - const file = await waitForPromiseWithSignal(session.start(), attempt.signal) if (session.connection.closed || attempt.signal.aborted) { throw new Error('Pi exited while starting') } return { process, acquisitionGeneration: session.generation, - link: piRpcProviderLink(launch, file, input.fence, randomUUID(), Date.now()), - ...(session.startListing ? { catalogListing: session.startListing } : {}) + providerChildPhase: 'starting' } } catch (error) { if (!session) { @@ -125,6 +122,19 @@ export class PiRpcSessionAdapter implements StructuredAgentSessionAdapter { } } + open: NonNullable = async (input) => { + const session = this.sessions.get(input.sessionId) + if ( + !session || + session.input.fence !== input.fence || + session.generation !== input.acquisitionGeneration || + session.connection.closed + ) { + throw new Error('Pi child changed before it opened') + } + return session.open(input.signal) + } + dispatch: StructuredAgentSessionAdapter['dispatch'] = async (input) => { const session = this.session(input.sessionId, input.fence) let prompt @@ -227,23 +237,16 @@ export class PiRpcSessionAdapter implements StructuredAgentSessionAdapter { } readOptions: NonNullable = async (input) => this.session(input.sessionId, input.fence).readOptions() - readCommands = (id: string) => this.sessions.get(id)?.commands readOptionRestoreFailures(id: string): readonly string[] { - return this.sessions.get(id)?.skipped ?? [] + return this.sessions.get(id)?.startup.skipped ?? [] } + readCommands = (id: string) => this.sessions.get(id)?.startup.commands holdsDispatch(id: string): boolean { return this.sessions.get(id)?.turns.holdsDispatch ?? false } - /** Started with no model listed: Pi keeps a placeholder, so every prompt this child takes fails - * as not signed in, even after a sign-in, which only a new Pi reads. */ - startUnavailable(id: string): AgentSessionUnavailable | undefined { + holdsLiveProviderProcess(id: string, generation: string): boolean { const session = this.sessions.get(id) - return session && - !session.connection.closed && - session.connection.rootVerdict !== 'exited' && - session.options?.models.length === 0 - ? { reason: 'notSignedIn' } - : undefined + return session?.generation === generation && !session.connection.closed } async closeSession(id: string, requested = true): Promise { @@ -291,7 +294,12 @@ export class PiRpcSessionAdapter implements StructuredAgentSessionAdapter { } private session(id: string, fence: number): PiRpcSession { const session = this.sessions.get(id) - if (!session || session.input.fence !== fence || session.connection.closed) { + if ( + !session || + session.input.fence !== fence || + session.connection.closed || + !session.openedLink + ) { throw new Error('Pi session is not live under this fence') } return session diff --git a/src/main/pi/rpc-session-startup.test.ts b/src/main/pi/rpc-session-startup.test.ts new file mode 100644 index 000000000000..e39c84ec3457 --- /dev/null +++ b/src/main/pi/rpc-session-startup.test.ts @@ -0,0 +1,55 @@ +import { afterEach, describe, expect, it } from 'vitest' +import { + closePiRpcAdapterFixtures, + restart, + sessionId, + setup, + state +} from './rpc-session-adapter.test-fixture' + +afterEach(closePiRpcAdapterFixtures) + +describe('Pi host awaited open', () => { + it('lets the host close a child held in the state request without a late open', async () => { + const h = await setup() + const stateReply = Promise.withResolvers() + const { acquired, child } = await restart(h, (connection) => { + connection.requestOverride = (command) => + command === 'get_state' ? stateReply.promise : undefined + connection.abortRequest = () => stateReply.reject(new Error('closed while starting')) + }) + const controller = new AbortController() + const opening = h.adapter + .open({ + sessionId, + fence: 8, + acquisitionGeneration: acquired.acquisitionGeneration ?? null, + signal: controller.signal + }) + .catch((error: unknown) => error) + expect(child.requests).toEqual(['get_state']) + controller.abort() + await h.adapter.closeSession(sessionId) + expect(await opening).toBeInstanceOf(Error) + expect(child.rootVerdict).toBe('exited') + expect(h.lifecycle).not.toHaveBeenCalledWith(expect.objectContaining({ type: 'started' })) + }) + + it('fails an open if its lane fails while the state request is pending', async () => { + const h = await setup() + const stateReply = Promise.withResolvers() + const { acquired, child } = await restart(h, (connection) => { + connection.requestOverride = (command) => + command === 'get_state' ? stateReply.promise : undefined + }) + const opening = h.adapter.open({ + sessionId, + fence: 8, + acquisitionGeneration: acquired.acquisitionGeneration ?? null, + signal: new AbortController().signal + }) + child.handlers.onClose?.(new Error('lane failed during startup')) + stateReply.resolve(state) + await expect(opening).rejects.toThrow('lane failed during startup') + }) +}) diff --git a/src/main/pi/rpc-session-startup.ts b/src/main/pi/rpc-session-startup.ts new file mode 100644 index 000000000000..b246ea9b9aab --- /dev/null +++ b/src/main/pi/rpc-session-startup.ts @@ -0,0 +1,180 @@ +import { randomUUID } from 'node:crypto' +import { waitForPromiseWithSignal } from '../../shared/abort-signal-reason' +import { providerDiagnostic, withProviderDiagnostic } from '../../shared/agent-session-failure' +import type { + AgentSessionOptionsResult, + AgentSessionSlashCommand +} from '../../shared/agent-session-wire' +import { JsonlRpcResponseError } from '../jsonl-rpc/peer' +import { + AgentSessionAcquisitionRefusal, + type StructuredAgentSessionAcquireInput +} from '../native-chat/agent-session-wire/structured-agent-session-adapter' +import type { StructuredAgentSessionProviderOpenResult } from '../native-chat/agent-session-wire/structured-agent-session-provider-open-contract' +import { + unpickedSessionConfiguredChoice, + withLiveCatalogListing, + type AgentModelCatalogLiveListing +} from '../native-chat/agent-model-catalog/agent-model-catalog-entry' +import { piRpcProviderLink, type PiRpcResolvedLaunch } from './rpc-launch-resolution' +import { + applyPiRpcSessionOption, + piModelOptionId, + readPiRpcCommands, + readPiRpcSessionOptions +} from './rpc-options' +import { piRpcStateSchema, type PiRpcState } from './rpc-protocol' +import type { PiRpcConnection, PiRpcSessionDeps } from './rpc-session' +import { STRUCTURED_AGENT_SESSION_STARTUP_CEILING_MS } from '../native-chat/agent-session-wire/structured-agent-session-startup-attempt-contract' + +/** What the start cannot go on without waits as long as the host's startup attempt does, its only + * deadline; the optional reads keep their own bound and never fail the start. */ +const START_REQUEST = { timeoutMs: STRUCTURED_AGENT_SESSION_STARTUP_CEILING_MS } + +export class PiRpcSessionStartup { + skipped: readonly string[] = [] + commands?: AgentSessionSlashCommand[] + /** What the start listed; a signed-out Pi lists no model. */ + options?: AgentSessionOptionsResult + + constructor( + private readonly input: StructuredAgentSessionAcquireInput, + private readonly connection: PiRpcConnection, + private readonly selected: Map, + private readonly deps: PiRpcSessionDeps, + /** A new Pi session starts on its config's model; a resumed or forked one keeps its own. */ + private readonly resolvesConfig: boolean, + private readonly setState: (state: PiRpcState) => void, + private readonly failedCause: () => Error | undefined + ) {} + + async run( + launch: PiRpcResolvedLaunch, + signal: AbortSignal + ): Promise { + const answer = await this.wait(this.connection.request('get_state', {}, START_REQUEST), signal) + this.assertLive(signal) + const state = piRpcStateSchema.parse(answer) + if (state.model?.provider === 'unknown') { + throw this.notSignedIn() + } + const link = piRpcProviderLink( + launch, + state.sessionFile, + this.input.fence, + randomUUID(), + Date.now() + ) + this.setState(state) + const current: AgentSessionOptionsResult['current'] = { + model: state.model ? piModelOptionId(state.model.provider, state.model.id) : '', + ...(state.thinkingLevel ? { effort: state.thinkingLevel } : {}) + } + const saved = this.input.options ?? {} + const skipped = Object.keys(saved).filter((key) => key !== 'model' && key !== 'effort') + // Keys a restore sent, whether or not Pi took them. + const picked = new Set() + for (const key of ['model', 'effort'] as const) { + const value = saved[key] + if (value === undefined) { + continue + } + picked.add(key) + try { + this.assertLive(signal) + await this.wait( + applyPiRpcSessionOption(this.connection, this.selected, key, value, START_REQUEST), + signal + ) + current[key] = value + } catch (error) { + if (!(error instanceof JsonlRpcResponseError)) { + throw error + } + skipped.push(key) + this.warn('Pi rejected a saved option', 'pi-option-restore', error, { key }) + } + } + this.assertLive(signal) + current.confirmed = [...(current.model ? ['model'] : []), ...(current.effort ? ['effort'] : [])] + // Part of the start, as other agents' handshakes answer with theirs: a Pi that lists no model + // is signed out, and what it started on names its config's default. + const options = await this.readListing(signal) + this.assertLive(signal) + if (options && options.models.length === 0) { + throw this.notSignedIn() + } + // What it started on names its config's default once; a later switch teaches nothing. + let catalogListing: AgentModelCatalogLiveListing | undefined + if (options) { + this.options = options + catalogListing = withLiveCatalogListing( + options, + unpickedSessionConfiguredChoice({ resolvesConfig: this.resolvesConfig, picked, ...options }) + ).catalogListing + } + this.skipped = skipped + await this.readCommands(signal) + this.assertLive(signal) + return { + link, + reportedOptions: options?.current ?? current, + ...(catalogListing ? { catalogListing } : {}) + } + } + + /** Never fails the start: without it the chat only lacks the listing and the sign-in verdict. */ + private async readListing(signal: AbortSignal): Promise { + try { + return await this.wait(readPiRpcSessionOptions(this.connection), signal) + } catch (error) { + this.assertLive(signal) + this.warn('Pi options could not be read', 'pi-options', error) + return undefined + } + } + + private async readCommands(signal: AbortSignal): Promise { + try { + this.assertLive(signal) + const result = await this.wait(readPiRpcCommands(this.connection), signal) + this.assertLive(signal) + this.commands = result.commands + } catch (error) { + this.warn('Pi commands could not be read', 'pi-commands', error) + } + } + + private assertLive(signal: AbortSignal): void { + const failure = this.failedCause() + if (failure) { + throw failure + } + if (signal.aborted || this.connection.closed) { + throw new Error('Pi closed while starting') + } + } + + private wait(promise: Promise, signal: AbortSignal): Promise { + return waitForPromiseWithSignal(promise, signal) + } + + private notSignedIn(): AgentSessionAcquisitionRefusal { + return withProviderDiagnostic( + new AgentSessionAcquisitionRefusal( + 'Pi has no authenticated model provider available.', + 'notSignedIn' + ), + providerDiagnostic('Pi has no authenticated model provider available.', 'person') + ) + } + + private warn(message: string, scope: string, error: unknown, fields = {}): void { + this.deps.logger.warn(message, { + scope, + sessionId: this.input.identity.sessionId, + error, + ...fields + }) + } +} diff --git a/src/main/pi/rpc-session.ts b/src/main/pi/rpc-session.ts index ae693653df4c..406a7cfb08df 100644 --- a/src/main/pi/rpc-session.ts +++ b/src/main/pi/rpc-session.ts @@ -1,7 +1,8 @@ -import type { - AgentSessionOptionsResult, - AgentSessionSlashCommand -} from '../../shared/agent-session-wire' +import { + agentSessionFailureFact, + providerDiagnostic, + providerDiagnosticOf +} from '../../shared/agent-session-failure' import type { StructuredAgentSessionAcquireInput, StructuredAgentSessionLifecycleEvent, @@ -15,15 +16,15 @@ import { type JsonlRpcAgentConnectionOptions } from '../jsonl-rpc/agent-connection' import type { ProviderProcessLaunch } from '../provider-process/provider-process-launch' +import { JsonlRpcResponseError } from '../jsonl-rpc/peer' import { PiRpcTurns } from './rpc-turns' import { PiRpcDialogCallbacks } from './rpc-dialog-callbacks' -import { piRpcStateSchema } from './rpc-protocol' -import { applyPiRpcSessionOption, readPiRpcCommands, readPiRpcSessionOptions } from './rpc-options' -import { - unpickedSessionConfiguredChoice, - withLiveCatalogListing, - type AgentModelCatalogLiveListing -} from '../native-chat/agent-model-catalog/agent-model-catalog-entry' +import { readPiRpcSessionOptions } from './rpc-options' +import { withLiveCatalogListing } from '../native-chat/agent-model-catalog/agent-model-catalog-entry' +import type { PiRpcResolvedLaunch } from './rpc-launch-resolution' +import type { StructuredAgentSessionProviderOpenResult } from '../native-chat/agent-session-wire/structured-agent-session-provider-open-contract' +import type { AgentSessionProviderHandleLink } from '../../shared/agent-session-provider-handle' +import { PiRpcSessionStartup } from './rpc-session-startup' export type PiRpcConnection = Pick< JsonlRpcAgentConnection, @@ -62,10 +63,8 @@ export class PiRpcSession { readonly turns: PiRpcTurns readonly dialogs: PiRpcDialogCallbacks readonly selected = new Map() - private startCatalogListing?: AgentModelCatalogLiveListing - readonly skipped: string[] = [] - commands?: AgentSessionSlashCommand[] - options?: AgentSessionOptionsResult + readonly startup: PiRpcSessionStartup + openedLink: AgentSessionProviderHandleLink | null = null requestedClose = false private publishedExit = false private failedCause?: Error @@ -77,6 +76,7 @@ export class PiRpcSession { readonly generation: string, launch: ProviderProcessLaunch, private readonly deps: PiRpcSessionDeps, + private readonly resolvedLaunch: PiRpcResolvedLaunch, /** A new Pi session starts on its config's model; a resumed or forked one keeps its own. */ readonly resolvesConfig = false ) { @@ -145,55 +145,23 @@ export class PiRpcSession { .finally(() => this.releaseAfterExit?.()) } }) + this.startup = new PiRpcSessionStartup( + input, + this.connection, + this.selected, + deps, + resolvesConfig, + (state) => { + this.lane.apply(this.turns.context.setModel(state.model ?? undefined, Date.now())) + }, + () => this.failedCause + ) } - async start(): Promise { - const state = piRpcStateSchema.parse(await this.connection.request('get_state')) - this.lane.apply(this.turns.context.setModel(state.model ?? undefined, Date.now())) - // Keys a restore sent, whether or not Pi took them. - const picked = new Set() - for (const [key, value] of Object.entries(this.input.options ?? {})) { - if (!['model', 'effort'].includes(key)) { - this.skipped.push(key) - continue - } - picked.add(key) - try { - await applyPiRpcSessionOption(this.connection, this.selected, key, value) - } catch (error) { - this.skipped.push(key) - this.deps.logger.warn('Pi rejected a saved option', { - scope: 'pi-option-restore', - sessionId: this.input.identity.sessionId, - key, - error - }) - } - } - this.options = await readPiRpcSessionOptions(this.connection) - this.startCatalogListing = withLiveCatalogListing( - this.options, - unpickedSessionConfiguredChoice({ - resolvesConfig: this.resolvesConfig, - picked, - ...this.options - }) - ).catalogListing - try { - this.commands = (await readPiRpcCommands(this.connection)).commands - } catch (error) { - this.deps.logger.warn('Pi commands could not be read', { - scope: 'pi-commands', - sessionId: this.input.identity.sessionId, - error - }) - } - return state.sessionFile - } - - /** What this child listed at its start, with what its config resolved; the host saves it once. */ - get startListing(): AgentModelCatalogLiveListing | undefined { - return this.startCatalogListing + async open(signal: AbortSignal): Promise { + const result = await this.startup.run(this.resolvedLaunch, signal) + this.openedLink = result.link ?? null + return result } /** What the child reports now; only the start says what its config resolved. */ @@ -214,6 +182,9 @@ export class PiRpcSession { sessionId: this.input.identity.sessionId, error }) + if (!this.openedLink) { + return + } void this.close(false) .then((result) => { if (result.root !== 'exited') { @@ -256,6 +227,7 @@ export class PiRpcSession { return } this.publishedExit = true + this.failedCause ??= error this.dialogs.cancelAll() this.turns.end() this.lane.finalize() @@ -281,6 +253,7 @@ export class PiRpcSession { } finally { clearTimeout(timer) } + const cause = this.failedCause ?? error this.deps.onLifecycle({ type: 'ended', sessionId: this.input.identity.sessionId, @@ -288,7 +261,15 @@ export class PiRpcSession { acquisitionGeneration: this.generation, observedAt: Date.now(), cause: this.requestedClose ? 'requested-close' : 'unexpected-exit', - reason: (this.failedCause ?? error).message + reason: cause.message, + failure: agentSessionFailureFact('providerExited', { + detail: + providerDiagnosticOf(cause) ?? + providerDiagnosticOf(error) ?? + (cause instanceof JsonlRpcResponseError + ? providerDiagnostic(cause.message, 'person') + : undefined) + }) }) } diff --git a/src/main/runtime/rpc/methods/structured-agent-session-options-read.test.ts b/src/main/runtime/rpc/methods/structured-agent-session-options-read.test.ts index 56db442ce2c2..2efcd0e0d000 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session-options-read.test.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session-options-read.test.ts @@ -2,15 +2,9 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { setStructuredAgentSessionHost } from '../../../native-chat/agent-session-wire/structured-agent-session-registry' -import type { AgentSessionUnavailable } from '../../../../shared/agent-session-availability' import type { AgentSessionRecord } from '../../../../shared/agent-session-record' -import type { AgentSessionModelCatalogResult } from '../../../../shared/agent-session-wire' import { agentSessionRecordFixture } from '../../../../shared/agent-session-record.test-fixture' import { FLOATING_TERMINAL_WORKTREE_ID } from '../../../../shared/constants' -import { - PI_STRUCTURED_DIALOGS_RUNTIME_CAPABILITY, - STRUCTURED_AGENT_SESSION_REGISTERED_AGENTS_RUNTIME_CAPABILITY -} from '../../../../shared/protocol-version' import { call, clearStructuredHostStub, @@ -166,68 +160,3 @@ describe('agentSession.modelCatalog before anything built the host', () => { expect(reply).toMatchObject({ ok: false }) }) }) - -const PI_CLIENT = { - ...STRUCTURED_CLIENT, - clientCapabilities: [ - ...STRUCTURED_CLIENT.clientCapabilities, - STRUCTURED_AGENT_SESSION_REGISTERED_AGENTS_RUNTIME_CAPABILITY, - PI_STRUCTURED_DIALOGS_RUNTIME_CAPABILITY - ] -} - -describe("agentSession.modelCatalog with what the running agent's start said", () => { - const read = vi.fn(async (): Promise => ({ origin: 'unknown' })) - const startUnavailable = vi.fn((_sessionId: string): AgentSessionUnavailable | undefined => ({ - reason: 'notSignedIn' - })) - const record: AgentSessionRecord = { ...agentSessionRecordFixture(), provider: 'pi' } - const readFor = (agent: string) => - call('agentSession.modelCatalog', { agent, sessionId: record.sessionId }, PI_CLIENT) - - beforeEach(() => { - read.mockReset() - read.mockResolvedValue({ origin: 'unknown' }) - startUnavailable.mockClear() - setStructuredAgentSessionHost( - Object.assign(hostStub(), { - deps: { - modelCatalog: { read }, - store: { getRecord: () => record }, - adapter: { startUnavailable } - } - }) - ) - }) - - it("answers a Pi chat's catalog with its signed-out start", async () => { - const reply = await readFor('pi') - expect(startUnavailable).toHaveBeenCalledWith(record.sessionId) - expect(reply).toMatchObject({ - ok: true, - result: { origin: 'unknown', unavailable: { reason: 'notSignedIn' } } - }) - }) - - it('keeps the reason the catalog itself found', async () => { - read.mockResolvedValue({ origin: 'unknown', unavailable: { reason: 'cliMissing' } }) - expect(await readFor('pi')).toMatchObject({ - ok: true, - result: { unavailable: { reason: 'cliMissing' } } - }) - }) - - it("never lends one agent's start to another agent's read", async () => { - const reply = await readFor('codex') - expect(startUnavailable).not.toHaveBeenCalled() - expect(reply).toMatchObject({ ok: true, result: { origin: 'unknown' } }) - expect(reply).not.toMatchObject({ result: { unavailable: expect.anything() } }) - }) - - it('stays quiet while the start said nothing', async () => { - startUnavailable.mockReturnValueOnce(undefined) - const reply = await readFor('pi') - expect(reply).toMatchObject({ ok: true }) - expect(reply).not.toMatchObject({ result: { unavailable: expect.anything() } }) - }) -}) diff --git a/src/main/runtime/rpc/methods/structured-agent-session-options-read.ts b/src/main/runtime/rpc/methods/structured-agent-session-options-read.ts index 346b17ee8f0f..3d18c9b9dc31 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session-options-read.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session-options-read.ts @@ -88,13 +88,7 @@ export const STRUCTURED_AGENT_SESSION_OPTIONS_READ_METHODS = [ .catch(() => null) return catalog.read({ ...params, workspacePath }) } - const answer = await read() - // The chat's running agent can say what no listing does: a Pi that started with no model. - const started = - record?.provider === params.agent - ? host.deps.adapter.startUnavailable?.(record.sessionId) - : undefined - return started && !answer.unavailable ? { ...answer, unavailable: started } : answer + return read() } }) ] 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 new file mode 100644 index 000000000000..fd2a38dab5da --- /dev/null +++ b/src/main/runtime/structured-agent-session-adapter-start-failure.test.ts @@ -0,0 +1,128 @@ +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, expect, it, vi } from 'vitest' +import { readWholeAgentSessionFailureFact } from '../../shared/agent-session-failure' +import { computeAgentSessionPayloadFingerprint } from '../../shared/agent-session-mutation-envelope' +import { + hostTestAttachParams, + hostTestMessage, + HOST_TEST_SESSION as SESSION +} from '../native-chat/agent-session-wire/structured-agent-session-host-test-data' +import { createStructuredAgentSessionLogger } from '../native-chat/agent-session-wire/structured-agent-session-logger' +import { piScriptedChild } from '../pi/pi-scripted-child.test-fixture' +import { codexScriptedChild } from '../codex/codex-scripted-child.test-fixture' +import { acpScriptedChild } from '../acp/acp-scripted-child.test-fixture' +import type { ScriptedAgentChild } from './structured-agent-scripted-child.test-fixture' +import { + ensureStructuredAgentSessionHost, + stopStructuredAgentSessionRuntime +} from './structured-agent-session-runtime' + +let directory: string | undefined +let operations = 0 +const operationId = () => `${Date.now()}-${(++operations).toString(16).padStart(32, '0')}` +const caller = { callerKey: 'adapter-start-failure-test' } + +afterEach(async () => { + await stopStructuredAgentSessionRuntime() + if (directory) { + await rm(directory, { recursive: true, force: true }) + directory = undefined + } +}) + +async function failedStart(agent: 'pi' | 'codex' | 'grok', script: ScriptedAgentChild) { + directory = await mkdtemp(join(tmpdir(), 'orca-adapter-start-failure-')) + const root = directory + const host = await ensureStructuredAgentSessionHost({ + logger: createStructuredAgentSessionLogger(), + stateDirectory: root, + hostId: 'local', + claimKeyId: 'key-1', + resolveWorkspacePath: async () => root, + resolveClaudeAuthPolicy: () => ({ account: 'managed' }), + resolveLaunchArgs: () => [], + resolveEnvironment: async () => ({ PATH: '' }), + ...script.deps + }) + const params = hostTestAttachParams( + null, + agent === 'codex' + ? { providerHandle: undefined } + : { + provider: agent, + agent, + providerHandle: undefined, + accountHome: + agent === 'pi' + ? { variable: 'PI_CODING_AGENT_DIR', path: root } + : { variable: 'GROK_HOME', path: root }, + location: { + executionHostId: 'local', + wslDistro: null, + workspaceId: 'folder-1', + workspaceKind: 'folder' + } + } + ) + params.envelope.clientOperationId = operationId() + const attached = await host.attach(caller, params) + if (!attached.ok) { + throw new Error(JSON.stringify(attached.refusal)) + } + const body = hostTestMessage('held during start') + const sent = await host.send(caller, { + envelope: { + sessionId: SESSION, + clientOperationId: operationId(), + expectedRuntimeFence: attached.value.fence, + payloadFingerprint: computeAgentSessionPayloadFingerprint({ + method: 'agentSession.send', + sessionId: SESSION, + fields: { body } + }) + }, + body + }) + if (!sent.ok) { + throw new Error(JSON.stringify(sent.refusal)) + } + await vi.waitFor(() => expect(script.spawns()).toBe(1)) + script.failHandshake?.('provider config is invalid: bad key') + const submission = async () => { + await host.flushStreamedEvents(SESSION) + return (await host.journalSnapshot(SESSION)).submissions.find( + (entry) => entry.clientMessageId === sent.value.clientMessageId + ) + } + await vi.waitFor(async () => expect((await submission())?.dispatchState).toBe('rejected'), { + timeout: 5_000 + }) + const settled = await submission() + const snapshot = await host.journalSnapshot(SESSION) + const failureKind = agent === 'codex' ? 'startFailed' : 'providerStartFailed' + expect(readWholeAgentSessionFailureFact(settled?.rejection)).toMatchObject({ + kind: failureKind, + detail: { text: 'provider config is invalid: bad key' } + }) + expect( + snapshot.items.filter( + (item) => + item.body.kind === 'status' && + readWholeAgentSessionFailureFact(item.body.failure)?.kind === failureKind + ) + ).toHaveLength(1) + expect(script.prompts()).toEqual([]) + expect(script.spawns()).toBe(1) +} + +it('Pi records one failure without dispatch after a failed start', async () => { + await failedStart('pi', piScriptedChild()) +}) +it('Codex records one failure without dispatch after a failed start', async () => { + await failedStart('codex', codexScriptedChild()) +}) +it('ACP records one failure without dispatch after a failed start', async () => { + await failedStart('grok', acpScriptedChild('grok')()) +}) diff --git a/src/main/runtime/structured-agent-session-pi-signed-out-start.test.ts b/src/main/runtime/structured-agent-session-pi-signed-out-start.test.ts new file mode 100644 index 000000000000..0d5faf66d5f6 --- /dev/null +++ b/src/main/runtime/structured-agent-session-pi-signed-out-start.test.ts @@ -0,0 +1,113 @@ +// A signed-out Pi fails construction once; a later send after sign-in opens a new child. + +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, expect, it, vi } from 'vitest' +import { readWholeAgentSessionFailureFact } from '../../shared/agent-session-failure' +import { computeAgentSessionPayloadFingerprint } from '../../shared/agent-session-mutation-envelope' +import { + hostTestMessage, + HOST_TEST_SESSION as SESSION +} from '../native-chat/agent-session-wire/structured-agent-session-host-test-data' +import { createTestParams } from '../native-chat/agent-session-wire/structured-agent-session-create-test-fixture' +import { createStructuredAgentSessionLogger } from '../native-chat/agent-session-wire/structured-agent-session-logger' +import { piScriptedChild } from '../pi/pi-scripted-child.test-fixture' +import { peekOpenedAgentSessionRecordStore } from './agent-session-record-store-slot' +import { + ensureStructuredAgentSessionHost, + stopStructuredAgentSessionRuntime +} from './structured-agent-session-runtime' + +let directory: string | undefined +let operations = 0 +const operationId = () => `${Date.now()}-${(++operations).toString(16).padStart(32, '0')}` +const caller = { callerKey: 'pi-signed-out-start-test' } + +afterEach(async () => { + await stopStructuredAgentSessionRuntime() + if (directory) { + await rm(directory, { recursive: true, force: true }) + directory = undefined + } +}) + +it('settles the held message once, closes Pi, and starts afresh for a later send after sign-in', async () => { + directory = await mkdtemp(join(tmpdir(), 'orca-pi-signed-out-')) + const root = directory + const script = piScriptedChild({ signedOut: true }) + const host = await ensureStructuredAgentSessionHost({ + logger: createStructuredAgentSessionLogger(), + stateDirectory: root, + hostId: 'local', + claimKeyId: 'key-1', + resolveWorkspacePath: async () => root, + resolveClaudeAuthPolicy: () => ({ account: 'managed' }), + resolveLaunchArgs: () => [], + resolveEnvironment: async () => ({}), + ...script.deps + }) + const firstMessage = { + clientMessageId: operationId(), + body: hostTestMessage('hello, signed-out Pi') + } + // As a new chat sends its first message: the start is made for it. + const params = createTestParams(firstMessage, { + provider: 'pi', + agent: 'pi', + accountHome: { variable: 'PI_CODING_AGENT_DIR', path: root }, + location: { + executionHostId: 'local', + wslDistro: null, + workspaceId: 'folder-1', + workspaceKind: 'folder' + } + }) + params.envelope.clientOperationId = operationId() + const created = await host.create(caller, params, { firstMessage }) + if (!created.ok) { + throw new Error(JSON.stringify(created.refusal)) + } + // Pi answers its start after the host has published it. + await vi.waitFor(() => + expect(host.collaboratorsForTests().sessions.get(SESSION)?.child?.phase).toBe('starting') + ) + script.releaseHandshake() + const submission = async () => { + await host.flushStreamedEvents(SESSION) + return (await host.journalSnapshot(SESSION)).submissions.find( + (entry) => entry.clientMessageId === firstMessage.clientMessageId + ) + } + + await vi.waitFor(async () => expect((await submission())?.dispatchState).toBe('rejected'), { + timeout: 5_000 + }) + expect(readWholeAgentSessionFailureFact((await submission())?.rejection)?.kind).toBe( + 'notSignedIn' + ) + await vi.waitFor(() => expect(script.closes()).toBe(1)) + expect(script.spawns()).toBe(1) + expect(script.prompts()).toEqual([]) + script.signIn() + const body = hostTestMessage('hello after sign-in') + const fields = { body } + const sent = await host.send(caller, { + envelope: { + sessionId: SESSION, + clientOperationId: operationId(), + expectedRuntimeFence: + peekOpenedAgentSessionRecordStore()?.getRecord(SESSION)?.lease.runtimeFence ?? null, + payloadFingerprint: computeAgentSessionPayloadFingerprint({ + method: 'agentSession.send', + sessionId: SESSION, + fields + }) + }, + body + }) + expect(sent.ok).toBe(true) + await vi.waitFor(() => expect(script.spawns()).toBe(2)) + script.releaseHandshake() + await vi.waitFor(() => expect(script.prompts()).toEqual(['hello after sign-in'])) +}) diff --git a/src/main/runtime/structured-agent-session-pi-startup.test.ts b/src/main/runtime/structured-agent-session-pi-startup.test.ts new file mode 100644 index 000000000000..ce452f7367a0 --- /dev/null +++ b/src/main/runtime/structured-agent-session-pi-startup.test.ts @@ -0,0 +1,178 @@ +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, expect, it, vi } from 'vitest' +import { computeAgentSessionPayloadFingerprint } from '../../shared/agent-session-mutation-envelope' +import { + hostTestAttachParams, + hostTestMessage, + HOST_TEST_SESSION +} from '../native-chat/agent-session-wire/structured-agent-session-host-test-data' +import { createStructuredAgentSessionLogger } from '../native-chat/agent-session-wire/structured-agent-session-logger' +import { piScriptedChild } from '../pi/pi-scripted-child.test-fixture' +import { peekOpenedAgentSessionRecordStore } from './agent-session-record-store-slot' +import { + ensureStructuredAgentSessionHost, + stopStructuredAgentSessionRuntime +} from './structured-agent-session-runtime' + +let directory: string | undefined +let operations = 0 +const operationId = () => `${Date.now()}-${(++operations).toString(16).padStart(32, '0')}` + +afterEach(async () => { + await stopStructuredAgentSessionRuntime() + if (directory) { + await rm(directory, { recursive: true, force: true }) + directory = undefined + } +}) + +async function setup() { + directory = await mkdtemp(join(tmpdir(), 'orca-pi-startup-')) + const root = directory + const script = piScriptedChild() + const host = await ensureStructuredAgentSessionHost({ + logger: createStructuredAgentSessionLogger(), + stateDirectory: root, + hostId: 'local', + claimKeyId: 'key-1', + resolveWorkspacePath: async () => root, + resolveClaudeAuthPolicy: () => ({ account: 'managed' }), + resolveLaunchArgs: () => [], + resolveEnvironment: async () => ({}), + ...script.deps + }) + const caller = { callerKey: 'pi-startup-test' } + const params = hostTestAttachParams(null, { + provider: 'pi', + agent: 'pi', + providerHandle: undefined, + options: { unknown: 'unavailable-saved-option' }, + accountHome: { variable: 'PI_CODING_AGENT_DIR', path: root }, + location: { + executionHostId: 'local', + wslDistro: null, + workspaceId: 'folder-1', + workspaceKind: 'folder' + } + }) + params.envelope.clientOperationId = operationId() + const attached = await host.attach(caller, params) + expect(attached.ok).toBe(true) + if (!attached.ok) { + throw new Error(JSON.stringify(attached.refusal)) + } + return { script, host, caller, fence: attached.value.fence } +} + +function envelope(method: string, fields: Record) { + return { + sessionId: HOST_TEST_SESSION, + clientOperationId: operationId(), + expectedRuntimeFence: + peekOpenedAgentSessionRecordStore()?.getRecord(HOST_TEST_SESSION)?.lease.runtimeFence ?? null, + payloadFingerprint: computeAgentSessionPayloadFingerprint({ + method, + sessionId: HOST_TEST_SESSION, + fields + }) + } +} + +it('attaches Pi at spawn and holds the opening send until the provider answers its handshake', async () => { + const { script, host, caller, fence } = await setup() + expect(script.spawns()).toBe(1) + const body = hostTestMessage('hello while Pi is starting') + const sent = await host.send(caller, { + envelope: { + sessionId: HOST_TEST_SESSION, + clientOperationId: operationId(), + expectedRuntimeFence: fence, + payloadFingerprint: computeAgentSessionPayloadFingerprint({ + method: 'agentSession.send', + sessionId: HOST_TEST_SESSION, + fields: { body } + }) + }, + body + }) + expect(sent.ok).toBe(true) + expect(script.prompts()).toEqual([]) + script.releaseHandshake() + await vi.waitFor(() => expect(script.prompts()).toEqual(['hello while Pi is starting'])) + await vi.waitFor(() => + expect(peekOpenedAgentSessionRecordStore()?.getRecord(HOST_TEST_SESSION)?.options).toEqual({ + unknown: 'unavailable-saved-option' + }) + ) + await stopStructuredAgentSessionRuntime() + expect(script.closes()).toBe(1) +}) + +it('writes startup model and effort picks before a message accepted after them', async () => { + const { script, host, caller } = await setup() + const picked = ( + [ + ['model', 'scripted/model-2'], + ['effort', 'low'] + ] as const + ).map(([key, value]) => + host.setOption(caller, { + envelope: envelope('agentSession.setOption', { key, value }), + key, + value + }) + ) + const body = hostTestMessage('first message') + const sent = host.send(caller, { envelope: envelope('agentSession.send', { body }), body }) + expect(script.wire()).not.toContain('set_model') + script.releaseHandshake() + expect((await Promise.all(picked)).every((result) => result.ok)).toBe(true) + expect((await sent).ok).toBe(true) + await vi.waitFor(() => expect(script.prompts()).toEqual(['first message'])) + const wire = script.wire() + expect(wire.indexOf('set_model')).toBeGreaterThanOrEqual(0) + expect(wire.indexOf('set_thinking_level')).toBeGreaterThan(wire.indexOf('set_model')) + expect(wire.indexOf('prompt:first message')).toBeGreaterThan(wire.indexOf('set_thinking_level')) + expect(peekOpenedAgentSessionRecordStore()?.getRecord(HOST_TEST_SESSION)?.options).toMatchObject({ + model: 'scripted/model-2', + effort: 'low' + }) +}) + +it('ends Pi on Stop and resumes its real session file when the next send starts a child', async () => { + const { script, host, caller } = await setup() + script.completeTurns = false + script.releaseHandshake() + const first = hostTestMessage('start a long turn') + expect( + ( + await host.send(caller, { + envelope: envelope('agentSession.send', { body: first }), + body: first + }) + ).ok + ).toBe(true) + await vi.waitFor(() => expect(script.prompts()).toEqual(['start a long turn'])) + expect((await host.cancel(caller, { envelope: envelope('agentSession.cancel', {}) })).ok).toBe( + true + ) + await vi.waitFor(() => expect(script.closes()).toBe(1)) + const second = hostTestMessage('resume after Stop') + expect( + ( + await host.send(caller, { + envelope: envelope('agentSession.send', { body: second }), + body: second + }) + ).ok + ).toBe(true) + await vi.waitFor(() => expect(script.spawns()).toBe(2)) + expect(script.resumes()).toBe(1) + expect(script.prompts()).toEqual(['start a long turn']) + script.releaseHandshake() + await vi.waitFor(() => + expect(script.prompts()).toEqual(['start a long turn', 'resume after Stop']) + ) +}) diff --git a/src/main/runtime/structured-agent-session-provider-operations-while-starting.test.ts b/src/main/runtime/structured-agent-session-provider-operations-while-starting.test.ts new file mode 100644 index 000000000000..e6cd3762a512 --- /dev/null +++ b/src/main/runtime/structured-agent-session-provider-operations-while-starting.test.ts @@ -0,0 +1,236 @@ +// Protocol operations wait for the real Codex adapter to open; only its child is scripted. + +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, expect, it, vi } from 'vitest' +import { agentJournalSubmissionKey } from '../../shared/agent-session-journal-item-key' +import { computeAgentSessionPayloadFingerprint } from '../../shared/agent-session-mutation-envelope' +import type { AgentSessionMutationEnvelope } from '../../shared/agent-session-wire' +import { CODEX_STRUCTURED_AGENT } from '../codex/codex-structured-agent-definition' +import type { StructuredAgentDefinition } from '../native-chat/agent-session-wire/structured-agent-definition' +import { + CODEX_SCRIPTED_THREAD, + codexScriptedChild +} from '../codex/codex-scripted-child.test-fixture' +import type { StructuredAgentSessionHost } from '../native-chat/agent-session-wire/structured-agent-session-host' +import { + HOST_TEST_SESSION as SESSION, + hostTestAttachParams, + hostTestMessage +} from '../native-chat/agent-session-wire/structured-agent-session-host-test-data' +import { createStructuredAgentSessionLogger } from '../native-chat/agent-session-wire/structured-agent-session-logger' +import type { ScriptedAgentChild } from './structured-agent-scripted-child.test-fixture' +import { + ensureStructuredAgentSessionHost, + stopStructuredAgentSessionRuntime +} from './structured-agent-session-runtime' + +const CALLER = { callerKey: 'provider-operations-test' } +let root: string | undefined +let operations = 0 +/** The attach that opened the chat; a reconnecting view replays it. */ +let openedWith: Parameters[1] | undefined + +afterEach(async () => { + await stopStructuredAgentSessionRuntime() + if (root) { + await rm(root, { recursive: true, force: true }) + root = undefined + } +}) + +const operationId = (): string => `${Date.now()}-${(++operations).toString(16).padStart(32, '0')}` + +const eventually = (assertion: () => void | Promise) => + vi.waitFor(assertion, { timeout: 10_000 }) + +function envelope( + host: StructuredAgentSessionHost, + method: string, + fields: Record +): AgentSessionMutationEnvelope { + return { + sessionId: SESSION, + clientOperationId: operationId(), + expectedRuntimeFence: host.deps.store.getRecord(SESSION)?.lease.runtimeFence ?? 0, + payloadFingerprint: computeAgentSessionPayloadFingerprint({ + method, + sessionId: SESSION, + fields + }) + } +} + +async function openChat( + child: ScriptedAgentChild, + definition: StructuredAgentDefinition = CODEX_STRUCTURED_AGENT +): Promise { + const { agent } = definition + root = await mkdtemp(join(tmpdir(), `orca-provider-operations-${agent}-`)) + const directory = root + const host = await ensureStructuredAgentSessionHost({ + logger: createStructuredAgentSessionLogger(), + stateDirectory: directory, + hostId: 'local', + claimKeyId: 'key-1', + resolveWorkspacePath: async () => directory, + resolveLaunchArgs: () => [], + resolveClaudeAuthPolicy: () => ({ account: 'managed' }), + ...child.deps + }) + const params = hostTestAttachParams(null, { + provider: agent, + agent, + accountHome: + definition.accountLocatorKind === 'opencode' + ? { kind: 'opencode', locator: { kind: 'unmanaged' } } + : { variable: definition.accountHomeVariable, path: join(directory, agent) }, + providerHandle: undefined + }) + // The runtime admits operation ids against its own clock. + params.envelope.clientOperationId = operationId() + const attached = await host.attach(CALLER, params) + expect(attached, JSON.stringify(attached)).toMatchObject({ ok: true }) + openedWith = params + return host +} + +async function send(host: StructuredAgentSessionHost, text: string) { + const body = hostTestMessage(text) + return host.send(CALLER, { envelope: envelope(host, 'agentSession.send', { body }), body }) +} + +/** A first turn the scripted Codex recorded, then the chat put to rest with a turn to target. */ +async function codexChatAtRestWithATurn(child: ReturnType) { + child.holdHandshakes = false + const host = await openChat(child) + child.releaseHandshake() + const first = await send(host, 'first turn') + if (!first.ok) { + throw new Error(JSON.stringify(first.refusal)) + } + const { clientMessageId } = first.value + let providerItemId: string | null = null + await eventually(async () => { + const submission = (await host.journalSnapshot(SESSION)).submissions.find( + (entry) => entry.clientMessageId === clientMessageId + ) + expect(submission?.providerItemId).toEqual(expect.any(String)) + providerItemId = submission?.providerItemId ?? null + }) + await host.close(SESSION, 'evict') + expect(child.closes()).toBe(1) + child.holdHandshakes = true + return { + host, + itemId: agentJournalSubmissionKey(clientMessageId), + providerItemId: providerItemId! + } +} + +/** Resolves to whether `pending` settled within a short wait. */ +async function settlesSoon(pending: Promise): Promise { + let settled = false + void pending.finally(() => (settled = true)) + await new Promise((resolve) => setTimeout(resolve, 50)) + return settled +} + +it('settles a rewind that goes in doubt while a send waits in the queue, by waiting out the start the send needs', async () => { + const child = codexScriptedChild({ + 'thread/read': () => ({ thread: { id: CODEX_SCRIPTED_THREAD, status: { type: 'idle' } } }), + 'thread/turns/list': () => ({ data: [], nextCursor: null }), + 'thread/items/list': () => ({ data: [], nextCursor: null }) + }) + const { host, itemId, providerItemId } = await codexChatAtRestWithATurn(child) + const { store } = host.deps + const expectedEpoch = (await host.journalSnapshot(SESSION)).cursor.epoch + const queueHeld = Promise.withResolvers() + void host.collaboratorsForTests().serialize(SESSION, () => queueHeld.promise) + + // Nothing is in doubt when the send arrives; it waits in the queue. + const sent = send(host, 'after the rewind') + expect(await settlesSoon(sent)).toBe(false) + const request = envelope(host, 'agentSession.rewind', { itemId, expectedEpoch }) + await store.admitMutationOperation({ + callerKey: CALLER.callerKey, + envelope: request, + hostFingerprint: request.payloadFingerprint, + now: Date.now() + }) + await store.transitionHandoff(SESSION, (record) => ({ + ...record, + rewind: { + operationId: request.clientOperationId, + callerKey: CALLER.callerKey, + itemId, + providerItemId, + expectedEpoch, + phase: 'prepared', + retained: [] + } + })) + queueHeld.resolve() + + // The send starts the agent to settle the rewind, and waits that start out. + await eventually(() => expect(child.spawns()).toBe(2)) + expect(await settlesSoon(sent)).toBe(false) + child.releaseHandshake() + + expect(await sent).toMatchObject({ ok: true }) + expect(store.getRecord(SESSION)?.rewind?.phase).toBe('completed') + await eventually(() => expect(child.prompts()).toEqual(['first turn', 'after the rewind'])) +}) + +it('answers a rewind asked of a resting Codex chat from the thread it opens, leaving nothing in doubt', async () => { + const child = codexScriptedChild({ + 'thread/resume': (params) => ({ + thread: { id: params?.threadId, historyMode: 'legacy' }, + model: 'scripted-model' + }) + }) + const { host, itemId } = await codexChatAtRestWithATurn(child) + const expectedEpoch = (await host.journalSnapshot(SESSION)).cursor.epoch + + const rewound = host.rewind(CALLER, { + itemId, + expectedEpoch, + envelope: envelope(host, 'agentSession.rewind', { itemId, expectedEpoch }) + }) + await eventually(() => expect(child.spawns()).toBe(2)) + expect(await settlesSoon(rewound)).toBe(false) + child.releaseHandshake() + + // Only the opened thread says its history cannot be rewound. + expect(await rewound).toMatchObject({ + ok: false, + refusal: { rewindReason: 'history-not-paginated' } + }) + expect(host.deps.store.getRecord(SESSION)?.rewind).toBeUndefined() + expect(await send(host, 'still sends')).toMatchObject({ ok: true }) + await eventually(() => expect(child.prompts()).toEqual(['first turn', 'still sends'])) +}) + +it('keeps a goal waiting on a starting Codex chat through a re-attach, then sets it once the thread is open', async () => { + const child = codexScriptedChild() + const host = await openChat(child) + const change = { kind: 'set' as const, objective: 'Ship the release' } + + const goal = host.changeThreadGoal(CALLER, { + envelope: envelope(host, 'agentSession.threadGoal', { change }), + change + }) + expect(await settlesSoon(goal)).toBe(false) + // A view reconnects while the chat is still starting: its replayed attach keeps the live child. + if (!openedWith) { + throw new Error('the chat was not opened') + } + expect(await host.attach(CALLER, openedWith)).toMatchObject({ ok: true, replayed: true }) + expect(child.spawns()).toBe(1) + expect(await settlesSoon(goal)).toBe(false) + child.releaseHandshake() + + expect(await goal).toMatchObject({ ok: true }) + expect(child.requests().some(({ method }) => method === 'thread/goal/set')).toBe(true) +}) diff --git a/src/main/runtime/structured-agent-session-transport-overrides.ts b/src/main/runtime/structured-agent-session-transport-overrides.ts index e1cbf8e26bbe..6bf042ca9607 100644 --- a/src/main/runtime/structured-agent-session-transport-overrides.ts +++ b/src/main/runtime/structured-agent-session-transport-overrides.ts @@ -5,10 +5,13 @@ import type { CodexStructuredSessionAdapterDeps } from '../codex/codex-structure import type { readCodexCliInstallation } from '../preflight/codex-cli-installation' import type { StructuredAgentSessionStartupLimits } from '../native-chat/agent-session-wire/structured-agent-session-startup-attempt-contract' import type { PiRpcSessionDeps } from '../pi/rpc-session' +import type { PiRpcSessionAdapterDeps } from '../pi/rpc-session-adapter' /** Provider transports are overridden only to drive the runtime against scripted children. */ export type StructuredAgentSessionTransportOverrides = { openPiConnection?: PiRpcSessionDeps['openConnection'] + /** Scripted Pi children resolve their launch here, so no test reaches a real Pi binary. */ + resolvePiLaunch?: PiRpcSessionAdapterDeps['resolveLaunch'] openCodexConnection?: CodexStructuredSessionAdapterDeps['openConnection'] /** Scripted transports must also replace the real host's version probe. */ readCodexInstallation?: typeof readCodexCliInstallation