From 14ee5ed1fa1609e5152de692f5111e28474f4873 Mon Sep 17 00:00:00 2001 From: Ewen Date: Wed, 5 Aug 2026 19:05:43 +0800 Subject: [PATCH] refactor(acp): extract provider prompt executor --- src/main/acp/provider-prompt-executor.test.ts | 273 +++++++++++++++++ src/main/acp/provider-prompt-executor.ts | 188 ++++++++++++ src/main/acp/runtime.test.ts | 43 +++ src/main/acp/runtime.ts | 288 +++++++----------- 4 files changed, 620 insertions(+), 172 deletions(-) create mode 100644 src/main/acp/provider-prompt-executor.test.ts create mode 100644 src/main/acp/provider-prompt-executor.ts diff --git a/src/main/acp/provider-prompt-executor.test.ts b/src/main/acp/provider-prompt-executor.test.ts new file mode 100644 index 00000000..e6ef56b2 --- /dev/null +++ b/src/main/acp/provider-prompt-executor.test.ts @@ -0,0 +1,273 @@ +import type { ActiveSession, PromptResponse, SessionNotification } from '@agentclientprotocol/sdk' +import { describe, expect, it, vi } from 'vitest' + +import { + AcpProviderPromptExecutor, + type ProviderPromptExecutionInput +} from './provider-prompt-executor' +import type { AcpProviderTurnAdapter, AcpProviderTurnProbe } from './provider-turn-adapter' + +type NextUpdate = Awaited> + +type Deferred = { + promise: Promise + resolve: (value: Value) => void + reject: (error: unknown) => void +} + +const deferred = (): Deferred => { + let resolve!: (value: Value) => void + let reject!: (error: unknown) => void + const promise = new Promise((promiseResolve, promiseReject) => { + resolve = promiseResolve + reject = promiseReject + }) + return { promise, resolve, reject } +} + +const notification: SessionNotification = { + sessionId: 'provider-1', + update: { + sessionUpdate: 'agent_message_chunk', + content: { type: 'text', text: 'hello' } + } +} + +const update = (): NextUpdate => ({ + kind: 'session_update', + notification, + update: notification.update +}) +const stop = (response: PromptResponse): NextUpdate => ({ kind: 'stop', response }) as NextUpdate + +const setup = ( + messages: NextUpdate[] = [] +): { + executor: AcpProviderPromptExecutor + input: ProviderPromptExecutionInput + session: ProviderPromptExecutionInput['session'] + probe: AcpProviderTurnProbe & { + observe: ReturnType + finalize: ReturnType + cancel: ReturnType + } + adapter: AcpProviderTurnAdapter & { begin: ReturnType } + accepted: ReturnType + captureStop: ReturnType + routeNotification: ReturnType + report: ReturnType +} => { + const queue = [...messages] + const session = { + sessionId: 'provider-1', + prompt: vi.fn(async () => undefined), + nextUpdate: vi.fn(async () => { + const message = queue.shift() + if (!message) return new Promise(() => undefined) + return message + }) + } as unknown as ProviderPromptExecutionInput['session'] + const probe = { + observe: vi.fn(), + finalize: vi.fn(async () => ({})), + cancel: vi.fn(async () => undefined) + } as AcpProviderTurnProbe & { + observe: ReturnType + finalize: ReturnType + cancel: ReturnType + } + const adapter = { + begin: vi.fn(async () => probe) + } as AcpProviderTurnAdapter & { begin: ReturnType } + const accepted = vi.fn() + const captureStop = vi.fn(() => true) + const routeNotification = vi.fn() + const report = vi.fn() + return { + executor: new AcpProviderPromptExecutor(), + session, + probe, + adapter, + accepted, + captureStop, + routeNotification, + report, + input: { + session, + content: 'prompt', + cwd: '/workspace', + adapter, + isCurrent: () => true, + beforeDispatch: async () => 'active', + captureStop, + onAccepted: accepted, + routeNotification, + reportBestEffortFailure: report + } + } +} + +describe('AcpProviderPromptExecutor', () => { + it('accepts once, routes updates in order, captures stop, and returns raw normalized facts', async () => { + const response: PromptResponse = { + stopReason: 'end_turn', + usage: { inputTokens: 12, outputTokens: 3, totalTokens: 15 } + } + const fixture = setup([update(), update(), stop(response)]) + fixture.probe.finalize = vi.fn(async () => ({ modelTurnCount: 2 })) + + const outcome = await fixture.executor.execute(fixture.input) + + expect(outcome).toEqual({ + kind: 'stopped', + response, + facts: { + turnUsage: { inputTokens: 12, cacheTokens: 0, outputTokens: 3 }, + modelTurnCount: 2 + } + }) + if (outcome.kind !== 'stopped') throw new Error('expected stopped outcome') + expect(outcome.response).toBe(response) + expect(fixture.adapter.begin).toHaveBeenCalledWith({ + providerSessionId: 'provider-1', + cwd: '/workspace' + }) + expect(fixture.accepted).toHaveBeenCalledOnce() + expect(fixture.routeNotification.mock.calls).toEqual([[notification], [notification]]) + expect(fixture.captureStop).toHaveBeenCalledOnce() + expect(fixture.probe.finalize).toHaveBeenCalledWith({ response }) + expect(fixture.probe.cancel).not.toHaveBeenCalled() + }) + + it('preserves prompt rejection before acceptance and cancels its probe once', async () => { + const fixture = setup() + const rejection = new Error('provider rejected') + ;(fixture.session.prompt as ReturnType).mockRejectedValueOnce(rejection) + + await expect(fixture.executor.execute(fixture.input)).rejects.toBe(rejection) + + expect(fixture.accepted).not.toHaveBeenCalled() + expect(fixture.routeNotification).not.toHaveBeenCalled() + expect(fixture.captureStop).not.toHaveBeenCalled() + expect(fixture.probe.finalize).not.toHaveBeenCalled() + expect(fixture.probe.cancel).toHaveBeenCalledOnce() + }) + + it('dispatches no prompt when superseded during asynchronous probe begin', async () => { + const fixture = setup() + const probeGate = deferred() + ;(fixture.adapter.begin as ReturnType).mockReturnValueOnce(probeGate.promise) + let current = true + const pending = fixture.executor.execute({ ...fixture.input, isCurrent: () => current }) + await vi.waitFor(() => expect(fixture.adapter.begin).toHaveBeenCalledOnce()) + + current = false + probeGate.resolve(fixture.probe) + await expect(pending).resolves.toEqual({ kind: 'not-dispatched' }) + + expect(fixture.session.prompt).not.toHaveBeenCalled() + expect(fixture.probe.cancel).toHaveBeenCalledOnce() + }) + + it('suppresses stale updates while draining the old provider through raw stop', async () => { + const response: PromptResponse = { stopReason: 'cancelled' } + const fixture = setup([update(), stop(response)]) + + const outcome = await fixture.executor.execute({ ...fixture.input, isCurrent: () => false }) + + expect(outcome).toEqual({ kind: 'not-dispatched' }) + expect(fixture.session.prompt).not.toHaveBeenCalled() + + let current = true + const acceptedThenStale = setup([update(), update(), stop(response)]) + acceptedThenStale.routeNotification.mockImplementationOnce(() => { + current = false + }) + await expect( + acceptedThenStale.executor.execute({ + ...acceptedThenStale.input, + isCurrent: () => current + }) + ).resolves.toEqual({ kind: 'superseded', response }) + expect(acceptedThenStale.accepted).toHaveBeenCalledOnce() + expect(acceptedThenStale.routeNotification).toHaveBeenCalledOnce() + expect(acceptedThenStale.captureStop).not.toHaveBeenCalled() + expect(acceptedThenStale.probe.finalize).not.toHaveBeenCalled() + expect(acceptedThenStale.probe.cancel).toHaveBeenCalledOnce() + }) + + it('treats a lost terminal capture race as superseded', async () => { + const response: PromptResponse = { stopReason: 'end_turn' } + const fixture = setup([stop(response)]) + fixture.captureStop.mockReturnValueOnce(false) + + await expect(fixture.executor.execute(fixture.input)).resolves.toEqual({ + kind: 'superseded', + response + }) + expect(fixture.accepted).toHaveBeenCalledOnce() + expect(fixture.probe.finalize).not.toHaveBeenCalled() + expect(fixture.probe.cancel).toHaveBeenCalledOnce() + }) + + it('captures stop before slow finalization and falls back when finalization fails', async () => { + const response: PromptResponse = { + stopReason: 'end_turn', + usage: { inputTokens: 7, cachedReadTokens: 2, outputTokens: 1, totalTokens: 10 } + } + const fixture = setup([stop(response)]) + const finalizeGate = deferred() + fixture.probe.finalize = vi.fn(() => finalizeGate.promise) + const pending = fixture.executor.execute(fixture.input) + await vi.waitFor(() => expect(fixture.captureStop).toHaveBeenCalledOnce()) + + finalizeGate.reject(new Error('usage unavailable')) + await expect(pending).resolves.toEqual({ + kind: 'stopped', + response, + facts: { + turnUsage: { inputTokens: 7, cacheTokens: 2, outputTokens: 1 } + } + }) + expect(fixture.report).toHaveBeenCalledWith('finalize', expect.any(Error)) + expect(fixture.probe.cancel).not.toHaveBeenCalled() + }) + + it('keeps begin and acceptance failures best-effort', async () => { + const response: PromptResponse = { stopReason: 'end_turn' } + const fixture = setup([update(), stop(response)]) + ;(fixture.adapter.begin as ReturnType).mockRejectedValueOnce( + new Error('probe unavailable') + ) + fixture.accepted.mockImplementationOnce(() => { + throw new Error('accept callback failed') + }) + + await expect(fixture.executor.execute(fixture.input)).resolves.toMatchObject({ + kind: 'stopped', + response + }) + expect(fixture.routeNotification).toHaveBeenCalledOnce() + expect(fixture.report.mock.calls.map(([stage]) => stage)).toEqual(['begin', 'accepted']) + }) + + it('routes provider observations only while the matching probe owns the attempt', async () => { + const response: PromptResponse = { stopReason: 'end_turn' } + const fixture = setup() + const nextUpdate = deferred() + ;(fixture.session.nextUpdate as ReturnType).mockReturnValue(nextUpdate.promise) + const pending = fixture.executor.execute(fixture.input) + await vi.waitFor(() => expect(fixture.session.prompt).toHaveBeenCalledOnce()) + + const providerMessage = { sessionId: 'provider-1', message: { type: 'result' } } + fixture.executor.observeProviderMessage(providerMessage) + fixture.executor.observeProviderMessage({ sessionId: 'other-provider' }) + expect(fixture.probe.observe).toHaveBeenCalledWith(providerMessage) + expect(fixture.probe.observe).toHaveBeenCalledOnce() + + nextUpdate.resolve(stop(response)) + await pending + fixture.executor.observeProviderMessage(providerMessage) + expect(fixture.probe.observe).toHaveBeenCalledOnce() + }) +}) diff --git a/src/main/acp/provider-prompt-executor.ts b/src/main/acp/provider-prompt-executor.ts new file mode 100644 index 00000000..0acb2e19 --- /dev/null +++ b/src/main/acp/provider-prompt-executor.ts @@ -0,0 +1,188 @@ +import type { + ActiveSession, + ContentBlock, + PromptResponse, + SessionNotification +} from '@agentclientprotocol/sdk' + +import { toAcpTurnTokenUsage } from '../../shared/acp' +import type { + AcpProviderTurnAdapter, + AcpProviderTurnProbe, + AcpProviderTurnResult +} from './provider-turn-adapter' + +type ProviderPromptObservationStage = 'accepted' | 'begin' | 'cancel' | 'finalize' | 'observe' + +type ProviderPromptExecutionInput = Readonly<{ + session: Pick + content: string | ContentBlock[] + cwd: string + adapter: AcpProviderTurnAdapter + isCurrent: () => boolean + beforeDispatch: () => Promise<'active' | 'cancelled'> + captureStop: () => boolean + onAccepted: () => void + routeNotification: (notification: SessionNotification) => void + reportBestEffortFailure?: (stage: ProviderPromptObservationStage, error: unknown) => void +}> + +type ProviderPromptOutcome = + | Readonly<{ kind: 'not-dispatched' }> + | Readonly<{ kind: 'superseded'; response: PromptResponse }> + | Readonly<{ kind: 'stopped'; response: PromptResponse; facts: AcpProviderTurnResult }> + +type ActiveObservation = Readonly<{ + token: symbol + probe: AcpProviderTurnProbe + report?: ProviderPromptExecutionInput['reportBestEffortFailure'] +}> + +const EMPTY_FACTS: AcpProviderTurnResult = Object.freeze({}) +const NOOP_PROBE: AcpProviderTurnProbe = Object.freeze({ + finalize: () => EMPTY_FACTS, + cancel: () => undefined +}) + +const reportBestEffort = ( + report: ProviderPromptExecutionInput['reportBestEffortFailure'], + stage: ProviderPromptObservationStage, + error: unknown +): void => { + try { + report?.(stage, error) + } catch { + // Diagnostics must not replace the provider outcome. + } +} + +const normalizeFacts = ( + response: PromptResponse, + facts: AcpProviderTurnResult +): AcpProviderTurnResult => { + const turnUsage = facts.turnUsage ?? toAcpTurnTokenUsage(response.usage) + return Object.freeze({ + ...(turnUsage ? { turnUsage: Object.freeze({ ...turnUsage }) } : {}), + ...(facts.modelTurnCount === undefined ? {} : { modelTurnCount: facts.modelTurnCount }), + ...(facts.contextUsedTokens === undefined ? {} : { contextUsedTokens: facts.contextUsedTokens }) + }) +} + +class AcpProviderPromptExecutor { + private readonly observations = new Map() + + observeProviderMessage(message: unknown): void { + if (typeof message !== 'object' || message === null || Array.isArray(message)) return + const providerSessionId = (message as { sessionId?: unknown }).sessionId + if (typeof providerSessionId !== 'string') return + const observation = this.observations.get(providerSessionId) + if (!observation?.probe.observe) return + try { + observation.probe.observe(message) + } catch (error) { + reportBestEffort(observation.report, 'observe', error) + } + } + + async execute(input: ProviderPromptExecutionInput): Promise { + const providerSessionId = input.session.sessionId + const token = Symbol(providerSessionId) + let probe = NOOP_PROBE + try { + probe = await input.adapter.begin({ providerSessionId, cwd: input.cwd }) + } catch (error) { + reportBestEffort(input.reportBestEffortFailure, 'begin', error) + } + + this.observations.set(providerSessionId, { + token, + probe, + report: input.reportBestEffortFailure + }) + let probeState: 'open' | 'cancelled' | 'finalized' = 'open' + const releaseObservation = (): void => { + if (this.observations.get(providerSessionId)?.token === token) { + this.observations.delete(providerSessionId) + } + } + const cancelProbe = async (): Promise => { + if (probeState !== 'open') return + probeState = 'cancelled' + releaseObservation() + try { + await probe.cancel() + } catch (error) { + reportBestEffort(input.reportBestEffortFailure, 'cancel', error) + } + } + + try { + const dispatch = await input.beforeDispatch() + if (dispatch === 'cancelled' || !input.isCurrent()) { + await cancelProbe() + return Object.freeze({ kind: 'not-dispatched' }) + } + + let promptRequest: Promise + try { + promptRequest = input.session.prompt(input.content) + } catch (error) { + await cancelProbe() + throw error + } + const promptFailure = promptRequest.then( + () => new Promise(() => undefined), + (error) => Object.freeze({ kind: 'provider-rejection' as const, error }) + ) + let accepted = false + + for (;;) { + const message = await Promise.race([input.session.nextUpdate(), promptFailure]) + if (message.kind === 'provider-rejection') throw message.error + + if (!input.isCurrent()) { + if (message.kind !== 'stop') continue + await cancelProbe() + return Object.freeze({ kind: 'superseded', response: message.response }) + } + + if (!accepted) { + accepted = true + try { + input.onAccepted() + } catch (error) { + reportBestEffort(input.reportBestEffortFailure, 'accepted', error) + } + } + if (message.kind !== 'stop') { + input.routeNotification(message.notification) + continue + } + if (!input.captureStop()) { + await cancelProbe() + return Object.freeze({ kind: 'superseded', response: message.response }) + } + + probeState = 'finalized' + releaseObservation() + let facts = EMPTY_FACTS + try { + facts = await probe.finalize({ response: message.response }) + } catch (error) { + reportBestEffort(input.reportBestEffortFailure, 'finalize', error) + } + return Object.freeze({ + kind: 'stopped', + response: message.response, + facts: normalizeFacts(message.response, facts) + }) + } + } finally { + await cancelProbe() + releaseObservation() + } + } +} + +export { AcpProviderPromptExecutor } +export type { ProviderPromptExecutionInput, ProviderPromptObservationStage, ProviderPromptOutcome } diff --git a/src/main/acp/runtime.test.ts b/src/main/acp/runtime.test.ts index bb89b7d0..0684c7cc 100644 --- a/src/main/acp/runtime.test.ts +++ b/src/main/acp/runtime.test.ts @@ -3799,6 +3799,49 @@ describe('ACP runtime session management', () => { expect(agent.prompts[0]?.text).toContain('replacement turn') }) + it('does not dispatch after reset wins during an asynchronous provider probe', async () => { + const process = new FakeAgentProcess() + const agent = startFakeAgent(process, ['remote-session-1', 'remote-session-2']) + const probeStarted = createDeferred() + const releaseProbe = createDeferred() + let usageFetchCount = 0 + const opencodeUsageFetch = vi.fn(async () => { + usageFetchCount += 1 + if (usageFetchCount === 1) { + probeStarted.resolve() + await releaseProbe.promise + } + return new Response(JSON.stringify([]), { + headers: { 'content-type': 'application/json' } + }) + }) + const framework = { ...opencodeFramework, spawn: () => asAgentProcess(process) } + const runtime = new AcpRuntime({ + appVersion: '0.1.0', + defaultCwd: '/workspace', + resolveBackend: () => ({ + framework, + executablePath: '/bin/opencode', + env: {}, + opencodeUsageApi: { + baseUrl: 'http://127.0.0.1:4242', + authorization: 'Basic test' + } + }), + framework, + opencodeUsageFetch + }) + + const session = await runtime.createSession({ cwd: '/workspace' }) + const stalePrompt = runtime.sendPrompt({ sessionId: session.sessionId, text: 'stale turn' }) + await probeStarted.promise + + await runtime.resetSessionContext({ sessionId: session.sessionId, cwd: '/workspace' }) + releaseProbe.resolve() + await expect(stalePrompt).resolves.toMatchObject({ stopReason: 'cancelled' }) + expect(agent.prompts).toHaveLength(0) + }) + it('does not let a superseded turn finally clear the replay turn in-flight lock', async () => { const root = await createTemporaryRoot() const artifactRepository = new ArtifactRepository(root) diff --git a/src/main/acp/runtime.ts b/src/main/acp/runtime.ts index 2551394f..99e9e2e7 100644 --- a/src/main/acp/runtime.ts +++ b/src/main/acp/runtime.ts @@ -29,11 +29,6 @@ import type { AcpSetPermissionProfileRequest, AcpStateSnapshot } from '../../shared/acp' -import { - ACP_MODEL_TURN_COUNT_META_KEY, - ACP_TURN_TOKEN_USAGE_META_KEY, - toAcpTurnTokenUsage -} from '../../shared/acp' import { ACP_PROMPT_FAILED_EVENT_TITLE } from '../../shared/acp' import { DEFAULT_PERMISSION_PROFILE, @@ -54,8 +49,7 @@ import { import { resolveCanonicalMcpToolIdentity } from '../agent-framework/app-mcp-names' import { createLogger, diagnosticErrorFields, errorLogFields } from '../logger' import { extractProviderToolName } from './runtime-events' -import { toCodexTurnTokenUsage } from './codex-turn-usage' -import { fetchOpenCodeUsageSnapshot, sumOpenCodeTurnUsage } from './opencode-turn-usage' +import { fetchOpenCodeUsageSnapshot } from './opencode-turn-usage' import { matchSessionModelOption, resolveSessionEffortOption } from './session-config' import { describePromptError, isProviderPromptError } from './prompt-error' import { AcpRuntimeSnapshotOwner } from './runtime-snapshot-owner' @@ -139,6 +133,11 @@ import { AcpSessionReplacementWorkflow } from './session-replacement-workflow' import { AcpSessionDeletionWorkflow } from './session-deletion-workflow' import { AcpPromptPreparationOwner, type PreparedPromptHandle } from './prompt-preparation-owner' import { AcpSessionPresentationPolicy } from './session-presentation-policy' +import { AcpProviderPromptExecutor } from './provider-prompt-executor' +import type { AcpProviderTurnAdapter } from './provider-turn-adapter' +import { claudeCodeTurnAdapter } from './claude-turn-adapter' +import { createCodexTurnAdapter } from './codex-turn-adapter' +import { AcpOpenCodeTurnAdapter } from './opencode-turn-adapter' import { AcpTurnSkillOwner, type AcpTurnSkillHooks, @@ -284,15 +283,6 @@ type AcpRuntimeSkillImportOptions = { ) => Promise<() => void> } -// Mirror claude-agent-acp's autonomous result lanes. Unknown future origins stay eligible so a -// newly introduced user lane does not silently lose the terminal SDK `num_turns` value. -const CLAUDE_AUTONOMOUS_RESULT_ORIGINS = new Set([ - 'task-notification', - 'peer', - 'coordinator', - 'observer', - 'observer-activity' -]) // An end_turn is final from the runtime's perspective, so promised work must be a tool call in the // current turn or an explicit request for user input rather than text that implies later execution. const TURN_CONTINUITY_SYSTEM_PROMPT_APPEND = [ @@ -395,6 +385,7 @@ class AcpRuntime { private readonly backendGeneration: AcpBackendGenerationOwner private readonly sessionConfigurator: AcpSessionConfigurator private readonly sessionUpdateProjector = new AcpSessionUpdateProjector() + private readonly providerPromptExecutor = new AcpProviderPromptExecutor() // Injectable lifecycle timers (defaults to real setTimeout/clearTimeout). private readonly setTimer: (fn: () => void, ms: number) => ReturnType private readonly clearTimer: (handle: ReturnType) => void @@ -1897,145 +1888,103 @@ class AcpRuntime { .lookup(request.sessionId) ?.aggregate.snapshot() const promptFramework = promptSessionSnapshot?.frameworkId ?? this.framework.id - const openCodeUsageApi = this.backendGeneration.openCodeUsageApi() - const opencodeUsageBefore = - promptFramework === 'opencode' && openCodeUsageApi - ? await fetchOpenCodeUsageSnapshot( - openCodeUsageApi, - activeSession.sessionId, - promptSessionSnapshot?.cwd ?? this.snapshotOwner.cwd, - this.options.opencodeUsageFetch - ) - : undefined - if ( - (await this.sessionInteractions.cancellationCheckpoint(promptInteraction)) === 'cancelled' - ) { - return finishCancelledBeforePrompt() - } - - // Start the prompt and race it against routed updates from the active session queue. - if (request.historyPreamble) { - log.info('session transcript replay dispatched', { - sessionId: request.sessionId, - historyTextLength: request.historyPreamble.length, - historyAttachmentCount: request.historyAttachments?.length ?? 0, - historyImageCount: request.historyImages?.length ?? 0, - ...this.diagnosticContext() - }) - } - const promptFailure = new Promise((_, reject) => { - activeSession.prompt(promptContent).catch(reject) - }) - let providerPromptAccepted = false - - for (;;) { - const message = await Promise.race([activeSession.nextUpdate(), promptFailure]) - - // A reset/replacement may supersede this provider turn while a queued update or terminal - // response is still draining. Settle the abandoned promise, but never project its state or - // terminal facts into the replacement interaction. - if (this.sessionInteractions.current(request.sessionId) !== promptInteraction) { - if (message.kind === 'stop') return message.response - continue - } - - if (!providerPromptAccepted) { - providerPromptAccepted = true + const providerOutcome = await this.providerPromptExecutor.execute({ + session: activeSession, + content: promptContent, + cwd: promptSessionSnapshot?.cwd ?? this.snapshotOwner.cwd, + adapter: this.providerTurnAdapter(promptFramework), + isCurrent: () => + this.sessionInteractions.current(request.sessionId) === promptInteraction && + this.activeSessionFor(request.sessionId) === activeSession, + beforeDispatch: async () => { + if ( + (await this.sessionInteractions.cancellationCheckpoint(promptInteraction)) === + 'cancelled' + ) { + return 'cancelled' + } + if (request.historyPreamble) { + log.info('session transcript replay dispatched', { + sessionId: request.sessionId, + historyTextLength: request.historyPreamble.length, + historyAttachmentCount: request.historyAttachments?.length ?? 0, + historyImageCount: request.historyImages?.length ?? 0, + ...this.diagnosticContext() + }) + } + return 'active' + }, + captureStop: () => this.sessionInteractions.captureTerminal(promptInteraction, 'stop'), + onAccepted: () => { try { this.callbacks.onProviderPromptAccepted?.(request.sessionId, promptAttemptId) } catch (error) { safeLogError('provider-prompt-accepted callback failed', errorLogFields(error)) } - } - - if (skillActivitiesStarted && !skillActivitiesFinalized) { - this.emitCodexSkillInputActivities( - request.sessionId, - promptTurn, - skillActivityInputs, - 'completed' - ) - skillActivitiesFinalized = true - } - - if (message.kind === 'stop') { - turnSkillOutcome = message.response.stopReason === 'cancelled' ? 'cancelled' : 'completed' - const codexTurnCount = message.response._meta?.[ACP_MODEL_TURN_COUNT_META_KEY] - const reportedTurnCount = - promptFramework === 'codex' && - Number.isSafeInteger(codexTurnCount) && - (codexTurnCount as number) > 0 - ? (codexTurnCount as number) - : undefined - observedPromptStop = { - response: message.response, - turnUsage: - promptFramework === 'codex' - ? (toCodexTurnTokenUsage(message.response._meta?.[ACP_TURN_TOKEN_USAGE_META_KEY]) ?? - toCodexTurnTokenUsage(message.response.usage)) - : toAcpTurnTokenUsage(message.response.usage), - ...(reportedTurnCount === undefined ? {} : { modelTurnCount: reportedTurnCount }) - } - // Freeze provider-terminal time before artifact work and provider-specific usage fetching. - // The outcome remains authoritative through close/reset while usage facts are finalized. - if (!this.sessionInteractions.captureTerminal(promptInteraction, 'stop')) { - return message.response + if (skillActivitiesStarted && !skillActivitiesFinalized) { + this.emitCodexSkillInputActivities( + request.sessionId, + promptTurn, + skillActivityInputs, + 'completed' + ) + skillActivitiesFinalized = true } - this.recordCodexPromptResponseContextUsage( - request.sessionId, - message.response, - promptTurn - ) - if (contextUsageTurn.complete()) this.emitState() - // Emit artifact metadata before stop so the renderer can attach files to the finished message. - await this.emitArtifactRunEvent(request.sessionId, artifactRun) - artifactEmitted = true - log.info('prompt stopped', { + }, + routeNotification: (notification) => + this.handleSessionUpdate(notification, request.sessionId), + reportBestEffortFailure: (stage, error) => + log.warn('provider prompt observation failed', { sessionId: request.sessionId, - stopReason: message.stopReason + stage, + ...errorLogFields(error) }) - const opencodeTurnUsage = - promptFramework === 'opencode' && openCodeUsageApi - ? sumOpenCodeTurnUsage( - opencodeUsageBefore, - await fetchOpenCodeUsageSnapshot( - openCodeUsageApi, - activeSession.sessionId, - this.sessionRegistry.lookup(request.sessionId)?.aggregate.snapshot().cwd ?? - this.snapshotOwner.cwd, - this.options.opencodeUsageFetch - ) - ) - : undefined - if (opencodeTurnUsage) observedPromptStop.turnUsage = opencodeTurnUsage - publishObservedPromptStop() - if ( - this.sessionInteractions.current(request.sessionId) === promptInteraction && - this.activeSessionFor(request.sessionId) === activeSession && - this.shouldAutoCompactContext(request.sessionId) - ) { - try { - await this.performNativeContextCompaction( - activeSession, - request.sessionId, - 'automatic' - ) - } catch (error) { - log.warn('automatic context compaction failed', { - sessionId: request.sessionId, - ...errorLogFields(error) - }) - } - } - return message.response - } + }) - // Route the update under the app-facing id so a session adopted onto a new agent (after a - // provider switch) still streams into the same conversation the renderer is watching. (No - // per-update log line here: it fires once per streamed chunk and floods the console for no - // signal — 'prompt start'/'prompt stopped' already bracket the turn.) - this.handleSessionUpdate(message.notification, request.sessionId) + if (providerOutcome.kind === 'not-dispatched') { + return finishCancelledBeforePrompt() + } + if (providerOutcome.kind === 'superseded') { + return providerOutcome.response + } + const { response, facts } = providerOutcome + turnSkillOutcome = response.stopReason === 'cancelled' ? 'cancelled' : 'completed' + observedPromptStop = { + response, + ...(facts.turnUsage ? { turnUsage: facts.turnUsage } : {}), + ...(facts.modelTurnCount === undefined ? {} : { modelTurnCount: facts.modelTurnCount }) + } + if (facts.contextUsedTokens !== undefined) { + this.recordProviderPromptContextUsage( + request.sessionId, + facts.contextUsedTokens, + promptTurn + ) + } + if (contextUsageTurn.complete()) this.emitState() + // Emit artifact metadata before stop so the renderer can attach files to the finished message. + await this.emitArtifactRunEvent(request.sessionId, artifactRun) + artifactEmitted = true + log.info('prompt stopped', { + sessionId: request.sessionId, + stopReason: response.stopReason + }) + publishObservedPromptStop() + if ( + this.sessionInteractions.current(request.sessionId) === promptInteraction && + this.activeSessionFor(request.sessionId) === activeSession && + this.shouldAutoCompactContext(request.sessionId) + ) { + try { + await this.performNativeContextCompaction(activeSession, request.sessionId, 'automatic') + } catch (error) { + log.warn('automatic context compaction failed', { + sessionId: request.sessionId, + ...errorLogFields(error) + }) + } } + return response } catch (error) { if (observedPromptStop) { // Provider stop/cancellation already won the outcome race. App-side finalization failure, @@ -2265,22 +2214,7 @@ class AcpRuntime { } private observeClaudeSdkMessage(params: Record): void { - if (typeof params.sessionId !== 'string') return - if (typeof params.message !== 'object' || params.message === null) return - - const message = params.message as Record - if (message.type !== 'result') return - const origin = - typeof message.origin === 'object' && message.origin !== null - ? (message.origin as Record).kind - : undefined - if (typeof origin === 'string' && CLAUDE_AUTONOMOUS_RESULT_ORIGINS.has(origin)) return - if (!Number.isSafeInteger(message.num_turns) || (message.num_turns as number) <= 0) return - - const appSessionId = this.sessionRegistry.resolveAppSessionId(params.sessionId) - const promptInteraction = this.currentPromptInteraction(appSessionId) - if (!promptInteraction) return - this.sessionInteractions.observeModelTurns(promptInteraction, message.num_turns as number) + this.providerPromptExecutor.observeProviderMessage(params) } // Looks up the workspace root bound to a session for filesystem operations. @@ -2572,26 +2506,36 @@ class AcpRuntime { this.permissionContext.cancelForSession(sessionId) } - // App-managed codex-acp emits the exact per-request numerator during generation. Codex's pinned - // adapter publishes uncached input and cached input as separate PromptResponse categories, so - // recombine them for the context numerator when applying the final per-request correction. - private recordCodexPromptResponseContextUsage( + private providerTurnAdapter(frameworkId: AgentFramework['id']): AcpProviderTurnAdapter { + if (frameworkId === 'claude-code') return claudeCodeTurnAdapter + if (frameworkId === 'codex') return createCodexTurnAdapter() + + const usageApi = this.backendGeneration.openCodeUsageApi() + return new AcpOpenCodeTurnAdapter((providerSessionId, cwd) => + usageApi + ? fetchOpenCodeUsageSnapshot( + usageApi, + providerSessionId, + cwd, + this.options.opencodeUsageFetch + ) + : Promise.resolve(undefined) + ) + } + + // Provider adapters normalize an exact latest-request context numerator when one exists. Apply it + // only while the same app turn still owns the Session so a delayed old stop cannot rewrite a reset. + private recordProviderPromptContextUsage( sessionId: string, - response: PromptResponse, + used: number, promptTurn: number ): void { if ( - this.framework.id !== 'codex' || this.pendingProviderReconnect || this.currentPromptInteraction(sessionId)?.sequence !== promptTurn ) { return } - - const usage = toCodexTurnTokenUsage(response.usage) - if (!usage) return - - const used = usage.inputTokens + usage.cacheTokens if (this.contextUsageTracker.reconcileUsed(sessionId, used)) this.emitState() }