From d0daa350cfd1c0d1daaf07ef183934075c6c47c9 Mon Sep 17 00:00:00 2001 From: Drew Stone Date: Thu, 13 Aug 2026 15:50:38 -0600 Subject: [PATCH] fix(bridge): finish active turn after accepted result --- src/runtime/supervise/runtime.ts | 41 ++++++++ src/runtime/supervise/supervise.ts | 10 +- tests/runtime/bridge-executor.test.ts | 136 +++++++++++++++++++++++++- 3 files changed, 181 insertions(+), 6 deletions(-) diff --git a/src/runtime/supervise/runtime.ts b/src/runtime/supervise/runtime.ts index bd18df24..196acc70 100644 --- a/src/runtime/supervise/runtime.ts +++ b/src/runtime/supervise/runtime.ts @@ -380,6 +380,9 @@ const routerSeamKey = 'router' const sandboxSeamKey = 'sandbox' const cliSeamKey = 'cli' const bridgeSeamKey = 'bridge' +/** Internal control seam used by Runtime-owned external supervisors. A completion request stops + * the next bridge turn; it must not abort the paid request that is already streaming. */ +export const bridgeStopSignalKey = '__bridge_stop_signal' const maxBridgeTimeoutMs = 2_147_483_647 const bridgeModelCredentialHeader = 'x-cli-bridge-model-credential' const bridgeModelBaseUrlHeader = 'x-cli-bridge-model-base-url' @@ -1791,6 +1794,7 @@ function bridgeProfileModel(profile: AgentProfile, context: string): string { export const bridgeExecutor: ExecutorFactory = (spec, ctx) => { const base = readSeam(ctx, bridgeSeamKey, 'bridge') + const stopSignal = readOptionalAbortSignal(ctx, bridgeStopSignalKey, 'bridge') const modelCredential = validateBridgeModelCredential( base.modelCredential, base.bridgeUrl, @@ -1935,6 +1939,7 @@ export const bridgeExecutor: ExecutorFactory = (spec, ctx) => { return streamBridgeSession({ task, signal, + ...(stopSignal === undefined ? {} : { stopSignal }), profile: effectiveProfile, seam, sessionId, @@ -1997,6 +2002,8 @@ export const bridgeExecutor: ExecutorFactory = (spec, ctx) => { interface StreamBridgeArgs { task: unknown signal: AbortSignal + /** Completion request from a Runtime-owned driver. It stops future turns after the current one. */ + stopSignal?: AbortSignal profile: AgentProfile seam: ResolvedBridgeSeam sessionId: string @@ -2268,6 +2275,10 @@ async function* streamBridgeSession(args: StreamBridgeArgs): AsyncIterable 0 they ARE the prompt (resume content). const pending = inbox.drain() if (pending.length) { @@ -2356,6 +2367,13 @@ async function* streamBridgeSession(args: StreamBridgeArgs): AsyncIterable(ctx: ExecutorContext, key: string, who: string): T { return seam as T } +function readOptionalAbortSignal( + ctx: ExecutorContext, + key: string, + who: string, +): AbortSignal | undefined { + const value = ctx.seams[key] + if (value === undefined) return undefined + if ( + value === null || + typeof value !== 'object' || + typeof (value as { aborted?: unknown }).aborted !== 'boolean' || + typeof (value as { addEventListener?: unknown }).addEventListener !== 'function' + ) { + throw new ValidationError(`${who} executor: seam "${key}" must be an AbortSignal`) + } + return value as AbortSignal +} + /** A leaf task is opaque (`unknown`). A string is the prompt verbatim; an object * with a `prompt`/`content`/`task` string field uses it; otherwise it serializes. * Module-exported (not package surface) so sibling leaf executors read a task diff --git a/src/runtime/supervise/supervise.ts b/src/runtime/supervise/supervise.ts index a1bc6835..8583b416 100644 --- a/src/runtime/supervise/supervise.ts +++ b/src/runtime/supervise/supervise.ts @@ -79,10 +79,10 @@ import { import { createFileRunContext, createInMemoryRunContext } from './run-context' import { bindReusableExecutorExecutionId, + bridgeStopSignalKey, captureReusableExecutorConfig, createExecutor, type ExecutorConfig, - mergeAbortSignals, snapshotExecutorConfig, } from './runtime' import { @@ -373,7 +373,7 @@ function driveHarnessFromBackend( const executor = baseFactory(spec, { signal: scope.signal, node: scopeOwnerExecutorNodeContext(scope), - seams: {}, + seams: stopSignal === undefined ? {} : { [bridgeStopSignalKey]: stopSignal }, }) activeExecutor = executor let completed = false @@ -539,9 +539,9 @@ function driveHarnessFromBackend( } started = true - const runSignal = - stopSignal === undefined ? scope.signal : mergeAbortSignals(scope.signal, stopSignal) - const run = executor.execute(task, runSignal) + // A coordination completion stops the NEXT external turn. The active bridge request must + // drain so its served model and terminal materialization remain valid evidence. + const run = executor.execute(task, scope.signal) if (isAsyncIterable(run)) { for await (const event of run) { if (event.kind === 'iteration') { diff --git a/tests/runtime/bridge-executor.test.ts b/tests/runtime/bridge-executor.test.ts index 9d93d6d8..4391ee3c 100644 --- a/tests/runtime/bridge-executor.test.ts +++ b/tests/runtime/bridge-executor.test.ts @@ -12,9 +12,14 @@ import { createExecutor, type ExecutorConfig, inlineSandboxClient } from '../../ import { createBudgetPool } from '../../src/runtime/supervise/budget' import { runtimeOwnedExecutorMaterialization, + runtimeOwnedExecutorProviderEvidence, runtimeOwnedPendingExecutorMaterialization, } from '../../src/runtime/supervise/materialization' -import { bridgeExecutor, createExecutorRegistry } from '../../src/runtime/supervise/runtime' +import { + bridgeExecutor, + bridgeStopSignalKey, + createExecutorRegistry, +} from '../../src/runtime/supervise/runtime' import { createScope } from '../../src/runtime/supervise/scope' import { workerFromBackend } from '../../src/runtime/supervise/supervise' import type { Agent, AgentSpec, Executor, UsageEvent } from '../../src/runtime/supervise/types' @@ -391,6 +396,135 @@ describe('bridgeExecutor over node:http', () => { expect(executor.resultArtifact().spent.iterations).toBe(1) }) + it('drains an active turn after completion and blocks the next bridge request', async () => { + let requests = 0 + const stop = new AbortController() + let deliver: (message: unknown) => void = () => {} + bridgeHttpHandler = () => { + requests += 1 + deliver({ steer: 'a queued continuation must not start' }) + if (!activeBridgePayload) throw new Error('active bridge payload missing') + const payload = activeBridgePayload + const profile = payload.agent_profile as AgentProfile + const model = String(payload.model) + const stream = new PassThrough() + stream.write( + `id: 1\ndata: ${JSON.stringify({ + model, + choices: [{ delta: { content: 'accepted answer' } }], + })}\n\n`, + ) + setImmediate(() => { + stop.abort('result accepted') + stream.end( + [ + `id: 2\ndata: ${JSON.stringify({ + model, + usage: { + prompt_tokens: 7, + completion_tokens: 3, + cost: 0.001, + cost_known: true, + cost_provenance: 'provider-receipt', + }, + })}\n\n`, + `id: 3\ndata: ${JSON.stringify({ + profile_materialization: { + schema: 'cli-bridge.profile-materialization.v2', + effectiveProfileDigest: canonicalAgentProfileDigest(profile), + harness: 'pi', + provider: 'tangle-router', + model, + reasoningEffort: { requested: null, applied: null }, + workspacePlanDigest: `sha256:${'b'.repeat(64)}`, + files: [], + unsupported: [], + }, + })}\n\n`, + 'data: [DONE]\n\n', + ].join(''), + ) + }) + return stream + } + const profile: AgentProfile = { + name: 'completion-stop-worker', + harness: 'pi', + model: { + provider: 'tangle-router', + default: 'deepseek-v4-flash', + metadata: { maxTurns: 3 }, + }, + } + const executor = bridgeExecutor( + { profile, harness: null }, + { + signal: new AbortController().signal, + seams: { + bridge: { bridgeUrl: 'http://bridge.test', bridgeBearer: 'secret' }, + [bridgeStopSignalKey]: stop.signal, + }, + }, + ) + deliver = (message) => executor.deliver?.(message) + + await drainExecutor(executor) + + expect(requests).toBe(1) + expect(runtimeOwnedExecutorMaterialization(executor)).toMatchObject({ + plan: { + terminalAcknowledgement: { + model: 'pi/tangle-router/deepseek-v4-flash', + }, + }, + }) + expect(runtimeOwnedExecutorProviderEvidence(executor)).toMatchObject({ + status: 'known', + attempts: [{ observations: ['pi/tangle-router/deepseek-v4-flash'] }], + }) + expect(executor.resultArtifact().out).toMatchObject({ content: 'accepted answer' }) + }) + + it('keeps identity and materialization unknown when an active stream aborts before [DONE]', async () => { + const abort = new AbortController() + bridgeHttpHandler = () => { + const stream = new PassThrough() + stream.write(frame(1, { choices: [{ delta: { content: 'partial' } }] })) + queueMicrotask(() => { + abort.abort('bridge disconnected') + stream.end() + }) + return stream + } + const profile = exactBridgeProfile('ambiguous-abort-worker', 'safe-model') + const executor = bridgeExecutor( + { profile, harness: null }, + { + signal: new AbortController().signal, + seams: { bridge: { bridgeUrl: 'http://bridge.test', bridgeBearer: 'secret' } }, + }, + ) + + const run = executor.execute('go', abort.signal) + if (!isUsageStream(run)) throw new Error('bridge worker must stream usage') + await expect( + (async () => { + for await (const _event of run) { + // consume the partial response before the transport aborts + } + })(), + ).rejects.toThrow(/aborted/u) + + expect(runtimeOwnedExecutorProviderEvidence(executor)).toEqual({ + status: 'unknown', + attempts: [{ observations: [] }], + models: [], + reason: 'provider-model-missing', + }) + expect(runtimeOwnedExecutorMaterialization(executor)).toBeUndefined() + expect(runtimeOwnedPendingExecutorMaterialization(executor)).toBeDefined() + }) + it('forwards the profile completion cap on initial and resumed bridge requests', async () => { const seen: Array> = [] let deliver: (message: unknown) => void = () => {}