Merge remote-tracking branch 'origin/master' into feat/tui-package

# Conflicts: # docs/architecture.md # docs/config-catalog.md # docs/cordis-catalog/events.md # docs/event-producer-consumer.md # docs/module-graph.md # examples/coding-agent/cordis.yml # examples/echo-agent/cordis.yml # knip.json # packages/core/agent-loop/README.md # packages/core/agent-loop/tests/config-session-id.spec.ts # packages/examples/README.md # packages/examples/stdio-demo/README.md # packages/examples/stdio-demo/tests/stdio-agent.spec.ts
2026-07-18 21:17:10 +08:00
parent 7f91a05ed7 417248eed8
commit 864b7bfd65
724 changed files with 40896 additions and 9028 deletions
--- a/packages/core/agent-loop/src/agent.ts
+++ b/packages/core/agent-loop/src/agent.ts
@@ -8,7 +8,7 @@

 import type { Context } from 'cordis'
 import { agentEvents } from '@deepseek-ai/dsh-agent'
-import type { AgentId, AgentOptions, AgentStatus, SendOptions } from '@deepseek-ai/dsh-agent'
+import type { AgentId, AgentOptions, AgentStatus, HookContext, InjectOptions, SendOptions } from '@deepseek-ai/dsh-agent'
 import type { Agent } from '@deepseek-ai/dsh-agent'
 import { deepFreeze } from '@deepseek-ai/dsh-llm'
 import type { ContentBlock, MessageSource } from '@deepseek-ai/dsh-llm'
@@ -54,15 +54,16 @@ export interface PreparedReactLoopAgent {
 * @param id - the concrete agent identity.
 * @param options - loop options for the agent.
 * @param session - the prepared session the agent will own.
+ * @param maxParallelToolCalls - resolved in-flight cap for this agent.
 * @returns the agent and closures bound only to that exact instance.
 */
 export function prepareReactLoopAgent(
-  ctx: Context, id: AgentId, options: AgentOptions, session: Session,
+  ctx: Context, id: AgentId, options: AgentOptions, session: Session, maxParallelToolCalls: number,
 ): PreparedReactLoopAgent {
  if (claimedDriverSessions.has(session)) {
    throw new Error(`session "${session.id}" already has a concrete agent driver`)
  }
-  const agent = new ReactLoopAgent(ctx, id, options, session)
+  const agent = new ReactLoopAgent(ctx, id, options, session, maxParallelToolCalls)
  claimedDriverSessions.add(session)
  const dispose = () => agent[stopDriver]()
  return {
@@ -143,19 +144,27 @@ export class ReactLoopAgent implements Agent {
   * the `disposed` transition fires and leave the promise hanging.
   */
  private idleWaiters: (() => void)[] = []
+  /** Maximum parallel-safe calls allowed in one step. */
+  private readonly maxParallelToolCalls: number
  /**
   * Durability checkpoints started by idle {@link inject} calls. `inject()` is
   * synchronous, so it cannot await them itself; the driver disposer drains
   * this set before the lifecycle unregisters the agent or detaches its session.
   */
  private pendingIdleFlushes = new Set<Promise<void>>()
+  /** Whether the current step is executing an assistant tool-call batch. */
+  private toolBatchActive = false
+  /** Open-turn injections waiting for the active assistant tool-call batch to close. */
+  private deferredInjections: HookContext[] = []

  constructor(
    private loopCtx: Context,
    public readonly id: AgentId,
    public readonly options: AgentOptions,
    public readonly session: Session,
+    maxParallelToolCalls: number,
  ) {
+    this.maxParallelToolCalls = maxParallelToolCalls
    const { promise, resolve } = Promise.withResolvers<void>()
    this.disposed = promise
    this.resolveDisposed = resolve
@@ -189,12 +198,11 @@ export class ReactLoopAgent implements Agent {
  }

  /**
-   * Accept one public send/steer payload as the exact detached record shared by
-   * the live notification and inbox. Lossless-JSON materialization reads every
-   * nested field once; deep freeze prevents an observer from rewriting queued
-   * work before the loop drains it.
+   * Accept one public message payload as a detached record. Lossless-JSON
+   * materialization reads every nested field once; deep freeze prevents later
+   * caller mutation before an inbox or deferred-injection queue drains it.
   */
-  private acceptInboxMessage(content: ContentBlock[], options?: SendOptions): InboxMessage {
+  private acceptMessage(content: ContentBlock[], options?: SendOptions): InboxMessage {
    const source = this.resolveSource(options)
    const accepted = snapshotJsonValue({ content, source })
    if (accepted === undefined) {
@@ -203,6 +211,15 @@ export class ReactLoopAgent implements Agent {
    return deepFreeze(accepted)
  }

+  /** Detach one context before it can outlive its caller in the active-batch FIFO. */
+  private acceptContext(context: HookContext): HookContext {
+    const accepted = snapshotJsonValue(context)
+    if (accepted === undefined) {
+      throw new TypeError('agent context must be losslessly JSON-serializable')
+    }
+    return deepFreeze(accepted)
+  }
+
  /** Reject a driving operation once teardown has synchronously closed the agent. */
  private assertNotDisposed(): void {
    if (this._status === 'disposed') throw new Error(`agent "${this.id}" is disposed`)
@@ -210,7 +227,7 @@ export class ReactLoopAgent implements Agent {

  send(content: ContentBlock[], options?: SendOptions): void {
    this.assertNotDisposed()
-    const accepted = this.acceptInboxMessage(content, options)
+    const accepted = this.acceptMessage(content, options)
    this.#inbox.enqueue(accepted)
    const info = { source: accepted.source, steering: false } as const
    agentEvents(this.loopCtx, this).emit('agent/queued', accepted.content, info)
@@ -219,20 +236,31 @@ export class ReactLoopAgent implements Agent {
  steer(content: ContentBlock[], options?: SendOptions): void {
    this.assertNotDisposed()
    if (this._status !== 'running') { this.send(content, options); return }
-    const accepted = this.acceptInboxMessage(content, options)
+    const accepted = this.acceptMessage(content, options)
    this.#inbox.steer(accepted)
    const info = { source: accepted.source, steering: true } as const
    agentEvents(this.loopCtx, this).emit('agent/queued', accepted.content, info)
  }

-  inject(content: ContentBlock[], options?: SendOptions): void {
+  inject(content: ContentBlock[], options?: InjectOptions): void {
    this.assertNotDisposed()
    const source = this.resolveSource(options)
+    const context = {
+      content,
+      source,
+      ...options?.envelope !== undefined ? { envelope: options.envelope } : {},
+      ...options?.meta !== undefined ? { meta: options.meta } : {},
+    }
    if (isTurnOpen(this.session)) {
-      // A turn is open in the LOG (decided from the log, not agent status —
-      // status can be `running` with no turn open): the context/message is
-      // turn-enclosed by that turn, so append it directly.
-      this.session.append('context/message', { content, source }, { surfaceOp: 'append' })
+      const accepted = this.acceptContext(context)
+      // Provider protocols require every assistant tool-call batch to be
+      // followed only by its tool results. Historical interrupted batches do
+      // not own new context; only the currently executing batch may defer it.
+      if (this.toolBatchActive) {
+        this.deferredInjections.push(accepted)
+        return
+      }
+      this.session.append('context/message', accepted, { surfaceOp: 'append' })
      return
    }
    // No turn open: wrap the injection in a one-shot turn so every event stays
@@ -244,7 +272,7 @@ export class ReactLoopAgent implements Agent {
    // are contained by Session and cannot create a false append failure.
    try {
      this.session.append('turn/start', { turn, trigger: { kind: 'injection', source } })
-      this.session.append('context/message', { content, source }, { surfaceOp: 'append' })
+      this.session.append('context/message', context, { surfaceOp: 'append' })
    } finally {
      // Close the turn if turn/start made it into the log. A pre-commit veto
      // must escape rather than being mistaken for a committed turn/end.
@@ -272,6 +300,34 @@ export class ReactLoopAgent implements Agent {
    }
  }

+  /** Append deferred open-turn injections after the loop closes a tool-result batch. */
+  private drainDeferredInjections(): void {
+    const pending = this.deferredInjections.splice(0)
+    for (const accepted of pending) {
+      this.session.append('context/message', accepted, { surfaceOp: 'append' })
+    }
+  }
+
+  /**
+   * 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
+    const acceptContext = (context: HookContext): void => {
+      this.deferredInjections.push(this.acceptContext(context))
+    }
+    try {
+      return await run(acceptContext)
+    } finally {
+      this.toolBatchActive = false
+      this.drainDeferredInjections()
+    }
+  }
+
  cancel(reason?: string): void {
    // Arm only for current work; an idle marker would cancel the next prompt.
    if (this._status === 'running' || this.currentAbort !== undefined || this.#inbox.hasQueued || this.#inbox.hasSteering) {
@@ -329,6 +385,7 @@ export class ReactLoopAgent implements Agent {
    this.driverStarted = true
    this.done = runLoop(this.loopCtx, this, {
      inbox: this.#inbox,
+      maxParallelToolCalls: this.maxParallelToolCalls,
      setStatus: (status) => { this.setStatus(status) },
      setAbort: controller => void (this.currentAbort = controller),
      disposed: this.disposed,
@@ -336,6 +393,7 @@ export class ReactLoopAgent implements Agent {
      isCancelled: () => this.cancelRequested,
      cancelReason: () => this.cancelReason,
      clearCancel: () => { this.cancelRequested = false },
+      withToolBatch: run => this.withToolBatch(run),
      // Pre-step cancellation re-parks without emitting a status transition.
      settleIdle: () => { this.settleIdleWaiters() },
    })
--- a/packages/core/agent-loop/src/constants.ts
+++ b/packages/core/agent-loop/src/constants.ts
@@ -0,0 +1,6 @@
+/** Shared agent-loop scheduler defaults.
+ * @module dsh-agent-loop/constants
+ */
+
+/** Default maximum in-flight parallel-safe calls per agent step. */
+export const DEFAULT_MAX_PARALLEL_TOOL_CALLS = 10
--- a/packages/core/agent-loop/src/index.ts
+++ b/packages/core/agent-loop/src/index.ts
@@ -32,6 +32,7 @@ import {
  ReactLoopAgent,
 } from './agent.ts'
 import type { PreparedReactLoopAgent } from './agent.ts'
+import { DEFAULT_MAX_PARALLEL_TOOL_CALLS } from './constants.ts'

 export { ReactLoopAgent } from './agent.ts'

@@ -73,6 +74,15 @@ function signalAbortError(id: AgentId, signal: AbortSignal): Error {
  return new Error(`agent "${id}" creation aborted`, { cause: signal.reason })
 }

+/** Resolve the deployment-wide scheduler cap at the owning config boundary. */
+function resolveMaxParallelToolCalls(value: number | undefined): number {
+  const maxParallelToolCalls = value ?? DEFAULT_MAX_PARALLEL_TOOL_CALLS
+  if (!Number.isInteger(maxParallelToolCalls) || maxParallelToolCalls < 1) {
+    throw new Error('maxParallelToolCalls must be a positive integer')
+  }
+  return maxParallelToolCalls
+}
+
 /**
 * Caller-owned create/resume transaction through rollback-covered publication
 * and quiescent teardown. Resources remain private until the final registry
@@ -163,13 +173,13 @@ class AgentCreationTransaction {
  }

  /** Construct the driver and scope, then install their complete ordered lifecycle. */
-  prepare(options: AgentOptions, session: Session): ReactLoopAgent {
+  prepare(options: AgentOptions, session: Session, maxParallelToolCalls: number): ReactLoopAgent {
    this.assertActive()
    const gate = Promise.withResolvers<void>()
    this.preparing = gate.promise
    try {
      this.session = session
-      const driver = prepareReactLoopAgent(this.loopCtx, this.id, options, session)
+      const driver = prepareReactLoopAgent(this.loopCtx, this.id, options, session, maxParallelToolCalls)
      this.driver = driver
      const agent = driver.agent
      const scope = createScope(this.loopCtx, agent)
@@ -318,8 +328,15 @@ declare module 'cordis' {
  }
 }

-/** Plugin configuration for declarative startup agents. */
+export { DEFAULT_MAX_PARALLEL_TOOL_CALLS }
+
+/** Agent-loop plugin configuration. */
 export interface Config {
+  /**
+   * Maximum parallel-safe calls in flight per agent step. `1` is serial;
+   * omission defaults to {@link DEFAULT_MAX_PARALLEL_TOOL_CALLS}.
+   */
+  maxParallelToolCalls?: number
  /** Agents created or resumed at plugin startup. */
  agents: (AgentOptions & {
    /** Registry identity for the live agent. */
@@ -337,8 +354,10 @@ export class AgentLoop extends Service implements AgentFactory {

  /** Runtime schema for declarative agents. */
  static Config = z.object({
+    maxParallelToolCalls: z.number().step(1).min(1).default(DEFAULT_MAX_PARALLEL_TOOL_CALLS),
    agents: z.array(z.object({
      id: z.string().required(),
+      provider: z.string(),
      model: z.string(),
      cwd: z.string(),
      resumeSessionId: z.string(),
@@ -346,15 +365,19 @@ export class AgentLoop extends Service implements AgentFactory {
  }) as unknown as z<Config>

  private readonly ownership: FactoryOwnership
+  /** Resolved concurrency cap for every driver created by this factory. */
+  private readonly maxParallelToolCalls: number
  /** Plain holder prevents Cordis from re-tracing the factory's dependency context through a caller shadow. */
  private readonly runtime: { ctx: Context }

  constructor(ctx: Context, public config: Config) {
    super(ctx, 'agentLoop')
+    this.maxParallelToolCalls = resolveMaxParallelToolCalls(config.maxParallelToolCalls)
    this.ownership = new FactoryOwnership(ctx.fiber)
    this.runtime = { ctx }
    ctx.effect(() => () => this.ownership.dispose(), 'agentLoop.transactions()')
    ctx.effect(() => ctx.agents.setFactory(this), 'agentLoop.setFactory()')
+    ctx.systemPrompt.variable('provider', context => context.agent?.options.provider)
    ctx.systemPrompt.variable('model', context => context.agent?.options.model)
    ctx.systemPrompt.variable('cwd', context => context.agent?.session.header.cwd)

@@ -403,7 +426,7 @@ export class AgentLoop extends Service implements AgentFactory {
    try {
      const sessionId = SessionId(`${id}-session-${randomUUID()}`)
      const session = loopCtx.sessions.prepare(sessionId, { meta })
-      const agent = transaction.prepare(options, session)
+      const agent = transaction.prepare(options, session, this.maxParallelToolCalls)
      transaction.publish('startup')
      return agent
    } catch (error: unknown) {
@@ -421,6 +444,7 @@ export class AgentLoop extends Service implements AgentFactory {
   * @returns the published handle.
   */
  async createAgent(ownerCtx: Context, options: CreateAgentOptions): Promise<AgentHandle> {
+    const agentOptions = options.agentOptions ?? {}
    const transaction = new AgentCreationTransaction(
      this.runtime.ctx,
      ownerCtx,
@@ -433,7 +457,7 @@ export class AgentLoop extends Service implements AgentFactory {
        ...options.seed === undefined ? {} : { seed: options.seed },
        ...options.meta === undefined ? {} : { meta: options.meta },
      })
-      const agent = transaction.prepare(options.agentOptions ?? {}, session)
+      const agent = transaction.prepare(agentOptions, session, this.maxParallelToolCalls)
      await transaction.waitFor(options.setup?.(agent.ctx))
      transaction.assertActive()
      return transaction.publish('startup')
@@ -465,6 +489,7 @@ export class AgentLoop extends Service implements AgentFactory {
    persistence: SessionPersistence,
    options: ResumeAgentOptions,
  ): Promise<AgentHandle> {
+    const agentOptions = options.agentOptions ?? {}
    const transaction = new AgentCreationTransaction(
      this.runtime.ctx,
      ownerCtx,
@@ -484,7 +509,7 @@ export class AgentLoop extends Service implements AgentFactory {
          ...loaded.meta.seedLength === undefined ? {} : { seedLength: loaded.meta.seedLength },
        },
      })
-      const agent = transaction.prepare(options.agentOptions ?? {}, session)
+      const agent = transaction.prepare(agentOptions, session, this.maxParallelToolCalls)
      await transaction.waitFor(options.setup?.(agent.ctx))
      transaction.assertActive()
      return transaction.publish('resume')
--- a/packages/core/agent-loop/src/loop.ts
+++ b/packages/core/agent-loop/src/loop.ts
@@ -6,7 +6,8 @@
 */

 import type { Context } from 'cordis'
-import type { FinishReason, GenerateOptions, LlmCallConfig, Message } from '@deepseek-ai/dsh-llm'
+import type { ContentBlock, FinishReason, GenerateOptions, LlmCallConfig, Message } from '@deepseek-ai/dsh-llm'
+import { isDeepStrictEqual } from 'node:util'
 import { BlockAssembler, HarnessError, deepFreeze } from '@deepseek-ai/dsh-llm'
 import { agentEvents, assembleContextFor } from '@deepseek-ai/dsh-agent'
 import type { AgentEventDispatch, ContinuationDecision, HookContext, PromptDecision } from '@deepseek-ai/dsh-agent'
@@ -17,6 +18,7 @@ import type { TransmissionLog } from './request-log.ts'
 import { renderPrompt } from '@deepseek-ai/dsh-system-prompt'
 import type { PromptAssembly } from '@deepseek-ai/dsh-system-prompt'
 import type {} from '@deepseek-ai/dsh-tools'
+import { executeToolCalls } from './tool-calls.ts'
 import type { ReactLoopAgent } from './agent.ts'
 import type { Inbox } from './inbox.ts'

@@ -72,6 +74,8 @@ function stepFinishReason(finish: FinishReason): TurnEndReason | undefined {
 export interface LoopHandle {
  /** Native-private agent inbox handed to the driver only at internal startup. */
  readonly inbox: Inbox
+  /** Maximum parallel-safe calls allowed in one step. */
+  readonly maxParallelToolCalls: number
  setStatus(status: 'idle' | 'running'): void
  setAbort(controller: AbortController | undefined): void
  /** Resolves when the agent is disposed — unblocks the idle wait. */
@@ -85,6 +89,8 @@ export interface LoopHandle {
  clearCancel(): void
  /** Settle idle waiters when pre-running cancellation skips a turn, without emitting `agent/status`. */
  settleIdle(): void
+  /** 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>
 }

 /**
@@ -234,10 +240,15 @@ async function runTurn(
      // `allow.content` REPLACES the prompt bytes (a rewrite); absent keeps them.
      const content = decision.content ?? message.content
      session.append('user/message', { content, source: message.source }, { surfaceOp: 'append' })
-      // `allow.additionalContext` is a SEPARATE context/message the next request
-      // also sees. The turn is open, so inject() appends it into THIS turn.
-      if (decision.additionalContext) {
-        agent.inject(decision.additionalContext.content, { source: decision.additionalContext.source })
+      // Every `allow.additionalContexts` entry is a separate context/message the
+      // next request also sees. The turn is open, so inject() appends each one
+      // into THIS turn without flattening provenance, framing, or metadata.
+      for (const context of decision.additionalContexts ?? []) {
+        agent.inject(context.content, {
+          source: context.source,
+          ...context.envelope !== undefined ? { envelope: context.envelope } : {},
+          ...context.meta !== undefined ? { meta: context.meta } : {},
+        })
      }
    }

@@ -325,7 +336,7 @@ async function runTurn(
      let stepOutcome: { hadToolCalls: boolean; finish: FinishReason } | { error: Error }
      try {
        stepOutcome = await runStep(
-          ctx, events, agent, turn, step, assembly, fullSystemPrompt, boundaryMessages, transmission, abort.signal)
+          ctx, events, agent, handle, turn, step, assembly, fullSystemPrompt, boundaryMessages, transmission, abort.signal)
      } catch (error: unknown) {
        stepOutcome = { error: toError(error) }
      } finally {
@@ -462,6 +473,7 @@ async function runStep(
  ctx: Context,
  events: AgentEventDispatch,
  agent: ReactLoopAgent,
+  handle: LoopHandle,
  turn: number,
  step: number,
  assembly: PromptAssembly,
@@ -477,12 +489,12 @@ async function runStep(
  const seedConfig: LlmCallConfig = deepFreeze(structuredClone(transmission.loggedHeader
    // eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- loggedHeader ⟹ a snapshot is in the log
    ? session.requestHeader()!.config
-    : { model: options.model ?? '' }))
+    : { provider: options.provider ?? '', model: options.model ?? '' }))

  // Listener replacements are recorded in the request header before dispatch.
  const config = await events.waterfall('agent/request', turn, step, seedConfig, () => Promise.resolve(seedConfig))
-  if (!config.model) {
-    throw new Error(`agent "${agent.id}" has no model: set AgentOptions.model or supply one via the agent/request waterfall`)
+  if (!config.provider || !config.model) {
+    throw new Error(`agent "${agent.id}" has no provider/model: set AgentOptions.provider and AgentOptions.model or supply both via the agent/request waterfall`)
  }

  // eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- runTurn composes the prefix before every runStep call
@@ -499,6 +511,7 @@ async function runStep(

  // Freeze the logged header plus boundary snapshot; the prefix precedes derived history.
  const request: GenerateOptions = deepFreeze({
+    provider: header.config.provider,
    model: header.config.model,
    messages: [...header.messagePrefix ?? [], ...boundaryMessages],
    ...header.system !== undefined ? { system: header.system } : {},
@@ -526,81 +539,108 @@ async function runStep(
  if (stepError) throw stepError

  if (assembler.finish.kind === 'max-tokens') {
-    let message: Message = withoutToolCalls(assembler.message())
-    message = withoutToolCalls(await events.waterfall('agent/step-result', turn, step, message, () => Promise.resolve(message)))
+    const assembled = assembler.message()
+    const assembledContent = structuredClone(assembled.content)
+    let message: Message = withoutToolCalls(assembled)
+    message = withoutToolCalls(await processStepResult(
+      events, session, turn, step, header.config, assembledContent, message, assembler, chunkSeqs,
+    ))
    // Preserve usage even when max-token truncation produced no content.
-    if (message.content.length > 0 || assembler.usage) {
-      // The finish chunk guarantees non-empty provenance here.
-      session.append(
-        'assistant/message',
-        { turn, step, content: message.content, ...(assembler.usage ? { usage: assembler.usage } : {}) },
-        { surfaceOp: 'append', sourceEventSeqs: chunkSeqs },
-      )
-    }
+    recordAssistantMessage(session, turn, step, header.config, assembledContent, message, assembler, chunkSeqs)
    return { hadToolCalls: false, finish: assembler.finish }
  }

  // Record the post-waterfall message that tool dispatch uses.
-  let message: Message = assembler.message()
-  message = await events.waterfall('agent/step-result', turn, step, message, () => Promise.resolve(message))
+  const assembled = assembler.message()
+  const assembledContent = structuredClone(assembled.content)
+  let message: Message = assembled
+  message = await processStepResult(
+    events, session, turn, step, header.config, assembledContent, message, assembler, chunkSeqs,
+  )

-  // Empty messages exist only to carry usage; omit empty provenance.
-  if (message.content.length > 0 || assembler.usage) {
-    session.append(
-      'assistant/message',
-      { turn, step, content: message.content, ...(assembler.usage ? { usage: assembler.usage } : {}) },
-      { surfaceOp: 'append', ...(chunkSeqs.length > 0 ? { sourceEventSeqs: chunkSeqs } : {}) },
-    )
-  }
+  // Every successful call records its completion anchor, including explicit
+  // empty chunk provenance for a contentless, usage-less provider response.
+  recordAssistantMessage(session, turn, step, header.config, assembledContent, message, assembler, chunkSeqs)

-  // Tool execution stays sequential; recheck abort around each normalized result.
+  // Dispatch may overlap; policy, durable results, and result context stay model-ordered.
  const toolCalls = message.content.filter(block => block.type === 'tool-call')
-  // Buffer context until all results are appended to preserve call/result adjacency.
-  const pendingContext: HookContext[] = []
-  for (const call of toolCalls) {
-    /* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
-    if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
-    const callEvent = session.append('tool/call', { turn, step, callId: call.id, name: call.name, arguments: call.arguments })
-    let parsedArguments: unknown
-    try {
-      parsedArguments = call.arguments ? JSON.parse(call.arguments) : {}
-    } catch {
-      parsedArguments = call.arguments
-    }
-    // TODO(pre-tool-input-rewrite): Keep logged history and live presentation aligned;
-    // see docs/rfc/proposed/feature/2026-06-30-pre-tool-input-rewrite.md.
-    const result = await ctx.tools.execute({
-      callId: call.id,
-      name: call.name,
-      arguments: parsedArguments,
-      agent,
-      signal,
-    })
-    session.append('tool/result', {
-      turn, step,
-      // Correlation comes from the immutable execution input; the result does
-      // not duplicate this authoritative transcript identity.
-      callId: call.id,
-      content: result.content,
-      isError: result.isError,
-      ...result.error ? { error: result.error } : {},
-      // Persist tool-owned presentation data for replay.
-      ...result.meta !== undefined ? { meta: result.meta } : {},
-    }, { surfaceOp: 'append', sourceEventSeqs: [callEvent.seq] })
-    if (result.additionalContext) pendingContext.push(result.additionalContext)
-    // The signal may flip while the tool is awaited.
-    /* v8 ignore start -- signal.reason default unreachable: cancel()/disposal always set it */
-    // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
-    if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
-    /* v8 ignore stop */
-  }
+  if (toolCalls.length === 0) return { hadToolCalls: false, finish: assembler.finish }
+  return handle.withToolBatch(async (acceptContext) => {
+    await executeToolCalls(
+      ctx, agent, turn, step, toolCalls, signal, handle.maxParallelToolCalls, acceptContext,
+    )
+    return { hadToolCalls: true, finish: assembler.finish }
+  })
+}

-  // Append buffered context after the complete result batch.
-  for (const context of pendingContext) {
-    agent.inject(context.content, { source: context.source })
+/** Preserve successful-call accounting without retaining output that result processing rejected. */
+async function processStepResult(
+  events: AgentEventDispatch,
+  session: Session,
+  turn: number,
+  step: number,
+  config: LlmCallConfig,
+  assembledContent: ContentBlock[],
+  message: Message,
+  assembler: BlockAssembler,
+  chunkSeqs: number[],
+): Promise<Message> {
+  try {
+    return await events.waterfall(
+      'agent/step-result', turn, step, message, () => Promise.resolve(message),
+    )
+  } catch (error: unknown) {
+    recordAssistantMessage(
+      session,
+      turn,
+      step,
+      config,
+      assembledContent,
+      { ...message, content: [] },
+      assembler,
+      chunkSeqs,
+      false,
+    )
+    throw error
  }
+}

-  return { hadToolCalls: toolCalls.length > 0, finish: assembler.finish }
+/** Record one content-or-usage assistant message with replay-safe provenance. */
+function recordAssistantMessage(
+  session: Session,
+  turn: number,
+  step: number,
+  config: LlmCallConfig,
+  assembledContent: ContentBlock[],
+  message: Message,
+  assembler: BlockAssembler,
+  chunkSeqs: number[],
+  preserveReplayState = true,
+): void {
+  session.append(
+    'assistant/message',
+    {
+      turn,
+      step,
+      content: message.content,
+      provenance: assistantProvenance(
+        config,
+        assembler.replayState,
+        preserveReplayState && isDeepStrictEqual(message.content, assembledContent),
+      ),
+      ...assembler.usage === undefined ? {} : { usage: assembler.usage },
+    },
+    { surfaceOp: 'append', sourceEventSeqs: chunkSeqs },
+  )
+}
+
+/** Build durable assistant provenance, dropping replay state after any content rewrite. */
+function assistantProvenance(config: LlmCallConfig, replayState: unknown, contentUnchanged: boolean): NonNullable<Message['provenance']> {
+  return {
+    provider: config.provider,
+    model: config.model,
+    ...contentUnchanged && replayState !== undefined ? { replayState } : {},
+  }
 }

 function withoutToolCalls(message: Message): Message {
--- a/packages/core/agent-loop/src/request-log.ts
+++ b/packages/core/agent-loop/src/request-log.ts
@@ -1,11 +1,12 @@
 /**
 * Per-loop-instance request-header bookkeeping for reconstructability. The
- * comparison baseline is the header folded from the session log, so a fresh
- * loop instance needs no special resume or fork state.
+ * comparison baseline is folded from the session log; a fresh instance anchors
+ * it with an initial/resume snapshot and later logs full changed snapshots.
+ *
 * @module dsh-agent-loop/request-log
 */

-import { diffHeader, headerEquals, applyHeaderDelta } from '@deepseek-ai/dsh-session'
+import { headerEquals } from '@deepseek-ai/dsh-session'
 import type { EpochHeader, Session } from '@deepseek-ai/dsh-session'
 import type { Message } from '@deepseek-ai/dsh-llm'

@@ -32,10 +33,8 @@ export function createTransmissionLog(): TransmissionLog {
 }

 /**
- * Append whatever header event makes the log reproduce this request's header.
- * The first request from an instance always records a full `initial` or `resume`
- * snapshot. Later requests record nothing when unchanged, a round-tripping
- * delta when expressible, or a full `fallback` snapshot otherwise.
+ * Append the full header snapshot owed by this request: initial/resume for the
+ * instance's first request, nothing when unchanged, or change otherwise.
 *
 * @param session - the session whose log explains the request.
 * @param state - this loop instance's bookkeeping (mutated on first log).
@@ -52,12 +51,5 @@ export function recordRequestHeader(session: Session, state: TransmissionLog, he
  // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
  const baseline = session.requestHeader()!
  if (headerEquals(baseline, header)) return
-  const delta = diffHeader(baseline, header)
-  /* v8 ignore next -- headerEquals false ⟹ diffHeader defined: both compare the same four parts */
-  if (delta === undefined) return
-  if (headerEquals(applyHeaderDelta(baseline, delta), header)) {
-    session.append('request/header-delta', delta)
-  } else {
-    session.append('request/header', { header, reason: 'fallback' })
-  }
+  session.append('request/header', { header, reason: 'change' })
 }
--- a/packages/core/agent-loop/src/tool-calls.ts
+++ b/packages/core/agent-loop/src/tool-calls.ts
@@ -0,0 +1,229 @@
+/**
+ * Schedules one assistant step's tool calls. Exclusive calls form barriers;
+ * parallel calls use a bounded rolling pool and are reclassified before start.
+ * Dispatch may overlap, while policy, results, and result context remain
+ * model-ordered. Abort stops replenishment and drains started calls.
+ *
+ * Each started call records `tool/call`; `tool/result` commits in model order,
+ * preserving derived history when audit events interleave with earlier results.
+ * @module dsh-agent-loop/tool-calls
+ */
+
+import type { Context } from 'cordis'
+import { assertNever, type ToolCallBlock } from '@deepseek-ai/dsh-llm'
+import type { HookContext } from '@deepseek-ai/dsh-agent'
+import type { Session } from '@deepseek-ai/dsh-session'
+import { TOOL_REGISTRY_SCHEDULER, type ToolExecutionInput, type ToolExecutionMode, type ToolExecutionResult, type ToolRunContext } from '@deepseek-ai/dsh-tools'
+import type { ReactLoopAgent } from './agent.ts'
+
+/** One tool call after argument parsing, ready to schedule. */
+interface PlannedCall {
+  block: ToolCallBlock
+  exec: ToolExecutionInput
+}
+
+/** Settled dispatch awaiting model-order finalization. */
+interface Slot {
+  exec: ToolRunContext
+  result: ToolExecutionResult
+  needsPost: boolean
+}
+
+/**
+ * Schedule one assistant step's tool calls by their live concurrency mode.
+ * Started calls receive ordered results. Abort drains them and rethrows after
+ * accepting their context into the batch FIFO owned by the caller.
+ *
+ * @param ctx - loop context that owns the tool registry.
+ * @param agent - agent and session receiving the call lifecycle.
+ * @param turn - current turn number.
+ * @param step - current step number.
+ * @param toolCalls - assistant calls in model order.
+ * @param signal - abort signal shared by the step.
+ * @param maxParallel - validated in-flight cap.
+ * @param acceptContext - accepts committed result context into the active batch.
+ */
+export async function executeToolCalls(
+  ctx: Context,
+  agent: ReactLoopAgent,
+  turn: number,
+  step: number,
+  toolCalls: ToolCallBlock[],
+  signal: AbortSignal,
+  maxParallel: number,
+  acceptContext: (context: HookContext) => void,
+): Promise<void> {
+  const { session } = agent
+
+  // Inputs are distinct because tools/execute wrappers may replace `exec.signal`.
+  const planned: PlannedCall[] = toolCalls.map(block => ({
+    block,
+    exec: {
+      callId: block.id,
+      name: block.name,
+      arguments: parseArguments(block.arguments),
+      agent,
+      signal,
+    },
+  }))
+
+  let next = 0
+  while (next < planned.length) {
+    // Commit before classifying again so registry changes affect unstarted calls.
+    // eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- bounded by the loop condition
+    const first = planned[next]!
+    const mode = ctx.tools.executionMode(first.exec).kind
+    const group = mode === 'parallel' ? planned.slice(next) : [first]
+    next += await runGroup(ctx, session, turn, step, group, mode, signal, maxParallel, acceptContext)
+  }
+}
+
+/** Parse model arguments, preserving invalid JSON as text and mapping empty input to `{}`. */
+function parseArguments(raw: string): unknown {
+  try {
+    return raw ? JSON.parse(raw) : {}
+  } catch {
+    return raw
+  }
+}
+
+/**
+ * Run one exclusive barrier or parallel pool. Later calls are reclassified
+ * before start; an exclusive reclassification waits for the current pool to
+ * drain and remains for the caller's next barrier. Results and contexts commit
+ * in model order. Abort stops starts, drains and commits started calls, accepts
+ * their contexts into the owning batch, and throws.
+ */
+async function runGroup(
+  ctx: Context,
+  session: Session,
+  turn: number,
+  step: number,
+  group: PlannedCall[],
+  mode: ToolExecutionMode['kind'],
+  signal: AbortSignal,
+  maxParallel: number,
+  acceptContext: (context: HookContext) => void,
+): Promise<number> {
+  /* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
+  if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
+  const slots: (Slot | undefined)[] = group.map(() => undefined)
+  // Started slots retain their tool/call seq for result provenance.
+  const callSeqs: number[] = group.map(() => -1)
+  let nextToStart = 0
+  let committed = 0
+  let started = 0
+  let aborted: boolean = signal.aborted
+
+  // `committed` advances only across contiguous model-order slots.
+  const commitReady = async (): Promise<void> => {
+    while (committed < group.length) {
+      const slot = slots[committed]
+      if (slot === undefined) break
+      const call = group[committed]
+      const result = slot.needsPost
+        ? await ctx.tools[TOOL_REGISTRY_SCHEDULER].finalize(slot.exec, slot.result)
+        : ctx.tools[TOOL_REGISTRY_SCHEDULER].finish(slot.exec, slot.result)
+      // eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- bounded index
+      appendToolResult(session, turn, step, call!.block, result, callSeqs[committed]!)
+      for (const context of result.additionalContexts ?? []) acceptContext(context)
+      committed++
+    }
+  }
+
+  const inFlight = new Map<number, Promise<number>>()
+
+  const startCall = async (index: number): Promise<void> => {
+    // eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- bounded index
+    const call = group[index]!
+    callSeqs[index] = appendToolCall(session, turn, step, call.block)
+    started++
+    const prepared = await ctx.tools[TOOL_REGISTRY_SCHEDULER].prepare(call.exec)
+    switch (prepared.kind) {
+      case 'dispatch': {
+        const promise = ctx.tools[TOOL_REGISTRY_SCHEDULER].dispatch(prepared.exec).then((outcome) => {
+          slots[index] = { exec: prepared.exec, result: outcome.result, needsPost: outcome.kind === 'post-result' }
+          return index
+        })
+        inFlight.set(index, promise)
+        break
+      }
+      case 'post-result':
+        slots[index] = { exec: prepared.exec, result: prepared.result, needsPost: true }
+        break
+      case 'final-result':
+        slots[index] = { exec: prepared.exec, result: prepared.result, needsPost: false }
+        break
+      /* v8 ignore next -- closed-union exhaustiveness guard */
+      default:
+        assertNever(prepared, 'tool-call scheduler prepare result')
+    }
+  }
+
+  const fillPool = async (): Promise<void> => {
+    while (!aborted && nextToStart < group.length && inFlight.size < maxParallel) {
+      // Re-read later modes after ordered commits so registry changes can create a barrier.
+      // eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- bounded by the loop condition
+      const nextCall = group[nextToStart]!
+      if (nextToStart > 0 && mode === 'parallel'
+        && ctx.tools.executionMode(nextCall.exec).kind !== 'parallel') break
+      await startCall(nextToStart)
+      nextToStart++
+      await commitReady()
+      // Abort may arrive while pre-execute awaits.
+      if (signal.aborted) aborted = true
+    }
+  }
+
+  // Ordered pre-execute may await; only dispatch/body overlaps.
+  // TODO: Drain every started call before rethrowing a scheduler error; tool
+  // bodies must not outlive the failed turn.
+  await fillPool()
+  while (inFlight.size > 0) {
+    const settledIndex = await Promise.race(inFlight.values())
+    inFlight.delete(settledIndex)
+    await commitReady()
+    // Abort may arrive while a tool or ordered commit awaits.
+    // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
+    if (signal.aborted) aborted = true
+    await fillPool()
+  }
+
+  if (aborted) {
+    // Started calls and accepted context settle before the turn records the abort.
+    /* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
+    throw new Error(String(signal.reason ?? 'aborted'))
+  }
+  /* v8 ignore next -- unreachable: a non-aborted group commits every started call */
+  if (committed !== started) throw new Error('tool-call scheduler: uncommitted settled calls')
+  return started
+}
+
+/** Append a started call and return its provenance sequence. */
+function appendToolCall(session: Session, turn: number, step: number, block: ToolCallBlock): number {
+  const event = session.append('tool/call', { turn, step, callId: block.id, name: block.name, arguments: block.arguments })
+  return event.seq
+}
+
+/** Append a model-ordered result linked to its call event. */
+function appendToolResult(
+  session: Session,
+  turn: number,
+  step: number,
+  block: ToolCallBlock,
+  result: ToolExecutionResult,
+  callSeq: number,
+): void {
+  session.append('tool/result', {
+    turn, step,
+    // Correlation stays with the loop's authoritative model-transcript call id;
+    // registry results deliberately do not duplicate it.
+    callId: block.id,
+    content: result.content,
+    isError: result.isError,
+    ...result.error ? { error: result.error } : {},
+    // The tool's private presentation payload (e.g. a result-time diff),
+    // persisted so a UI bridge reproduces the card on replay.
+    ...result.meta !== undefined ? { meta: result.meta } : {},
+  }, { surfaceOp: 'append', sourceEventSeqs: [callSeq] })
+}