diff --git a/src/main/acp/acp-session-option-write.ts b/src/main/acp/acp-session-option-write.ts new file mode 100644 index 00000000000..0498210ba70 --- /dev/null +++ b/src/main/acp/acp-session-option-write.ts @@ -0,0 +1,64 @@ +import { abortSignalReason } from '../../shared/abort-signal-reason' +import type { AcpStructuredConnection } from './acp-structured-connection' +import type { AcpOptionWrite, AcpStructuredOptions } from './acp-structured-options' + +/** Only the current, still-awaited write may establish reusable option proof. */ +export function writeAcpSessionOption( + connection: Pick, + options: AcpStructuredOptions, + write: AcpOptionWrite, + bound: { agent: string; timeoutMs: number; signal?: AbortSignal; key: string; value: string } +): Promise { + const { agent, timeoutMs, signal } = bound + if (signal?.aborted) { + return Promise.reject(abortSignalReason(signal)) + } + const revision = ++options.writeRevision + options.currentProof.delete(bound.key) + if (bound.key === 'model') { + options.currentProof.delete('effort') + } + const adopt = (apply: () => void): void => { + if (options.writeRevision !== revision) { + throw new Error(`${agent} option write was superseded`) + } + apply() + options.currentProof.add(bound.key) + options.applied.set(bound.key, bound.value) + } + const applied = + write.method === 'config' + ? connection + .setConfigOption(write.configId, write.value) + .then((result) => adopt(() => options.adoptConfigOptions(result.configOptions))) + : connection + .setModel(write.modelId) + .then(() => adopt(() => options.adoptModel(write.modelId))) + return new Promise((resolve, reject) => { + const fail = (error: Error): void => { + if (options.writeRevision === revision) { + options.writeRevision += 1 + } + settle() + reject(error) + } + const onAbort = (): void => { + if (signal) { + fail(abortSignalReason(signal)) + } + } + const timer = setTimeout( + () => fail(new Error(`${agent} did not answer an option write within ${timeoutMs}ms`)), + timeoutMs + ) + const settle = (): void => { + clearTimeout(timer) + signal?.removeEventListener('abort', onAbort) + } + signal?.addEventListener('abort', onAbort, { once: true }) + applied.then(() => { + settle() + resolve() + }, fail) + }) +} diff --git a/src/main/acp/acp-structured-options.ts b/src/main/acp/acp-structured-options.ts index 1338487eb2d..15bb91d572f 100644 --- a/src/main/acp/acp-structured-options.ts +++ b/src/main/acp/acp-structured-options.ts @@ -1,5 +1,3 @@ -import { abortSignalReason } from '../../shared/abort-signal-reason' - import type { AgentSessionModelOption, AgentSessionOptionChoice, @@ -95,9 +93,32 @@ export class AcpStructuredOptions { adoptConfigOptions(configOptions: SessionConfigOption[] | null | undefined): void { if (configOptions) { this.configOptions = configOptions + this.adoptOwnChoice() } } + /** Until Orca writes a key, a later report is still the agent's own (OpenCode's start answers + * before its catalog loads); an effort belongs to its model, so it stops with either write. */ + private adoptOwnChoice(): void { + if (this.picked.has('model')) { + return + } + const reported = this.reported() + const own = this.picked.has('effort') ? ['model'] : ['model', 'effort'] + const choice: Record = { ...this.initialChoice } + for (const key of own) { + const value = reported[key] + if (value === undefined) { + delete choice[key] + this.currentProof.delete(key) + } else { + choice[key] = value + this.currentProof.add(key) + } + } + this.initialChoice = choice + } + adoptModel(modelId: string): void { if (this.models) { this.models = { ...this.models, currentModelId: modelId } @@ -262,64 +283,3 @@ export async function readAcpLiveOptions( acceptsImages } } - -/** Only the current, still-awaited write may establish reusable option proof. */ -export function writeAcpSessionOption( - connection: Pick, - options: AcpStructuredOptions, - write: AcpOptionWrite, - bound: { agent: string; timeoutMs: number; signal?: AbortSignal; key: string; value: string } -): Promise { - const { agent, timeoutMs, signal } = bound - if (signal?.aborted) { - return Promise.reject(abortSignalReason(signal)) - } - const revision = ++options.writeRevision - options.currentProof.delete(bound.key) - if (bound.key === 'model') { - options.currentProof.delete('effort') - } - const adopt = (apply: () => void): void => { - if (options.writeRevision !== revision) { - throw new Error(`${agent} option write was superseded`) - } - apply() - options.currentProof.add(bound.key) - options.applied.set(bound.key, bound.value) - } - const applied = - write.method === 'config' - ? connection - .setConfigOption(write.configId, write.value) - .then((result) => adopt(() => options.adoptConfigOptions(result.configOptions))) - : connection - .setModel(write.modelId) - .then(() => adopt(() => options.adoptModel(write.modelId))) - return new Promise((resolve, reject) => { - const fail = (error: Error): void => { - if (options.writeRevision === revision) { - options.writeRevision += 1 - } - settle() - reject(error) - } - const onAbort = (): void => { - if (signal) { - fail(abortSignalReason(signal)) - } - } - const timer = setTimeout( - () => fail(new Error(`${agent} did not answer an option write within ${timeoutMs}ms`)), - timeoutMs - ) - const settle = (): void => { - clearTimeout(timer) - signal?.removeEventListener('abort', onAbort) - } - signal?.addEventListener('abort', onAbort, { once: true }) - applied.then(() => { - settle() - resolve() - }, fail) - }) -} diff --git a/src/main/acp/acp-structured-session-adapter.test.ts b/src/main/acp/acp-structured-session-adapter.test.ts index 4ea1e92dcea..422e1e2fa36 100644 --- a/src/main/acp/acp-structured-session-adapter.test.ts +++ b/src/main/acp/acp-structured-session-adapter.test.ts @@ -8,6 +8,7 @@ import { AgentSessionAcquisitionRootExitObservedError } from '../native-chat/age import { GENERIC_ACP_DIALECT } from './acp-dialects/acp-dialect' import { GROK, + GROK_CONFIG_OPTIONS, openAcpAdapterRig, PID, PROVIDER_SESSION, @@ -236,6 +237,58 @@ describe('ACP structured session adapter: turns', () => { ]) }) + it.each([ + ['no choice', {}], + ['an effort choice', { effort: 'low' }] + ])( + 'runs a message with %s on the model the agent moved to before Orca wrote one', + async (_name, executionOptions: { effort?: string }) => { + // As OpenCode 2: session/new answers before its catalog loads, then reports the configured model. + const rig = await openAcpAdapterRig() + await rig.acquire() + if (executionOptions.effort) { + await rig.adapter.setOption({ + sessionId: SESSION, + key: 'effort', + value: executionOptions.effort, + fence: 1 + }) + } + rig.child().agent.notify('session/update', { + sessionId: PROVIDER_SESSION, + update: { + sessionUpdate: 'config_option_update', + configOptions: GROK_CONFIG_OPTIONS.map((option) => + option.id === 'model' + ? { ...option, currentValue: 'grok-4.6' } + : { ...option, currentValue: executionOptions.effort ?? option.currentValue } + ) + } + }) + await waitFor(() => + expect(rig.lifecycle).toContainEqual( + expect.objectContaining({ + type: 'options-reported', + reportedOptions: expect.objectContaining({ model: 'grok-4.6' }) + }) + ) + ) + await rig.adapter.dispatch({ + sessionId: SESSION, + fence: 1, + clientMessageId: 'unchosen-send', + body: hello, + executionOptions + }) + await rig.frame('session/prompt') + expect(rig.sent('session/set_config_option').map((frame) => frame.params)).toEqual( + executionOptions.effort + ? [{ sessionId: PROVIDER_SESSION, configId: 'reasoning_effort', value: 'low' }] + : [] + ) + } + ) + it("sends Grok's prompt identity only to an agent whose dialect echoes it", async () => { const rig = await openAcpAdapterRig({ deps: { spec: { ...GROK, dialect: GENERIC_ACP_DIALECT } } diff --git a/src/main/acp/acp-structured-session-adapter.ts b/src/main/acp/acp-structured-session-adapter.ts index 779ac73b57d..76c37392c99 100644 --- a/src/main/acp/acp-structured-session-adapter.ts +++ b/src/main/acp/acp-structured-session-adapter.ts @@ -29,7 +29,8 @@ import { ACP_STOP_GRACE_MS, type AcpStructuredSessionAdapterDeps } from './acp-structured-session-adapter-deps' -import { readAcpLiveOptions, writeAcpSessionOption } from './acp-structured-options' +import { readAcpLiveOptions } from './acp-structured-options' +import { writeAcpSessionOption } from './acp-session-option-write' import { stopAcpChildren, acpChildStopCapabilities } from './acp-structured-child-stop' import { acpStartupFailure, isAcpReadySession } from './acp-structured-child' import { AcpStructuredChildren } from './acp-structured-child-lifecycle' diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-message-choices.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-message-choices.test.ts index 9db575ef542..7e3f94ba472 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-message-choices.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-message-choices.test.ts @@ -53,6 +53,10 @@ it('keeps the model and effort a chat with none saved first ran, after its confi await rig.dispose() const agent = { restartable: true as const, reportsModel: 'gpt-ran', reportsEffort: 'xhigh' } rig = await createQueuedMessageTestRig(agent) + rig.dispatch.mockResolvedValueOnce({ + state: 'admitted', + ranOptions: { model: 'gpt-ran', effort: 'xhigh' } + }) const first = rig.send('first') expect(await first.result).toMatchObject({ ok: true }) await eventually(() => @@ -72,6 +76,7 @@ it('keeps the model and effort a chat with none saved first ran, after its confi it('records what a later readback reports when the start named only a filled-in default', async () => { // Model Claude’s delayed settings readback after startup without a model. + rig.dispatch.mockResolvedValueOnce({ state: 'admitted', ranOptions: {} }) const first = rig.send('first') expect(await first.result).toMatchObject({ ok: true }) await eventually(() => expect(rig.dispatch).toHaveBeenCalledTimes(1)) @@ -90,10 +95,32 @@ it('records what a later readback reports when the start named only a filled-in }) preparation.reject(new Error('content preparation failed')) await eventually(() => - expect(rig.store.getRecord(SESSION)?.options).toEqual({ model: 'opus[1m]', effort: 'high' }) + expect(rig.store.getRecord(SESSION)?.options).toEqual({ model: 'opus[1m]' }) ) }) +it('saves no model for an agent that reports options but gives no evidence of what ran', async () => { + // As OpenCode: its session/new names one model and a config update moments later the one it runs. + await rig.dispose() + rig = await createQueuedMessageTestRig({ restartable: true, reportsModel: 'startup-model' }) + const first = rig.send('first') + expect(await first.result).toMatchObject({ ok: true }) + await eventually(() => expect(rig.dispatch).toHaveBeenCalledTimes(1)) + await rig.host.handleAdapterEvent({ + type: 'options-reported', + sessionId: SESSION, + fence: rig.store.getRecord(SESSION)!.lease.runtimeFence, + acquisitionGeneration: 'generation-1', + reportedOptions: { model: 'live-model' } + }) + await rig.settleAccepted(first.id, 'a') + const second = rig.send('second') + expect(await second.result).toMatchObject({ ok: true }) + await eventually(() => expect(rig.dispatch).toHaveBeenCalledTimes(2)) + expect(rig.dispatch.mock.calls[1]?.[0].executionOptions).toEqual({}) + expect(rig.store.getRecord(SESSION)?.options).toBeUndefined() +}) + it('rejects a refused choice before prompt input and leaves the next pick saved', async () => { await pickModel('model-b') rig.dispatch.mockRejectedValueOnce( diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-ran-options.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-ran-options.ts index 8323dfad49f..f075186b3e3 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-ran-options.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-ran-options.ts @@ -1,4 +1,4 @@ -// Fill unsaved choices from execution evidence, never host-filled settings/defaults. +// Fill unsaved choices from execution evidence, never startup/config reports or host defaults. import type { AgentSessionOptionsResult } from '../../../shared/agent-session-wire' import type { AgentSessionExecutionOptions } from '../../../shared/agent-session-execution-options' @@ -17,7 +17,7 @@ type RanOptionsContext = { now: () => number } -type RanOptions = Partial> +type RanOptions = Partial> export type StructuredAgentSessionHandedOptions = { message?: AgentSessionExecutionOptions @@ -64,18 +64,12 @@ export function noteStructuredAgentSessionRuntimePermission( } function reportedRan(reported: AgentSessionOptionsResult['current']): RanOptions | null { - const { model, effort, serviceTier, fastMode, fastModeState } = reported + const { model, effort } = reported const filled = reported.defaulted ?? [] if (!model || model === 'default' || filled.includes('model')) { return null } - return { - model, - ...(effort && !filled.includes('effort') ? { effort } : {}), - ...(serviceTier ? { serviceTier } : {}), - // Persist only confirmed Fast-on; off may mean this account cannot use Fast. - ...(fastMode && fastModeState === 'on' ? { fastMode: 'true' } : {}) - } + return { model, ...(effort && !filled.includes('effort') ? { effort } : {}) } } /** Resolve delayed model evidence for an already handed-over turn. */ @@ -96,7 +90,7 @@ export async function noteStructuredAgentSessionReportedOptions( const owed = runs.owed await recordStructuredAgentSessionRanOptions(context, sessionId, fence, { ...owed, - ...(owed.executed ? { executed: { ...owed.executed, model: ran.model } } : {}) + executed: { ...owed.executed, model: ran.model } }) } } @@ -150,20 +144,16 @@ export async function recordStructuredAgentSessionRanOptions( } const { message, executed } = handed const reported = childRuns.get(child)?.reported + // A report only completes an agent's own evidence of the dispatch, never stands in for it. const newerModel = checkpoint && reported !== checkpoint.reported ? reported?.model : undefined - const model = executed - ? (executed.model ?? message?.model ?? newerModel) - : (message?.model ?? reported?.model) - childRuns.set(child, { ...childRuns.get(child), owed: model ? undefined : handed }) + const model = executed ? (executed.model ?? message?.model ?? newerModel) : message?.model + childRuns.set(child, { ...childRuns.get(child), owed: executed && !model ? handed : undefined }) if (!model) { return } - // Prefer dispatch proof; fallback tier/Fast must match the reported model because steers inherit turn choices. - const report: RanOptions = reported?.model === model ? reported : {} + // Prefer dispatch proof, else the message's effort; its tier or Fast may not have run. const proposed: Readonly> = executed ?? { - effort: message?.effort ?? report.effort, - serviceTier: report.serviceTier, - fastMode: report.fastMode + effort: message?.effort } const ran: Record = {} for (const [key, value] of Object.entries({ ...proposed, model })) { diff --git a/src/main/pi/rpc-message-dispatch.ts b/src/main/pi/rpc-message-dispatch.ts index d301516ffcd..a09da025f9e 100644 --- a/src/main/pi/rpc-message-dispatch.ts +++ b/src/main/pi/rpc-message-dispatch.ts @@ -12,7 +12,7 @@ import type { } from '../native-chat/agent-session-wire/structured-agent-session-adapter' import { AgentSessionOptionRejectedError } from '../native-chat/agent-session-wire/structured-agent-session-option-error' import { applyPiRpcSessionOption } from './rpc-options' -import { PiRpcPromptError, preparePiRpcPrompt } from './rpc-prompt' +import { PiRpcPromptError, preparePiRpcPrompt, type PiRpcPrompt } from './rpc-prompt' import type { PiRpcSession, PiRpcSessionDeps } from './rpc-session' type DispatchInput = Parameters[0] @@ -60,12 +60,7 @@ export async function dispatchPiRpcMessage( const model = input.executionOptions?.model ?? session.initialOptions.model if (!input.executionOptions) { input.signal?.throwIfAborted() - return session.submit( - input.clientMessageId, - input.requestedAt ?? Date.now(), - { ...prompt }, - input.beforeDispatch - ) + return submitPiRpcPrompt(session, input, prompt) } for (const key of ['model', 'effort'] as const) { const value = @@ -109,10 +104,27 @@ export async function dispatchPiRpcMessage( if (!isCurrent() || session.connection.closed) { return exited() } - return session.submit( + return submitPiRpcPrompt(session, input, prompt) +} + +/** Pi runs the prompt on the model and effort it proved: its own state, never a stale report. */ +async function submitPiRpcPrompt( + session: PiRpcSession, + input: DispatchInput, + prompt: PiRpcPrompt +): Promise { + let ranOptions: Record = {} + const outcome = await session.submit( input.clientMessageId, input.requestedAt ?? Date.now(), { ...prompt }, - input.beforeDispatch + async () => { + await input.beforeDispatch?.() + input.onExecutionSnapshot?.() + ranOptions = Object.fromEntries(session.optionProof.current) + } ) + return outcome.state === 'accepted' || outcome.state === 'admitted' + ? { ...outcome, ranOptions } + : outcome } diff --git a/src/main/pi/rpc-session-adapter.test.ts b/src/main/pi/rpc-session-adapter.test.ts index 52dccb27692..addc5a0c191 100644 --- a/src/main/pi/rpc-session-adapter.test.ts +++ b/src/main/pi/rpc-session-adapter.test.ts @@ -154,7 +154,7 @@ describe('Pi RPC session ownership and delivery', () => { clientMessageId: 'send-1', body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'hello' }] } }) - expect(result).toEqual({ state: 'admitted' }) + expect(result).toEqual({ state: 'admitted', ranOptions: { model: 'anthropic/model-1' } }) expect(h.settled).not.toHaveBeenCalled() h.connection.receive({ type: 'agent_start' }) expect(h.settled).toHaveBeenCalledWith({ @@ -200,7 +200,10 @@ describe('Pi RPC session ownership and delivery', () => { body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'chosen' }] }, executionOptions: { model: 'anthropic/model-2', effort: 'low' } }) - expect(result).toEqual({ state: 'admitted' }) + expect(result).toEqual({ + state: 'admitted', + ranOptions: { model: 'anthropic/model-2', effort: 'low' } + }) expect(h.connection.requests.slice(-2)).toEqual(['set_model', 'set_thinking_level']) expect(h.connection.sent).toMatchObject([{ type: 'prompt', message: 'chosen' }]) }) @@ -226,7 +229,7 @@ describe('Pi RPC session ownership and delivery', () => { }, executionOptions: { model } }) - ).resolves.toEqual({ state: 'admitted' }) + ).resolves.toMatchObject({ state: 'admitted', ranOptions: { model } }) } // Back on the opening model, its opening effort is restored; elsewhere Pi keeps its own. expect(h.connection.requests.filter((command) => command.startsWith('set_'))).toEqual([ @@ -293,7 +296,10 @@ describe('Pi RPC session ownership and delivery', () => { executionOptions: { model: 'anthropic/model-1', effort: 'low' }, body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'A again' }] } }) - ).resolves.toEqual({ state: 'admitted' }) + ).resolves.toEqual({ + state: 'admitted', + ranOptions: { model: 'anthropic/model-1', effort: 'low' } + }) expect(h.connection.requests.filter((command) => command === 'set_model')).toHaveLength(2) unanswered.resolve({}) await Promise.resolve()