mirror of
https://github.com/stablyai/orca.git
synced 2026-11-03 18:47:40 +00:00
fix(native-chat): keep the model an agent actually runs, not its startup report (#32006)
* fix(native-chat): save only the model a chat's turn ran, never a startup report OpenCode's session/new names a model before its catalog loads; that report became the chat's saved model and later sends switched to it. * refactor(acp): move option proof with the agent's own late report * test(pi): expect ranOptions on model-only sends * fix(acp): follow the agent's model after an effort-only pick An effort pick stopped the native model from tracking the agent's later report, so a send with no model wrote the stale startup model back. * test(acp): assert the agent's late model report arrives * fix(pi): read run evidence at the prompt write * refactor(acp): move the option write lifecycle into its own module Keeps acp-structured-options.ts under the 300-line cap after the agent's late report is adopted.
This commit is contained in:
@@ -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<AcpStructuredConnection, 'setConfigOption' | 'setModel'>,
|
||||
options: AcpStructuredOptions,
|
||||
write: AcpOptionWrite,
|
||||
bound: { agent: string; timeoutMs: number; signal?: AbortSignal; key: string; value: string }
|
||||
): Promise<void> {
|
||||
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<void>((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)
|
||||
})
|
||||
}
|
||||
@@ -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<string, string> = { ...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<AcpStructuredConnection, 'setConfigOption' | 'setModel'>,
|
||||
options: AcpStructuredOptions,
|
||||
write: AcpOptionWrite,
|
||||
bound: { agent: string; timeoutMs: number; signal?: AbortSignal; key: string; value: string }
|
||||
): Promise<void> {
|
||||
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<void>((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)
|
||||
})
|
||||
}
|
||||
|
||||
@@ -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 } }
|
||||
|
||||
@@ -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'
|
||||
|
||||
+28
-1
@@ -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(
|
||||
|
||||
@@ -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<Record<'model' | 'effort' | 'serviceTier' | 'fastMode', string>>
|
||||
type RanOptions = Partial<Record<'model' | 'effort', string>>
|
||||
|
||||
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<Record<string, string | undefined>> = executed ?? {
|
||||
effort: message?.effort ?? report.effort,
|
||||
serviceTier: report.serviceTier,
|
||||
fastMode: report.fastMode
|
||||
effort: message?.effort
|
||||
}
|
||||
const ran: Record<string, string> = {}
|
||||
for (const [key, value] of Object.entries({ ...proposed, model })) {
|
||||
|
||||
@@ -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<StructuredAgentSessionAdapter['dispatch']>[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<AgentSessionDispatchOutcome> {
|
||||
let ranOptions: Record<string, string> = {}
|
||||
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
|
||||
}
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user