瀏覽代碼

Preserve interrupted post-tool context

Hypatia May 3 月之前
父節點
當前提交
edfc4dc4ac

+ 5 - 4
docs/core-data-structures/tools.md

@@ -151,10 +151,11 @@ interface ToolExecutionResult {
    * NEXT request (Claude Code's PostToolUse `additionalContext`). It is NOT part
    * NEXT request (Claude Code's PostToolUse `additionalContext`). It is NOT part
    * of this call's `content` — `content`/`feedback` shape the tool RESULT, but
    * of this call's `content` — `content`/`feedback` shape the tool RESULT, but
    * `additionalContext` is a SEPARATE `context/message`. A step can carry
    * `additionalContext` is a SEPARATE `context/message`. A step can carry
-   * multiple tool calls, so the loop BUFFERS every call's `additionalContext`
-   * and appends them only AFTER all `tool/result`s for the step, keeping
-   * tool-call/result adjacency intact. Carried on the result purely to ferry it
-   * from `execute()` up to the loop's per-step buffer.
+   * multiple tool calls, so the loop accepts every call's `additionalContext`
+   * into the active-batch FIFO and appends it only when that batch settles. A
+   * successful batch places context AFTER all its `tool/result`s; an interrupted
+   * batch places it after every recorded result and before turn close. Carried
+   * on the result purely to ferry it from `execute()` up to that FIFO.
    */
    */
   additionalContext?: HookContext
   additionalContext?: HookContext
   /**
   /**

+ 1 - 1
docs/rfc/implemented/feature/2026-06-30-interception-seams.md

@@ -36,7 +36,7 @@ Core dispatch and the tool body sit inside normalization boundaries, so tool, li
 
 
 1. **Open the turn before prompt policy.** A fully blocked batch becomes a zero-step `rejected` turn, preserving enclosure and giving ACP a durable terminal event. Every veto also records `prompt/blocked` with the original prompt and reason, so mixed batches retain blocked inputs. Allowed `additionalContext` is injected into the open turn.
 1. **Open the turn before prompt policy.** A fully blocked batch becomes a zero-step `rejected` turn, preserving enclosure and giving ACP a durable terminal event. Every veto also records `prompt/blocked` with the original prompt and reason, so mixed batches retain blocked inputs. Allowed `additionalContext` is injected into the open turn.
 
 
-2. **Post-tool `additionalContext` is buffered and appended AFTER all `tool/result`s.** `content`/`feedback` shape the result `execute()` returns, but `additionalContext` is a SEPARATE `context/message`, and a single step can carry multiple tool calls. Appending context right after each result would interleave `result(c1) → context → result(c2)` and break tool-call/result adjacency. So `execute()` surfaces `additionalContext` on its `ToolExecutionResult`, and the loop buffers every per-call context for the step and appends them as `context/message`(s) only after every `tool/result` is appended.
+2. **Post-tool `additionalContext` enters the active-batch FIFO and appends when that batch settles.** `content`/`feedback` shape the result `execute()` returns, but `additionalContext` is a SEPARATE `context/message`, and a single step can carry multiple tool calls. Appending context right after each result would interleave `result(c1) → context → result(c2)` and break tool-call/result adjacency. So `execute()` surfaces `additionalContext` on its `ToolExecutionResult`; the loop accepts each context into the same FIFO as asynchronous injections, then appends the FIFO after the complete result batch on success or after every recorded result before an interrupted turn closes.
 
 
 3. **A forced `continue` `reason` is enqueued through the steering channel**, so the next step's top-of-loop drain records it as steering for the continued turn — next-*step* steering within the SAME turn, not a next-*turn* prompt (matching the existing `hasSteering` force-continue override).
 3. **A forced `continue` `reason` is enqueued through the steering channel**, so the next step's top-of-loop drain records it as steering for the continued turn — next-*step* steering within the SAME turn, not a next-*turn* prompt (matching the existing `hasSteering` force-continue override).
 
 

+ 2 - 2
docs/tool-execution-pipeline.md

@@ -20,9 +20,9 @@ flowchart TD
   owned["Tool-owned session events<br/><code>todo/write</code>, <code>fs/observed</code>, <code>hook/invoked</code>, <code>hook/result</code>, <code>tool/code-dispatch</code>"]
   owned["Tool-owned session events<br/><code>todo/write</code>, <code>fs/observed</code>, <code>hook/invoked</code>, <code>hook/result</code>, <code>tool/code-dispatch</code>"]
   post["<code>tools/post-execute</code> waterfall<br/>accept, block, replace, add context"]
   post["<code>tools/post-execute</code> waterfall<br/>accept, block, replace, add context"]
   final["<code>tools/result</code> synchronous notification<br/>frozen authoritative outcome"]
   final["<code>tools/result</code> synchronous notification<br/>frozen authoritative outcome"]
-  context["Buffered additionalContext<br/>context/message after all tool results"]
+  context["Batch-deferred additionalContext<br/>context/message after recorded tool results"]
   toolResult["Session event: <code>tool/result</code><br/>single model-facing outcome"]
   toolResult["Session event: <code>tool/result</code><br/>single model-facing outcome"]
-  allResults["All calls in the step settled<br/>and tool/result events recorded"]
+  allResults["Tool batch settled<br/>recorded tool/result events complete"]
   presentResult["UI completed card<br/>presentResult(args, result)"]
   presentResult["UI completed card<br/>presentResult(args, result)"]
   model --> toolCall
   model --> toolCall
   toolCall --> presentCall
   toolCall --> presentCall

+ 14 - 4
packages/core/agent-loop/src/agent.ts

@@ -8,7 +8,7 @@
 
 
 import type { Context } from 'cordis'
 import type { Context } from 'cordis'
 import { agentEvents } from '@deepseek-ai/dsh-agent'
 import { agentEvents } from '@deepseek-ai/dsh-agent'
-import type { AgentId, AgentOptions, AgentStatus, SendOptions } from '@deepseek-ai/dsh-agent'
+import type { AgentId, AgentOptions, AgentStatus, HookContext, SendOptions } from '@deepseek-ai/dsh-agent'
 import type { Agent } from '@deepseek-ai/dsh-agent'
 import type { Agent } from '@deepseek-ai/dsh-agent'
 import { deepFreeze } from '@deepseek-ai/dsh-llm'
 import { deepFreeze } from '@deepseek-ai/dsh-llm'
 import type { ContentBlock, MessageSource } from '@deepseek-ai/dsh-llm'
 import type { ContentBlock, MessageSource } from '@deepseek-ai/dsh-llm'
@@ -288,11 +288,21 @@ export class ReactLoopAgent implements Agent {
     }
     }
   }
   }
 
 
-  /** Run one tool-call batch and drain its deferred context before resolving or rejecting. */
-  private async withToolBatch<T>(run: () => Promise<T>): Promise<T> {
+  /**
+   * Run one tool-call batch and drain its deferred context before settlement.
+   * The loop-owned acceptor remains valid after public disposal begins because
+   * the interrupted turn stays open until this batch settles.
+   */
+  private async withToolBatch<T>(
+    run: (acceptContext: (context: HookContext) => void) => Promise<T>,
+  ): Promise<T> {
     this.toolBatchActive = true
     this.toolBatchActive = true
+    const acceptContext = (context: HookContext): void => {
+      const accepted = this.acceptMessage(context.content, { source: context.source })
+      this.deferredInjections.push(accepted)
+    }
     try {
     try {
-      return await run()
+      return await run(acceptContext)
     } finally {
     } finally {
       this.toolBatchActive = false
       this.toolBatchActive = false
       this.drainDeferredInjections()
       this.drainDeferredInjections()

+ 6 - 11
packages/core/agent-loop/src/loop.ts

@@ -85,8 +85,8 @@ export interface LoopHandle {
   clearCancel(): void
   clearCancel(): void
   /** Settle idle waiters when pre-running cancellation skips a turn, without emitting `agent/status`. */
   /** Settle idle waiters when pre-running cancellation skips a turn, without emitting `agent/status`. */
   settleIdle(): void
   settleIdle(): void
-  /** Run an active tool-call batch and drain deferred context before resolving or rejecting. */
-  readonly withToolBatch: <T>(run: () => Promise<T>) => Promise<T>
+  /** Run an active tool-call batch, accepting post-tool context into the FIFO drained before settlement. */
+  readonly withToolBatch: <T>(run: (acceptContext: (context: HookContext) => void) => Promise<T>) => Promise<T>
 }
 }
 
 
 /**
 /**
@@ -559,9 +559,7 @@ async function runStep(
   // Tool execution stays sequential; recheck abort around each normalized result.
   // Tool execution stays sequential; recheck abort around each normalized result.
   const toolCalls = message.content.filter(block => block.type === 'tool-call')
   const toolCalls = message.content.filter(block => block.type === 'tool-call')
   if (toolCalls.length === 0) return { hadToolCalls: false, finish: assembler.finish }
   if (toolCalls.length === 0) return { hadToolCalls: false, finish: assembler.finish }
-  return handle.withToolBatch(async () => {
-    // Buffer context until all results are appended to preserve call/result adjacency.
-    const pendingContext: HookContext[] = []
+  return handle.withToolBatch(async (acceptContext) => {
     for (const call of toolCalls) {
     for (const call of toolCalls) {
       /* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
       /* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
       if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
       if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
@@ -591,7 +589,9 @@ async function runStep(
         // Persist tool-owned presentation data for replay.
         // Persist tool-owned presentation data for replay.
         ...result.meta !== undefined ? { meta: result.meta } : {},
         ...result.meta !== undefined ? { meta: result.meta } : {},
       }, { surfaceOp: 'append', sourceEventSeqs: [callEvent.seq] })
       }, { surfaceOp: 'append', sourceEventSeqs: [callEvent.seq] })
-      if (result.additionalContext) pendingContext.push(result.additionalContext)
+      // Accept into the batch FIFO immediately; it remains deferred until every
+      // result settles and survives abort, cancellation, or disposal afterward.
+      if (result.additionalContext) acceptContext(result.additionalContext)
       // The signal may flip while the tool is awaited.
       // The signal may flip while the tool is awaited.
       /* v8 ignore start -- signal.reason default unreachable: cancel()/disposal always set it */
       /* v8 ignore start -- signal.reason default unreachable: cancel()/disposal always set it */
       // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
       // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
@@ -599,11 +599,6 @@ async function runStep(
       /* v8 ignore stop */
       /* v8 ignore stop */
     }
     }
 
 
-    // Append buffered context after the complete result batch.
-    for (const context of pendingContext) {
-      agent.inject(context.content, { source: context.source })
-    }
-
     return { hadToolCalls: true, finish: assembler.finish }
     return { hadToolCalls: true, finish: assembler.finish }
   })
   })
 }
 }

+ 80 - 5
packages/core/agent-loop/tests/contract-regressions.spec.ts

@@ -3,7 +3,7 @@ import { Context } from 'cordis'
 import LlmService, { CallId, ContentBlock, MessageSource, StreamChunk } from '@deepseek-ai/dsh-llm'
 import LlmService, { CallId, ContentBlock, MessageSource, StreamChunk } from '@deepseek-ai/dsh-llm'
 import SessionStore, { Session, SessionEvent, SessionId, TurnEndReason } from '@deepseek-ai/dsh-session'
 import SessionStore, { Session, SessionEvent, SessionId, TurnEndReason } from '@deepseek-ai/dsh-session'
 import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
 import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
-import ToolRegistry, { defineTool } from '@deepseek-ai/dsh-tools'
+import ToolRegistry, { defineTool, type PostToolDecision } from '@deepseek-ai/dsh-tools'
 import AgentRegistry, { AgentId, type ContinuationDecision } from '@deepseek-ai/dsh-agent'
 import AgentRegistry, { AgentId, type ContinuationDecision } from '@deepseek-ai/dsh-agent'
 import AgentLoop, { ReactLoopAgent } from '@deepseek-ai/dsh-agent-loop'
 import AgentLoop, { ReactLoopAgent } from '@deepseek-ai/dsh-agent-loop'
 import { prepareReactLoopAgent } from '../src/agent.ts'
 import { prepareReactLoopAgent } from '../src/agent.ts'
@@ -154,6 +154,13 @@ describe('abort during tool execution ends the turn', () => {
         return [{ type: 'text', text: 'done' }]
         return [{ type: 'text', text: 'done' }]
       },
       },
     }))
     }))
+    ctx.on('tools/post-execute', async (): Promise<PostToolDecision> => ({
+      kind: 'accept',
+      additionalContext: {
+        content: [{ type: 'text', text: 'accepted result context after abort' }],
+        source: { kind: 'plugin', plugin: 'test' },
+      },
+    }))
 
 
     send(agent, 'go')
     send(agent, 'go')
     await waitForIdle(ctx, agent)
     await waitForIdle(ctx, agent)
@@ -163,9 +170,65 @@ describe('abort during tool execution ends the turn', () => {
       .filter(event => event.type === 'tool/result' || event.type === 'context/message'
       .filter(event => event.type === 'tool/result' || event.type === 'context/message'
         || event.type === 'step/end' || event.type === 'turn/end')
         || event.type === 'step/end' || event.type === 'turn/end')
       .map(event => event.type))
       .map(event => event.type))
-      .toEqual(['tool/result', 'context/message', 'step/end', 'turn/end'])
+      .toEqual(['tool/result', 'context/message', 'context/message', 'step/end', 'turn/end'])
+    expect(events
+      .filter(event => event.type === 'context/message')
+      .map(event => event.data.content))
+      .toEqual([
+        [{ type: 'text', text: 'accepted before abort' }],
+        [{ type: 'text', text: 'accepted result context after abort' }],
+      ])
+  })
+
+  it('records post-tool context when a later call aborts the batch', async () => {
+    const adapter = new MockAdapter([[
+      { type: 'block-start', index: 0, blockType: 'tool-call' },
+      { type: 'block-end', index: 0, block: { type: 'tool-call', id: CallId('c1'), name: 'first', arguments: '{}' } },
+      { type: 'block-start', index: 1, blockType: 'tool-call' },
+      { type: 'block-end', index: 1, block: { type: 'tool-call', id: CallId('c2'), name: 'aborter', arguments: '{}' } },
+      { type: 'finish', reason: { kind: 'tool-calls' } },
+    ] satisfies StreamChunk[]])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(AgentId('a-later-abort-context'), { model: 'mock' })
+    ctx.tools.register(defineTool({
+      name: 'first',
+      description: '',
+      parameters: {},
+      async execute() {
+        return [{ type: 'text', text: 'first done' }]
+      },
+    }))
+    ctx.tools.register(defineTool({
+      name: 'aborter',
+      description: '',
+      parameters: {},
+      async execute() {
+        ;(agent as unknown as { currentAbort?: AbortController }).currentAbort?.abort('user interrupt')
+        return [{ type: 'text', text: 'aborted' }]
+      },
+    }))
+    ctx.on('tools/post-execute', async (exec, _result, next): Promise<PostToolDecision> => {
+      if (exec.callId !== CallId('c1')) return next()
+      return {
+        kind: 'accept',
+        additionalContext: {
+          content: [{ type: 'text', text: 'accepted after first result' }],
+          source: { kind: 'plugin', plugin: 'test' },
+        },
+      }
+    })
+
+    send(agent, 'go')
+    await waitForIdle(ctx, agent)
+
+    const events = [...agent.session.events]
+    expect(events
+      .filter(event => event.type === 'tool/result' || event.type === 'context/message'
+        || event.type === 'step/end' || event.type === 'turn/end')
+      .map(event => event.type))
+      .toEqual(['tool/result', 'tool/result', 'context/message', 'step/end', 'turn/end'])
     expect(events.find(event => event.type === 'context/message')?.data.content)
     expect(events.find(event => event.type === 'context/message')?.data.content)
-      .toEqual([{ type: 'text', text: 'accepted before abort' }])
+      .toEqual([{ type: 'text', text: 'accepted after first result' }])
   })
   })
 
 
   it('drains deferred context before disposal reaches quiescence', async () => {
   it('drains deferred context before disposal reaches quiescence', async () => {
@@ -192,13 +255,25 @@ describe('abort during tool execution ends the turn', () => {
         return [{ type: 'text', text: 'done' }]
         return [{ type: 'text', text: 'done' }]
       },
       },
     }))
     }))
+    ctx.on('tools/post-execute', async (): Promise<PostToolDecision> => ({
+      kind: 'accept',
+      additionalContext: {
+        content: [{ type: 'text', text: 'accepted result context during disposal' }],
+        source: { kind: 'plugin', plugin: 'test' },
+      },
+    }))
 
 
     send(agent, 'go')
     send(agent, 'go')
     await started.promise
     await started.promise
     await fiber.dispose()
     await fiber.dispose()
 
 
-    expect(agent.session.events.find(event => event.type === 'context/message')?.data.content)
-      .toEqual([{ type: 'text', text: 'accepted before disposal' }])
+    expect(agent.session.events
+      .filter(event => event.type === 'context/message')
+      .map(event => event.data.content))
+      .toEqual([
+        [{ type: 'text', text: 'accepted before disposal' }],
+        [{ type: 'text', text: 'accepted result context during disposal' }],
+      ])
     expect(agent.session.events.find(event => event.type === 'turn/end')?.data.reason)
     expect(agent.session.events.find(event => event.type === 'turn/end')?.data.reason)
       .toEqual({ kind: 'disposed' })
       .toEqual({ kind: 'disposed' })
   })
   })

+ 3 - 3
packages/core/tools/src/index.ts

@@ -238,8 +238,8 @@ export interface ToolExecutionResult {
    */
    */
   error?: ToolErrorInfo
   error?: ToolErrorInfo
   /**
   /**
-   * Model-facing context for the next request, separate from this tool result.
-   * The loop buffers it until all step results are logged, preserving pairing.
+   * Model-facing context for the next request, separate from this tool result. The loop
+   * accepts it into the active-batch FIFO, then appends after recorded results even if interrupted.
    */
    */
   additionalContext?: HookContext
   additionalContext?: HookContext
   /**
   /**
@@ -843,7 +843,7 @@ export class ToolRegistry extends Service {
    * its {@link PostToolDecision}: `accept` keeps the call successful (replacing
    * its {@link PostToolDecision}: `accept` keeps the call successful (replacing
    * `content` when given), `block` turns it into an `isError` whose content is
    * `content` when given), `block` turns it into an `isError` whose content is
    * the corrective `feedback`. Either decision may attach `additionalContext`,
    * the corrective `feedback`. Either decision may attach `additionalContext`,
-   * which is ferried on the returned result for the loop's per-step buffer.
+   * which is ferried on the returned result for the loop's active-batch FIFO.
    * Runs inside `execute`'s outer try/catch (a throwing listener → isError).
    * Runs inside `execute`'s outer try/catch (a throwing listener → isError).
    */
    */
   private async postExecute(exec: ToolExecution, result: ToolExecutionResult): Promise<ToolExecutionResult> {
   private async postExecute(exec: ToolExecution, result: ToolExecutionResult): Promise<ToolExecutionResult> {

+ 2 - 2
scripts/gen-doc-graphs.ts

@@ -850,9 +850,9 @@ function renderToolPipeline(): string {
     `  owned["Tool-owned session events<br/>${mermaidCode('todo/write')}, ${mermaidCode('fs/observed')}, ${mermaidCode('hook/invoked')}, ${mermaidCode('hook/result')}, ${mermaidCode('tool/code-dispatch')}"]`,
     `  owned["Tool-owned session events<br/>${mermaidCode('todo/write')}, ${mermaidCode('fs/observed')}, ${mermaidCode('hook/invoked')}, ${mermaidCode('hook/result')}, ${mermaidCode('tool/code-dispatch')}"]`,
     `  post["${mermaidCode('tools/post-execute')} waterfall<br/>accept, block, replace, add context"]`,
     `  post["${mermaidCode('tools/post-execute')} waterfall<br/>accept, block, replace, add context"]`,
     `  final["${mermaidCode('tools/result')} synchronous notification<br/>frozen authoritative outcome"]`,
     `  final["${mermaidCode('tools/result')} synchronous notification<br/>frozen authoritative outcome"]`,
-    '  context["Buffered additionalContext<br/>context/message after all tool results"]',
+    '  context["Batch-deferred additionalContext<br/>context/message after recorded tool results"]',
     `  toolResult["Session event: ${mermaidCode('tool/result')}<br/>single model-facing outcome"]`,
     `  toolResult["Session event: ${mermaidCode('tool/result')}<br/>single model-facing outcome"]`,
-    '  allResults["All calls in the step settled<br/>and tool/result events recorded"]',
+    '  allResults["Tool batch settled<br/>recorded tool/result events complete"]',
     '  presentResult["UI completed card<br/>presentResult(args, result)"]',
     '  presentResult["UI completed card<br/>presentResult(args, result)"]',
     '  model --> toolCall',
     '  model --> toolCall',
     '  toolCall --> presentCall',
     '  toolCall --> presentCall',