fix(bridge): finish active turn after accepted result

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<unknown> = (spec, ctx) => {
   const base = readSeam<BridgeSeam>(ctx, bridgeSeamKey, 'bridge')
+  const stopSignal = readOptionalAbortSignal(ctx, bridgeStopSignalKey, 'bridge')
   const modelCredential = validateBridgeModelCredential(
     base.modelCredential,
     base.bridgeUrl,
@@ -1935,6 +1939,7 @@ export const bridgeExecutor: ExecutorFactory<unknown> = (spec, ctx) => {
       return streamBridgeSession({
         task,
         signal,
+        ...(stopSignal === undefined ? {} : { stopSignal }),
         profile: effectiveProfile,
         seam,
         sessionId,
@@ -1997,6 +2002,8 @@ export const bridgeExecutor: ExecutorFactory<unknown> = (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<Usage
   // on the SAME session. `nextPrompt` is undefined once there's nothing pending.
   let nextPrompt: string | undefined = taskToPrompt(args.task)
   for (let t = 0; args.maxTurns === 0 || t < args.maxTurns; t += 1) {
+    if (args.stopSignal?.aborted) {
+      observation.note = 'settled after completion request'
+      break
+    }
     // Drain queued down-messages; on turns > 0 they ARE the prompt (resume content).
     const pending = inbox.drain()
     if (pending.length) {
@@ -2356,6 +2367,13 @@ async function* streamBridgeSession(args: StreamBridgeArgs): AsyncIterable<Usage
       }
     }
     try {
+      // A completion request may arrive while the previous turn is still streaming. The current
+      // request owns the provider receipt and terminal materialization, so let it drain before
+      // checking the request at the next turn boundary.
+      if (args.stopSignal?.aborted) {
+        observation.note = 'settled after completion request'
+        break
+      }
       args.onProviderAttemptStart()
       for await (const chunk of streamDurableBridgeRun({
         seam,
@@ -2531,6 +2549,11 @@ async function* streamBridgeSession(args: StreamBridgeArgs): AsyncIterable<Usage
       continue
     }
 
+    if (args.stopSignal?.aborted) {
+      observation.note = 'settled after completion request'
+      break
+    }
+
     // Before settling, drain once more — the worker can't finish while a steer it
     // never read is pending (the sandbox/router settle contract). A pending steer
     // becomes the next resume turn; otherwise the session is truly done.
@@ -4536,6 +4559,24 @@ function readSeam<T>(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

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<UsageEvent>(run)) {
         for await (const event of run) {
           if (event.kind === 'iteration') {

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<Record<string, unknown>> = []
     let deliver: (message: unknown) => void = () => {}