diff --git a/docs/api/mcp.md b/docs/api/mcp.md index 47c11914..3cb78888 100644 --- a/docs/api/mcp.md +++ b/docs/api/mcp.md @@ -3836,6 +3836,22 @@ Epoch ms from the durable terminal record — the resolution a progress-based st > `readonly` **perWorker**: [`Budget`](index.md#budget-4) +##### onStop? + +> `readonly` `optional` **onStop?**: (`reason`) => `void` + +Called once when this manager declares completion through `stop` or an accepted submission. + +###### Parameters + +###### reason + +`string` \| `undefined` + +###### Returns + +`void` + ##### deliverable? > `readonly` `optional` **deliverable?**: [`DeliverableSpec`](runtime.md#deliverablespec)\<`unknown`\> diff --git a/docs/api/runtime.md b/docs/api/runtime.md index bc408d0b..644a2014 100644 --- a/docs/api/runtime.md +++ b/docs/api/runtime.md @@ -14650,6 +14650,12 @@ The standing instruction assembled from the profile: its system prompt in either `string` +###### stopSignal? + +`AbortSignal` + +Fires when the coordination server accepts a result or declares completion. + ###### coordinationTools readonly `Omit`\<[`McpToolDescriptor`](mcp.md#mcptooldescriptor), `"handler"`\>[] @@ -24383,6 +24389,12 @@ Stand up the coordination MCP over a live scope. The HOST address is `127.0.0.1` Independent completion check exposed to the driver as `submit_result`. +###### onStop? + +(`reason`) => `void` + +Called once when the external manager accepts a result or declares completion. + ###### maxLiveWorkers? `number` diff --git a/src/mcp/tools/coordination.ts b/src/mcp/tools/coordination.ts index c38f6a8b..99c82566 100644 --- a/src/mcp/tools/coordination.ts +++ b/src/mcp/tools/coordination.ts @@ -337,6 +337,8 @@ export interface CoordinationToolsOptions { readonly blobs: ResultBlobStore readonly makeWorkerAgent: MakeWorkerAgent readonly perWorker: Budget + /** Called once when this manager declares completion through `stop` or an accepted submission. */ + readonly onStop?: (reason: string | undefined) => void /** * The same independent completion check used for workers. When present, the driver receives a * `submit_result` tool and may finish work itself instead of being forced to delegate it. The @@ -716,12 +718,19 @@ export function createCoordinationTools(opts: CoordinationToolsOptions): Coordin const deliverable = opts.deliverable let stopped = false let reason: string | undefined + let stopNotified = false let submitted: { readonly result: unknown } | undefined let questionSeq = 0 const ledger: SettledWorker[] = [] const questions: QuestionRecord[] = [...(opts.priorQuestions ?? [])] const questionPolicy = opts.questionPolicy ?? 'auto' + const notifyStop = (): void => { + if (stopNotified) return + stopNotified = true + opts.onStop?.(reason) + } + // Keyed-assignment bookkeeping for the live-worker fence. `completedKeys` is every key this run // can already answer from committed work — seeded from the prior journal on a resume, extended as // keyed workers deliver in THIS process. A spawn under such a key starts nothing and occupies no @@ -2243,6 +2252,7 @@ export function createCoordinationTools(opts: CoordinationToolsOptions): Coordin submitted = Object.freeze({ result }) stopped = true reason = 'result-accepted' + notifyStop() return { accepted: true, retained: 'this-result', stop: true } }, } satisfies McpToolDescriptor, @@ -2267,6 +2277,7 @@ export function createCoordinationTools(opts: CoordinationToolsOptions): Coordin stopped = true const r = obj(raw).reason reason = typeof r === 'string' ? r : undefined + notifyStop() return Promise.resolve({ stopped: true }) }, }, diff --git a/src/runtime/supervise/coordination-mcp.ts b/src/runtime/supervise/coordination-mcp.ts index 249ecd4a..2b7cf90f 100644 --- a/src/runtime/supervise/coordination-mcp.ts +++ b/src/runtime/supervise/coordination-mcp.ts @@ -85,6 +85,8 @@ export async function serveCoordinationMcp(opts: { perWorker: Budget /** Independent completion check exposed to the driver as `submit_result`. */ deliverable?: DeliverableSpec + /** Called once when the external manager accepts a result or declares completion. */ + onStop?: (reason: string | undefined) => void /** Hard cap on simultaneously-LIVE workers — `spawn_agent` fails closed once this many are in * flight (a concurrency fence on top of the conserved-pool fence). Omit/`<= 0` = no cap. */ maxLiveWorkers?: number @@ -147,6 +149,7 @@ export async function serveCoordinationMcp(opts: { ...(opts.authorizeDownMessage ? { authorizeDownMessage: opts.authorizeDownMessage } : {}), perWorker: opts.perWorker, ...(opts.deliverable ? { deliverable: opts.deliverable } : {}), + ...(opts.onStop ? { onStop: opts.onStop } : {}), ...(opts.maxLiveWorkers !== undefined ? { maxLiveWorkers: opts.maxLiveWorkers } : {}), awaitTimeoutMs: opts.awaitTimeoutMs ?? DEFAULT_AWAIT_EVENT_TIMEOUT_MS, ...(opts.analysts ? { analysts: opts.analysts } : {}), diff --git a/src/runtime/supervise/supervise.ts b/src/runtime/supervise/supervise.ts index db3ff72c..a1bc6835 100644 --- a/src/runtime/supervise/supervise.ts +++ b/src/runtime/supervise/supervise.ts @@ -82,6 +82,7 @@ import { captureReusableExecutorConfig, createExecutor, type ExecutorConfig, + mergeAbortSignals, snapshotExecutorConfig, } from './runtime' import { @@ -331,6 +332,7 @@ function driveHarnessFromBackend( task, scope, coordinationMcpUrl, + stopSignal, coordinationTools, }) => { const initialBudget = scope.budget @@ -537,7 +539,9 @@ function driveHarnessFromBackend( } started = true - const run = executor.execute(task, scope.signal) + const runSignal = + stopSignal === undefined ? scope.signal : mergeAbortSignals(scope.signal, stopSignal) + const run = executor.execute(task, runSignal) if (isAsyncIterable(run)) { for await (const event of run) { if (event.kind === 'iteration') { diff --git a/src/runtime/supervise/supervisor-agent.ts b/src/runtime/supervise/supervisor-agent.ts index 29a3fb7e..a8093e9b 100644 --- a/src/runtime/supervise/supervisor-agent.ts +++ b/src/runtime/supervise/supervisor-agent.ts @@ -241,6 +241,8 @@ export interface DriveHarness { readonly task: unknown readonly scope: Scope readonly coordinationMcpUrl: string + /** Fires when the coordination server accepts a result or declares completion. */ + readonly stopSignal?: AbortSignal /** Data-only product tool surface mounted on the coordination MCP. Runtime-owned drivers include * this in their materialization evidence without persisting executable handlers. */ readonly coordinationTools: ReadonlyArray> @@ -572,6 +574,7 @@ function buildSupervisorAgent( ? await bindSupervisorTools(resolveTools, context, scope.signal) : undefined const onEvent = bindSupervisorNodeObserver(context, observeNodeEvent, deps.onEvent) + const stopController = new AbortController() const mcp = await serveCoordinationMcp({ scope, blobs: deps.blobs, @@ -587,6 +590,11 @@ function buildSupervisorAgent( ? { allowUnauthenticatedRemote: true } : {}), ...(deps.deliverable ? { deliverable: deps.deliverable } : {}), + onStop: (reason) => { + if (!stopController.signal.aborted) { + stopController.abort(reason ?? 'coordination stop') + } + }, ...(deps.maxLiveWorkers !== undefined ? { maxLiveWorkers: deps.maxLiveWorkers } : {}), ...(deps.analysts ? { analysts: deps.analysts } : {}), ...(deps.analyzeOnSettle ? { analyzeOnSettle: deps.analyzeOnSettle } : {}), @@ -616,6 +624,7 @@ function buildSupervisorAgent( task, scope, coordinationMcpUrl: mcp.url, + stopSignal: stopController.signal, coordinationTools: (nodeTools ?? []).map(({ name, description, inputSchema }) => ({ name, description, @@ -626,7 +635,7 @@ function buildSupervisorAgent( // Once the injected check has accepted a result, a later backend shutdown/timeout // cannot erase that completed work — and there is nothing left to retry FOR. Without // an accepted submission the backend error propagates into the retry decision. - if (!mcp.submittedResult()) throw error + if (!mcp.submittedResult() && !mcp.isStopped()) throw error } }, progress: () => ({ diff --git a/tests/kernel/coordination.test.ts b/tests/kernel/coordination.test.ts index 3a18c0a3..aa06eab0 100644 --- a/tests/kernel/coordination.test.ts +++ b/tests/kernel/coordination.test.ts @@ -149,11 +149,13 @@ describe('coordination tools', () => { expect(withoutCheck.tools.map((t) => t.name)).not.toContain('submit_result') const checked: unknown[] = [] + const stopReasons: Array = [] const withCheck = createCoordinationTools({ scope, blobs, makeWorkerAgent, perWorker: { maxIterations: 1, maxTokens: 10 }, + onStop: (reason) => stopReasons.push(reason), deliverable: { describe: 'an object whose answer is 42', check(result) { @@ -193,7 +195,11 @@ describe('coordination tools', () => { retained: 'earlier-passing-result', stop: true, }) + expect(await tool(withCheck, 'stop').handler({ reason: 'redundant-stop' })).toEqual({ + stopped: true, + }) expect(checked).toHaveLength(3) + expect(stopReasons).toEqual(['result-accepted']) expect(withCheck.submittedResult()).toEqual({ result: { answer: 42 } }) }) diff --git a/tests/kernel/supervisor-agent.test.ts b/tests/kernel/supervisor-agent.test.ts index ea993b18..ef33c185 100644 --- a/tests/kernel/supervisor-agent.test.ts +++ b/tests/kernel/supervisor-agent.test.ts @@ -188,6 +188,55 @@ describe('supervisorAgent — the brain is resolved from profile.harness (backen if (result.kind === 'winner') expect(result.out).toEqual({ answer: 42 }) }) + it('SANDBOX arm stops the active harness before a provider turn after accepted submission', async () => { + const blobs = new InMemoryResultBlobStore() + const journal = new InMemorySpawnJournal() + let providerCalls = 0 + let observedStop = false + const driveHarness: DriveHarness = async ({ coordinationMcpUrl, stopSignal }) => { + await jsonRpc(coordinationMcpUrl, 'tools/call', { + name: 'submit_result', + arguments: { result: { answer: 42 } }, + }) + await new Promise((resolve) => { + const timer = setTimeout(() => { + providerCalls += 1 + resolve() + }, 10) + if (stopSignal === undefined) return + const stopped = () => { + clearTimeout(timer) + observedStop = true + resolve() + } + if (stopSignal.aborted) stopped() + else stopSignal.addEventListener('abort', stopped, { once: true }) + }) + if (providerCalls > 0) throw new Error('provider call after accepted submission') + } + const root = supervisorAgent( + testAgentProfile('sup', { + harness: 'pi', + prompt: { systemPrompt: 'solve or delegate' }, + }), + { + blobs, + makeWorkerAgent: () => deliveringLeaf('unused', {}), + perWorker, + driveHarness, + deliverable: { + describe: 'an object whose answer is 42', + check: (result) => (result as { answer?: unknown }).answer === 42, + }, + }, + ) + + const result = await runSupervisor(root, blobs, journal) + expect(result.kind).toBe('winner') + expect(observedStop).toBe(true) + expect(providerCalls).toBe(0) + }) + it('fails loud when a sandboxed-harness supervisor has no driveHarness substrate', () => { const blobs = new InMemoryResultBlobStore() expect(() =>