diff --git a/docs/design/2026-10-11-never-started-dispatch-reconcile.md b/docs/design/2026-10-11-never-started-dispatch-reconcile.md index e84a94615d6..f6596b3dd37 100644 --- a/docs/design/2026-10-11-never-started-dispatch-reconcile.md +++ b/docs/design/2026-10-11-never-started-dispatch-reconcile.md @@ -168,6 +168,18 @@ independent process ledger also returns its definitive 404. Only the typed Turn. A present row, unknown result, wrong identity or rejected probe leaves the record unresolved. The publication error itself is never proof. +For a failed turn with multiple background Shell bindings, each grant-owner +`NOT_STARTED` decision settles its record immediately. All remaining +grant-owner decisions finish before the permitted process-ledger probes start together. +A silent Broker therefore consumes one 30-second probe window rather than one +window per binding. The reload scanner and live-turn recovery share the same +probe decision: only an absent row proves never-started; a present row reports +`ambiguous_state ()`; and a rejected lookup reports +`probe_rejected ()`. Diagnostics name the execution id, distinguish +the two known Broker contract errors and the transport error type without +interpolating rejection messages, and record each successful process-ledger +settlement. + This fallback applies only to background Shell requests. Monitor dispatches create their own `:process` rows, but `request.background` deliberately excludes them; their settlement stays with the monitor-specific grant and remote-evidence diff --git a/docs/design/2026-10-11-never-started-dispatch-reconcile.zh-CN.md b/docs/design/2026-10-11-never-started-dispatch-reconcile.zh-CN.md index 83ced6af8d1..a2027500c23 100644 --- a/docs/design/2026-10-11-never-started-dispatch-reconcile.zh-CN.md +++ b/docs/design/2026-10-11-never-started-dispatch-reconcile.zh-CN.md @@ -149,6 +149,16 @@ Broker。任何明确状态都是反证，不得被后备查询覆盖。因此 p 行存在、unknown、身份不匹配或查询被拒绝时，记录继续未结算；publication 报错本身永远不是未启动证明。 +当一个失败回合包含多个后台 Shell binding 时，每个 grant owner 返回的 +`NOT_STARTED` 都会立即结算对应记录；剩余 grant owner 判定全部完成后，再并发 +启动所有允许执行的进程账本探测。因此 Broker 静默无响应只消耗 +一个 30 秒探测窗口，而不是每个 binding 各消耗一个窗口。reload 扫描器和 +回合内恢复分支共用同一探测判定：只有账本行缺失才能证明 never-started； +账本行存在时报告 `ambiguous_state ()`；查询被拒绝时报告 +`probe_rejected ()`。诊断包含 execution id，可在不插入 +rejection message 的前提下区分两个已知 Broker 契约错误与传输错误类型， +并记录每个由进程账本证明后成功结算的记录。 + 此后备取证仅适用于后台 Shell 请求。Monitor 派发会创建自己的 `:process` 行，但 `request.background` 会有意排除它们；其结算继续由 Monitor 专属的 grant 与远程证据路径负责。不增加 acquire、release、派发或协议字段。记录 diff --git a/packages/cli/src/serve/hosted-never-started-reconcile.test.ts b/packages/cli/src/serve/hosted-never-started-reconcile.test.ts index c4ded61cca7..0288c426028 100644 --- a/packages/cli/src/serve/hosted-never-started-reconcile.test.ts +++ b/packages/cli/src/serve/hosted-never-started-reconcile.test.ts @@ -20,7 +20,10 @@ import { managedExtensionRecordKey } from '@qwen-code/qwen-code-core/managed-run import * as stdio from '../utils/stdioHelpers.js'; import { HostedChildAgentSession } from './hosted-child-agent-session.js'; import { HostedChildRunSession } from './hosted-child-run-session.js'; -import { reconcileNeverStartedChildRuns } from './hosted-never-started-reconcile.js'; +import { + probeNeverStarted, + reconcileNeverStartedChildRuns, +} from './hosted-never-started-reconcile.js'; import type { HostedWorkspaceBrokerOptions } from './hosted-workspace-broker.js'; const enablement = vi.hoisted(() => ({ childRun: true })); @@ -258,6 +261,69 @@ async function reopen(): Promise { await open('reopened-worker'); } +describe('probeNeverStarted', () => { + it.each([ + [undefined, { proven: true }], + [ + { state: 'executing' }, + { proven: false, cause: 'ambiguous_state (executing)' }, + ], + [ + { state: 'unknown' }, + { proven: false, cause: 'ambiguous_state (unknown)' }, + ], + ] as const)( + 'classifies a Broker answer without widening absence', + async (answer, expected) => { + const backgroundProcessStatus = vi.fn(async () => answer); + + await expect( + probeNeverStarted({ backgroundProcessStatus }, 'shell-1'), + ).resolves.toEqual(expected); + }, + ); + + it.each([ + [ + Object.assign(new Error('secret broker URL'), { + status: 401, + code: 'runtime_broker_auth_rejected', + }), + 'probe_rejected (status 401, code runtime_broker_auth_rejected)', + ], + [ + new Error('Runtime execution identity changed.'), + 'probe_rejected (identity_changed)', + ], + [ + new Error('Runtime execution outcome is unknown.'), + 'probe_rejected (outcome_unknown)', + ], + [new TypeError('fetch failed'), 'probe_rejected (error TypeError)'], + [ + new DOMException('timed out', 'TimeoutError'), + 'probe_rejected (error TimeoutError)', + ], + ['unexpected rejection', 'probe_rejected (no status)'], + ])( + 'classifies a rejected probe without logging its message', + async (rejection, expectedCause) => { + const backgroundProcessStatus = vi.fn(async () => { + throw rejection; + }); + + const result = await probeNeverStarted( + { backgroundProcessStatus }, + 'shell-1', + ); + + expect(result).toEqual({ proven: false, cause: expectedCause }); + expect(JSON.stringify(result)).not.toContain('secret broker URL'); + expect(JSON.stringify(result)).not.toContain('fetch failed'); + }, + ); +}); + describe('reconcileNeverStartedChildRuns', () => { it('settles a stranded dispatch-claimed record whose process provably does not exist', async () => { await open(); diff --git a/packages/cli/src/serve/hosted-never-started-reconcile.ts b/packages/cli/src/serve/hosted-never-started-reconcile.ts index 3dec71081f8..27270c35a3b 100644 --- a/packages/cli/src/serve/hosted-never-started-reconcile.ts +++ b/packages/cli/src/serve/hosted-never-started-reconcile.ts @@ -38,15 +38,20 @@ export interface NeverStartedReconcileResult { readonly skipped: readonly string[]; } +type NeverStartedProbeResult = + | { readonly proven: true } + | { readonly proven: false; readonly cause: string }; + type RouteResolution = | { readonly runtimeSessionId: string } | { readonly failure: 'no_identity' | 'route_foreign' }; // The stderr diagnostic carries these fields and nothing more: the // rejection's message embeds the request URL with its session ids (and a -// mis-typed logger could leak the bearer token), so only the status/code -// pair is ever interpolated. -export function rejectionDetail(cause: unknown): string { +// mis-typed logger could leak the bearer token), so it is never interpolated. +// Only status/code, stable tokens for known contract errors, or an +// Error/DOMException name describe a rejection. +function rejectionDetail(cause: unknown): string { const bag = cause !== null && typeof cause === 'object' ? (cause as Record) @@ -54,7 +59,34 @@ export function rejectionDetail(cause: unknown): string { const parts: string[] = []; if (typeof bag['status'] === 'number') parts.push(`status ${bag['status']}`); if (typeof bag['code'] === 'string') parts.push(`code ${bag['code']}`); - return parts.length > 0 ? parts.join(', ') : 'no status'; + if (parts.length > 0) return parts.join(', '); + // A fetch timeout's DOMException has Error.prototype, so this arm reports + // its name; its numeric `code` cannot pass the route-code check above. + if (cause instanceof Error) { + if (cause.message === 'Runtime execution identity changed.') + return 'identity_changed'; + if (cause.message === 'Runtime execution outcome is unknown.') + return 'outcome_unknown'; + return `error ${cause.name}`; + } + return 'no status'; +} + +export async function probeNeverStarted( + broker: Pick, + executionCallId: string, +): Promise { + try { + const answer = await broker.backgroundProcessStatus(executionCallId); + return answer === undefined + ? { proven: true } + : { proven: false, cause: `ambiguous_state (${answer.state})` }; + } catch (cause) { + return { + proven: false, + cause: `probe_rejected (${rejectionDetail(cause)})`, + }; + } } function asDurableRef(value: unknown): ManagedSessionDurableRef | undefined { @@ -188,15 +220,9 @@ export async function reconcileNeverStartedChildRuns(input: { // the proof never needs one. for (const record of group) { const executionCallId = record.run.executionCallId ?? record.shellId; - let answer: { state: string } | undefined; - try { - answer = await broker.backgroundProcessStatus(executionCallId); - } catch (cause) { - skip(record, `probe_rejected (${rejectionDetail(cause)})`); - continue; - } - if (answer !== undefined) { - skip(record, `ambiguous_state (${answer.state})`); + const proof = await probeNeverStarted(broker, executionCallId); + if (!proof.proven) { + skip(record, proof.cause); continue; } await input.childRuns.settleFailed(record.shellId, { diff --git a/packages/cli/src/serve/hosted-workspace-tool-turn.test.ts b/packages/cli/src/serve/hosted-workspace-tool-turn.test.ts index bae0d21c1ef..8da8d58f29a 100644 --- a/packages/cli/src/serve/hosted-workspace-tool-turn.test.ts +++ b/packages/cli/src/serve/hosted-workspace-tool-turn.test.ts @@ -5573,6 +5573,8 @@ function backgroundTurnRig( notStartedProven?: boolean; reserveRefused?: boolean; publicationMissing?: boolean; + closeResults?: ReadonlyArray<'NOT_STARTED' | 'MISSING'>; + executionCallIds?: readonly string[]; } = {}, ) { enablement.childRun = true; @@ -5627,23 +5629,32 @@ function backgroundTurnRig( } if (route === '/grants' && operation === 'close_not_started') { order.push('close_not_started'); - if (options.publicationMissing) + const closeResult = + options.closeResults?.[ + order.filter((item) => item === 'close_not_started').length - 1 + ]; + if (options.publicationMissing || closeResult === 'MISSING') throw new ManagedSessionStoreHttpError( 404, 'publication_not_found', 'Publication does not exist', ); return { - state: options.notStartedProven === false ? 'RUNNING' : 'NOT_STARTED', + state: + closeResult ?? + (options.notStartedProven === false ? 'RUNNING' : 'NOT_STARTED'), }; } throw new Error('Unexpected publication route ' + route); }), } as unknown as HttpToolPublicationOwner; - broker.prepareV3.mockResolvedValue({ - executionCallId: 'shell-execution', - runtimeBindingId: 'binding-1', - bindingGeneration: '1', + broker.prepareV3.mockImplementation(async () => { + const index = broker.prepareV3.mock.calls.length - 1; + return { + executionCallId: options.executionCallIds?.[index] ?? 'shell-execution', + runtimeBindingId: `binding-${index + 1}`, + bindingGeneration: '1', + }; }); broker.executeV3.mockImplementation(async () => { order.push('execute'); @@ -6803,6 +6814,9 @@ it('settles the admitted child run when grant reservation exhausts its retries', }); it('settles an in-turn dispatch with no publication and no durable process row', async () => { + const stderr = vi + .spyOn(stdio, 'writeStderrLineSafe') + .mockImplementation(() => {}); const { call, parts } = backgroundCall(); const rig = backgroundTurnRig( { executionStatus: 'success', responseParts: [], capture: null }, @@ -6826,6 +6840,258 @@ it('settles an in-turn dispatch with no publication and no durable process row', expect(broker.backgroundProcessStatus).toHaveBeenCalledWith( 'shell-execution', ); + expect(stderr).toHaveBeenCalledWith( + 'qwen serve: Settled never-started background child run shell-execution from durable process-ledger absence.', + ); + stderr.mockRestore(); +}); + +it('probes every inconclusive background binding within one timeout window', async () => { + const realSetTimeout = globalThis.setTimeout; + const realClearTimeout = globalThis.clearTimeout; + vi.useFakeTimers(); + let guard: ReturnType | undefined; + try { + const template = backgroundCall().call; + const backgroundCalls = [1, 2].map((index) => ({ + ...template, + callId: `background-call-${index}`, + })); + const backgroundParts: Part[] = backgroundCalls.map((call) => ({ + functionCall: { + id: call.callId, + name: call.name, + args: call.args, + }, + })); + const rig = backgroundTurnRig( + { executionStatus: 'success', responseParts: [], capture: null }, + { + publicationMissing: true, + executionCallIds: ['shell-execution-1', 'shell-execution-2'], + }, + ); + turn = rig.turn; + broker.executeV3.mockRejectedValue(new Error('trigger recovery')); + let resolveBothProbes!: () => void; + const bothProbesStarted = new Promise((resolve) => { + resolveBothProbes = resolve; + }); + broker.backgroundProcessStatus.mockImplementation( + () => + new Promise((_resolve, reject) => { + if (broker.backgroundProcessStatus.mock.calls.length === 2) + resolveBothProbes(); + setTimeout( + () => reject(new DOMException('timed out', 'TimeoutError')), + 30_000, + ); + }), + ); + let completed = false; + const outcome = turn + .execute( + backgroundCalls, + backgroundParts, + 'model', + new AbortController().signal, + ) + .then( + () => ({ rejection: undefined }), + (rejection: unknown) => ({ rejection }), + ) + .finally(() => { + completed = true; + }); + const guardedStart = new Promise((_resolve, reject) => { + guard = realSetTimeout( + () => reject(new Error('both probes did not start concurrently')), + 1_000, + ); + }); + + await Promise.race([bothProbesStarted, guardedStart]); + if (guard !== undefined) realClearTimeout(guard); + guard = undefined; + expect(broker.backgroundProcessStatus).toHaveBeenCalledTimes(2); + expect(completed).toBe(false); + + await vi.advanceTimersByTimeAsync(29_999); + expect(completed).toBe(false); + + await vi.advanceTimersByTimeAsync(1); + const { rejection } = await outcome; + expect(completed).toBe(true); + expect(rejection).toBeInstanceOf(HostedToolRecoveryRequiredError); + expect( + rig.order.filter((item) => item === 'close_not_started'), + ).toHaveLength(2); + expect( + rig.orchestrator.calls.filter(([name]) => name === 'settleFailed'), + ).toHaveLength(0); + } finally { + if (guard !== undefined) realClearTimeout(guard); + vi.useRealTimers(); + turn = createTurn(); + } +}); + +it('settles grant-proven bindings before waiting for process probes', async () => { + const stderr = vi + .spyOn(stdio, 'writeStderrLineSafe') + .mockImplementation(() => {}); + const realSetTimeout = globalThis.setTimeout; + const realClearTimeout = globalThis.clearTimeout; + vi.useFakeTimers(); + let guard: ReturnType | undefined; + try { + const template = backgroundCall().call; + const backgroundCalls = [1, 2].map((index) => ({ + ...template, + callId: `background-call-${index}`, + })); + const backgroundParts: Part[] = backgroundCalls.map((call) => ({ + functionCall: { + id: call.callId, + name: call.name, + args: call.args, + }, + })); + const rig = backgroundTurnRig( + { executionStatus: 'success', responseParts: [], capture: null }, + { + closeResults: ['NOT_STARTED', 'MISSING'], + executionCallIds: ['shell-execution-1', 'shell-execution-2'], + }, + ); + turn = rig.turn; + broker.executeV3.mockRejectedValue(new Error('trigger recovery')); + let resolveProbeStarted!: () => void; + const probeStarted = new Promise((resolve) => { + resolveProbeStarted = resolve; + }); + broker.backgroundProcessStatus.mockImplementation(() => { + resolveProbeStarted(); + return new Promise((_resolve, reject) => { + setTimeout( + () => reject(new DOMException('timed out', 'TimeoutError')), + 30_000, + ); + }); + }); + const outcome = turn + .execute( + backgroundCalls, + backgroundParts, + 'model', + new AbortController().signal, + ) + .then( + () => ({ rejection: undefined }), + (rejection: unknown) => ({ rejection }), + ); + const guardedStart = new Promise((_resolve, reject) => { + guard = realSetTimeout( + () => reject(new Error('process probe did not start')), + 1_000, + ); + }); + + await Promise.race([probeStarted, guardedStart]); + if (guard !== undefined) realClearTimeout(guard); + guard = undefined; + expect(broker.backgroundProcessStatus).toHaveBeenCalledOnce(); + expect(rig.orchestrator.calls.at(-1)).toEqual([ + 'settleFailed', + { + id: 'shell-execution-1', + params: { stopReason: 'start_failed', started: false }, + }, + ]); + // Grant-owner proof is not process-ledger proof. + expect(stderr).not.toHaveBeenCalledWith( + expect.stringContaining('from durable process-ledger absence'), + ); + + await vi.advanceTimersByTimeAsync(30_000); + const { rejection } = await outcome; + expect(rejection).toBeInstanceOf(HostedToolRecoveryRequiredError); + expect( + rig.orchestrator.calls.filter(([name]) => name === 'settleFailed'), + ).toHaveLength(1); + } finally { + if (guard !== undefined) realClearTimeout(guard); + vi.useRealTimers(); + turn = createTurn(); + } +}); + +it('settles only the binding whose own probe proves absence in a mixed fan-out', async () => { + const stderr = vi + .spyOn(stdio, 'writeStderrLineSafe') + .mockImplementation(() => {}); + try { + const template = backgroundCall().call; + const backgroundCalls = [1, 2].map((index) => ({ + ...template, + callId: `background-call-${index}`, + })); + const backgroundParts: Part[] = backgroundCalls.map((call) => ({ + functionCall: { + id: call.callId, + name: call.name, + args: call.args, + }, + })); + const rig = backgroundTurnRig( + { executionStatus: 'success', responseParts: [], capture: null }, + { + publicationMissing: true, + executionCallIds: ['shell-execution-1', 'shell-execution-2'], + }, + ); + turn = rig.turn; + broker.executeV3.mockRejectedValue(new Error('trigger recovery')); + broker.backgroundProcessStatus.mockImplementation(async () => + broker.backgroundProcessStatus.mock.calls.length === 1 + ? { state: 'executing' } + : undefined, + ); + const outcome = turn + .execute( + backgroundCalls, + backgroundParts, + 'model', + new AbortController().signal, + ) + .then( + () => ({ rejection: undefined }), + (rejection: unknown) => ({ rejection }), + ); + + const { rejection } = await outcome; + expect(rejection).toBeInstanceOf(HostedToolRecoveryRequiredError); + expect(broker.backgroundProcessStatus).toHaveBeenCalledTimes(2); + expect( + rig.orchestrator.calls.filter(([name]) => name === 'settleFailed'), + ).toEqual([ + [ + 'settleFailed', + { + id: 'shell-execution-2', + params: { stopReason: 'start_failed', started: false }, + }, + ], + ]); + expect(stderr).toHaveBeenCalledWith( + 'qwen serve: Background process absence could not be confirmed for shell-execution-1 (ambiguous_state (executing)).', + ); + expect(stderr).toHaveBeenCalledWith( + 'qwen serve: Settled never-started background child run shell-execution-2 from durable process-ledger absence.', + ); + } finally { + stderr.mockRestore(); + } }); it('keeps an in-flight dispatch blocked when cancellation fails before process admission', async () => { @@ -6880,10 +7146,12 @@ it.each(['present', 'unavailable'] as const)( 'admit', 'dispatchStarted', ]); - if (proof === 'unavailable') - expect(stderr).toHaveBeenCalledWith( - 'qwen serve: Background process absence could not be confirmed for shell-execution (status 503, code store_unavailable).', - ); + expect(stderr).toHaveBeenCalledWith( + proof === 'present' + ? 'qwen serve: Background process absence could not be confirmed for shell-execution (ambiguous_state (executing)).' + : 'qwen serve: Background process absence could not be confirmed for shell-execution (probe_rejected (status 503, code store_unavailable)).', + ); + stderr.mockRestore(); }, ); diff --git a/packages/cli/src/serve/hosted-workspace-tool-turn.ts b/packages/cli/src/serve/hosted-workspace-tool-turn.ts index 1dc9e94b434..a65aa331183 100644 --- a/packages/cli/src/serve/hosted-workspace-tool-turn.ts +++ b/packages/cli/src/serve/hosted-workspace-tool-turn.ts @@ -9,7 +9,7 @@ import { reconcileHostedTurnCleanup, confirmReplacedHostedTurnOwner, } from './hosted-turn-cleanup.js'; -import { rejectionDetail } from './hosted-never-started-reconcile.js'; +import { probeNeverStarted } from './hosted-never-started-reconcile.js'; import { readHostedFileHistory, @@ -4660,13 +4660,53 @@ export class HostedWorkspaceToolTurn { if (result.status === 'fulfilled') cancellationBarriers.add(reservedIds[index]!); }); + const probeBindings: Array<{ + readonly executionCallId: string; + readonly request: (typeof requests)[number] | undefined; + }> = []; + const settleNeverStarted = async ( + executionCallId: string, + request: (typeof requests)[number] | undefined, + processProven: boolean, + ) => { + if (request?.background) { + try { + if (this.childRuns !== undefined) { + await this.childRuns.settleFailed(executionCallId, { + stopReason: 'start_failed', + started: false, + }); + if (processProven) + writeStderrLineSafe( + `qwen serve: Settled never-started background child run ${executionCallId} from durable process-ledger absence.`, + ); + } + } catch (settleCause) { + writeStderrLineSafe( + 'qwen serve: Unstarted child run record was not settled: ' + + String(settleCause), + ); + } + } + if (request?.monitoring) + await this.monitors + ?.settleFailed(executionCallId, { + stopReason: 'start_failed', + started: false, + }) + .catch((settleCause: unknown) => + writeStderrLineSafe( + 'qwen serve: Unstarted monitor run record was not settled: ' + + String(settleCause), + ), + ); + }; for (const executionCallId of shellBindings.keys()) { const saved = shellBindings.get(executionCallId); if (!saved) continue; const request = requests.find( (_, index) => reserved.get(index) === executionCallId, ); - let proven = false; let grantOwnerState: unknown; try { const owner = this.publication!.owner; @@ -4685,57 +4725,39 @@ export class HostedWorkspaceToolTurn { grantOwnerState = (closed as Record)['state']; if (grantOwnerState !== 'NOT_STARTED') throw new Error('Original execution was not proven unstarted.'); - proven = true; } catch (closeCause) { writeStderrLineSafe( 'qwen serve: Tool publication close was not confirmed: ' + String(closeCause), ); } - if ( - !proven && - grantOwnerState === undefined && - request?.background && - cancellationBarriers.has(executionCallId) - ) { - try { - proven = - (await this.broker.backgroundProcessStatus(executionCallId)) === - undefined; - } catch (cause) { - writeStderrLineSafe( - `qwen serve: Background process absence could not be confirmed for ${executionCallId} (${rejectionDetail(cause)}).`, - ); - } + if (grantOwnerState === 'NOT_STARTED') { + await settleNeverStarted(executionCallId, request, false); + continue; } - if (!proven) continue; - // The grant owner proved a Shell or Monitor start never happened, - // or the original process ledger proved it for a background Shell. - // Publication absence alone proves nothing. - if (request?.background) - await this.childRuns - ?.settleFailed(executionCallId, { - stopReason: 'start_failed', - started: false, - }) - .catch((settleCause: unknown) => - writeStderrLineSafe( - 'qwen serve: Unstarted child run record was not settled: ' + - String(settleCause), - ), - ); - if (request?.monitoring) - await this.monitors - ?.settleFailed(executionCallId, { - stopReason: 'start_failed', - started: false, - }) - .catch((settleCause: unknown) => - writeStderrLineSafe( - 'qwen serve: Unstarted monitor run record was not settled: ' + - String(settleCause), - ), - ); + if ( + grantOwnerState === undefined && + request?.background === true && + cancellationBarriers.has(executionCallId) + ) + probeBindings.push({ executionCallId, request }); + } + const probeOutcomes = await Promise.all( + probeBindings.map(async (binding) => ({ + ...binding, + probe: await probeNeverStarted(this.broker, binding.executionCallId), + })), + ); + for (const { executionCallId, request, probe } of probeOutcomes) { + if (!probe.proven) { + writeStderrLineSafe( + `qwen serve: Background process absence could not be confirmed for ${executionCallId} (${probe.cause}).`, + ); + continue; + } + // The original process ledger proved the background Shell never + // started. Publication absence alone proves nothing. + await settleNeverStarted(executionCallId, request, true); } throw new HostedToolRecoveryRequiredError(cause); } finally {