Merge branch 'codex/simp-agent-entry-state' into codex/simp-unify-agent-session-id

# Conflicts:
#	docs/config-catalog.md
#	docs/cordis-catalog/services.md
#	packages/core/agent-loop/README.md
#	packages/core/agent-loop/src/agent.ts
#	packages/core/agent-loop/src/index.ts
#	packages/core/agent-loop/tests/agent.spec.ts
#	packages/core/agent-loop/tests/contract-regressions.spec.ts
#	website/zh-CN/api/harness/agent-loop.md

packages/core/agent-loop/src/agent.ts

@@ -54,15 +54,20 @@ 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: SessionId, options: AgentOptions, session: Session,
+  ctx: Context,
+  id: SessionId,
+  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,6 +148,8 @@ 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
@@ -159,7 +166,9 @@ export class ReactLoopAgent implements Agent {
     public readonly id: SessionId,
     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
@@ -380,6 +389,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,

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

packages/core/agent-loop/src/index.ts

@@ -31,6 +31,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'
 
@@ -97,6 +98,15 @@ function signalAbortError(id: SessionId, 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
@@ -187,13 +197,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)
@@ -354,8 +364,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 & {
     /** Stable config label used in logs and as the fresh combined-id prefix. */
@@ -393,6 +410,7 @@ 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(),
       sessionId: z.string().min(1),
@@ -404,12 +422,15 @@ 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')
     validateConfiguredAgents(config.agents)
+    this.maxParallelToolCalls = resolveMaxParallelToolCalls(config.maxParallelToolCalls)
     this.ownership = new FactoryOwnership(ctx.fiber)
     this.runtime = { ctx }
     ctx.effect(() => () => this.ownership.dispose(), 'agentLoop.transactions()')
@@ -524,7 +545,7 @@ export class AgentLoop extends Service implements AgentFactory {
     const transaction = new AgentCreationTransaction(loopCtx, this.ctx, this.ownership, id)
     try {
       const session = loopCtx.sessions.prepare(id, { meta })
-      const agent = transaction.prepare(options, session)
+      const agent = transaction.prepare(options, session, this.maxParallelToolCalls)
       transaction.publish('startup')
       return agent
     } catch (error: unknown) {
@@ -542,6 +563,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,
@@ -554,7 +576,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')
@@ -586,6 +608,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,
@@ -605,7 +628,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')

packages/core/agent-loop/src/loop.ts

@@ -18,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'
 
@@ -73,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. */
@@ -559,49 +562,13 @@ async function runStep(
   // 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')
   if (toolCalls.length === 0) return { hadToolCalls: false, finish: assembler.finish }
   return handle.withToolBatch(async (acceptContext) => {
-    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] })
-      // Accept into the batch FIFO immediately; entries remain deferred until
-      // every recorded result settles and survive abort or disposal afterward.
-      for (const context of result.additionalContexts ?? []) acceptContext(context)
-      // 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 */
-    }
+    await executeToolCalls(
+      ctx, agent, turn, step, toolCalls, signal, handle.maxParallelToolCalls, acceptContext,
+    )
     return { hadToolCalls: true, finish: assembler.finish }
   })
 }

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] })
+}