turn/step：ReactLoopAgent 的驱动循环

1/**2 * Default Agent driver over queued turns and step-boundary input. Every request3 * is derived from the session log.4 * @module dsh-agent-loop/agent5 */

38type Phase =39  | { kind: 'idle'; lastTurn: number }40  | {41    kind: 'maintenance'42    abort: AbortController43    lastTurn: number44    wakeRequested: boolean45  }46  | { kind: 'running'; abort: AbortController; turn: number; step: number; wakeRequested: boolean }4748type StepEndReason = Extract<TurnEndReason, { kind: 'completed' | 'max-tokens' }>4950type PreparedStep =51  | { kind: 'reject' }52  | { kind: 'enter'; messages: UserMessage[]; assembly: PromptAssembly }

64export class ReactLoopAgent implements Agent {65  readonly inbox: Inbox66  private phase: Phase67  private activityDone: Promise<void> = Promise.resolve()6869  /** The agent-scoped registration boundary; the lifecycle owner unwinds it after the driver exits. */70  readonly scope: Scope71  readonly ctx: Context7273  /** Fused dispatcher, built once in the constructor so hot-path dispatches never allocate. */74  private readonly dispatch: AgentEventDispatch7576  /** Whether this loop instance has appended its initial/resume request anchor. */77  private requestHeaderLogged = false78  private readonly runtimeContext: RuntimeContextProjection7980  constructor(81    private loopCtx: Context,82    public readonly id: SessionId,83    public readonly options: AgentOptions,84    public readonly session: Session,85  ) {86    this.dispatch = agentEvents(loopCtx, this)87    this.inbox = new Inbox(session, {88      inserted: (message) => { this.dispatch.emit('agent/inbox/inserted', { message }) },89      discarded: (message) => { this.dispatch.emit('agent/inbox/discarded', { message }) },90      claimed: (message, turn) => { this.dispatch.emit('agent/inbox/claimed', { message, turn }) },91    })92    const lastTurn = session.events.findLast(event => event.type === 'turn/start')?.data.turn ?? 093    this.phase = { kind: 'idle', lastTurn }94    this.scope = createScope(loopCtx, this)95    this.ctx = this.scope.ctx.extend({ agent: this })96    this.runtimeContext = new RuntimeContextProjection(this.ctx, session)97  }

113  send(message: UserMessage, target: InboxTarget, wakeup: boolean): void {114    // Waking input cannot join an aborted activity, so it starts the next turn.115    // Captured before the insertion so a reentrant cancel from a splice observer cannot reclassify it.116    const wakingAfterAbort = wakeup && this.phase.kind !== 'idle' && this.phase.abort.signal.aborted117    const resolvedTarget = wakingAfterAbort ? 'next-turn' : target118    this.inbox.splice(resolvedTarget, Infinity, 0, [message])119    if (wakeup) this.wakeDriver(wakingAfterAbort)120  }

164  /**165   * Start one driver, or latch its wake behind maintenance or an aborted166   * activity. A wake sent while idle always opens its turn boundary, even167   * when its message was cleared; only a latched replay is suppressed when168   * the queue no longer holds the wake.169   * @param wakeAfterAbort - the {@link send} classification, captured before170   *   the inbox insertion so a reentrant cancel cannot reclassify it.171   */172  private wakeDriver(wakeAfterAbort = false): void {173    if (this.phase.kind !== 'idle') {174      // Maintenance and aborted drivers cannot deliver the wake: latch it for175      // replay at convergence. Live drivers claim queued work themselves;176      // disposal never latches, so teardown waits on no model turn.177      const reason = this.phase.abort.signal.reason as AgentCancelCause | undefined178      if (reason?.kind !== 'disposed' && (this.phase.kind === 'maintenance' || wakeAfterAbort)) {179        this.phase.wakeRequested = true180      }181      return182    }183    const driver = Promise.withResolvers<void>()184    this.activityDone = driver.promise185    this.setPhase({186      kind: 'running',187      abort: new AbortController(),188      turn: this.phase.lastTurn,189      step: 0,190      wakeRequested: false,191    })192    this.loopCtx.agents.withInitiator(this, () => this.kick()).then(driver.resolve, driver.reject)193  }

210  private async kick(): Promise<void> {211    try {212      while (await this.turn()) {}213    } catch (_error) {214      // Reported failures and cancellation are contained at the driver boundary.215    } finally {216      /* v8 ignore next -- kick owns a running phase until this driver boundary */217      if (this.phase.kind === 'running') {218        const { turn, wakeRequested } = this.phase219        this.setPhase({ kind: 'idle', lastTurn: turn })220        if (wakeRequested && this.inbox.hasPending) this.wakeDriver()221      }222    }223  }

245  /** Open one turn before claiming its first proposed step. */246  private async turn(): Promise<boolean> {247    if (this.phase.kind !== 'running') {248      this.throwError(new Error(`agent "${this.id}": turn without driver reservation`))249    }250    const phase = this.phase251    const { signal } = phase.abort252    signal.throwIfAborted()253    const turn = phase.turn + 1254    try {255      this.session.append('turn/start', { turn })256    } catch (error: unknown) {257      this.throwError(error)258    }259    phase.turn = turn260    let turnEnds: TurnEndReason | null = null261    let target: InboxTarget = 'next-turn'262    try {263      while (true) {264        signal.throwIfAborted()265        const step = phase.step + 1266        const decision = await this.preStep(target, { turn, step })267        if (decision.kind === 'reject') {268          turnEnds = { kind: 'blocked' }269          return false270        }271        if (turnEnds && decision.messages.length === 0) break272        // A removed waking message or an enter decision rewritten to empty273        // still owns the initial turn boundary, but it spends no model call.274        if (phase.step === 0 && decision.messages.length === 0) {275          turnEnds = { kind: 'completed' }276          return false277        }278        signal.throwIfAborted()279        this.session.append('step/start', { turn, step })

285          // max-tokens is sticky: once any step hits the ceiling, later steps286          // that complete normally must not downgrade the turn outcome.287          const stepEnd = await this.step(decision.assembly)288          // max-tokens stays sticky: a later completed step must not289          // downgrade the turn outcome.290          if (turnEnds === null || turnEnds.kind !== 'max-tokens') turnEnds = stepEnd291        } finally {292          this.session.append('step/end', { turn, step })293        }294        signal.throwIfAborted()295        if (turnEnds && this.inbox.nextStep.length === 0) {296          await this.dispatch.serial('agent/turn-stopping', { turn, signal })297          signal.throwIfAborted()298        }299        if (turnEnds && this.inbox.nextStep.length === 0) break300        target = 'next-step'

302    } catch (error: unknown) {303      if (signal.aborted) {304        turnEnds = { kind: 'aborted', reason: signal.reason as AgentCancelCause }305        throw error306      }307      // Every failure is structured: an `LlmError` keeps its facts, anything308      // else flattens to `errorChain` text under the `UNKNOWN` code.309      turnEnds = {310        kind: 'error',311        error: error instanceof LlmError312          ? error.failure313          : { message: errorChain(error), code: 'UNKNOWN' },314      }315      this.throwError(error)316    } finally {317      try {318        // oxlint-disable-next-line typescript/no-non-null-assertion -- every exit assigns a turn ending319        this.session.append('turn/end', { turn, reason: turnEnds! })320      } catch (error: unknown) {321        this.throwError(error)322      }323    }324    if (!this.inbox.hasPending) return false325    phase.abort = new AbortController()326    // A fresh controller makes a latch set on the old one stale: the live driver claims the queue itself.327    phase.wakeRequested = false328    phase.step = 0329    return true330  }

225  private async preStep(target: InboxTarget, position: { turn: number; step: number }): Promise<PreparedStep> {226    /* v8 ignore next -- private callers establish the running phase before proposing a step */227    if (this.phase.kind !== 'running') throw new Error(`agent "${this.id}": pre-step outside running phase`)228    const signal = this.phase.abort.signal229    const claimed = this.inbox.claim(target, position.turn)230    const assembly = await this.loopCtx.systemPrompt.assemble(assembleContextFor(this, signal))231    signal.throwIfAborted()232    const sections = renderContextSections(assembly)233    const context = this.runtimeContext.project(joinContextSections(sections), sections)234    const decision = await this.dispatch.waterfall(235      'agent/pre-step', { messages: claimed, ...position, signal },236      (): Promise<PreStepDecision> => Promise.resolve<PreStepDecision>({237        kind: 'enter',238        messages: context === undefined ? claimed : [...claimed, context],239      }),240    )241    signal.throwIfAborted()242    return decision.kind === 'reject' ? decision : { ...decision, assembly }243  }

63  /**64   * Remove and return the complete batch proposed for one step, publishing65   * each claimed message. The durable splices are pure deletions.66   * @param target - whether this boundary also consumes one queued turn.67   * @param turn - turn that will own the claimed batch.68   * @returns next-step input followed by the queued turn, when requested.69   * @internal - The agent loop's step-boundary operation, not a plugin extension point.70   */71  claim(target: InboxTarget, turn: number): UserMessage[] {72    const claimed = this.mutate('next-step', 0, this.nextStep.length, [], false)73    if (target === 'next-turn') {74      claimed.push(...this.mutate('next-turn', 0, 1, [], false))75    }76    for (const message of claimed) this.notifications.claimed(message, turn)77    return claimed78  }

128  /**129   * Apply standard splice semantics and durably record the normalized result.130   * The durable event commits before the live projection mutates, so synchronous131   * `session/event` observers see the pre-splice lists and can reconstruct the132   * removed messages from the normalized coordinates.133   * @param target - pending list to mutate.134   * @param start - splice position.135   * @param deleteCount - maximum number of messages to remove.136   * @param inserted - messages to insert at the resolved position.137   * @returns messages removed by the splice.138   */139  splice(140    target: InboxTarget,141    start: number,142    deleteCount: number,143    inserted: UserMessage[],144  ): UserMessage[] {145    return this.mutate(target, start, deleteCount, inserted, true)146  }

332  private async step(assembly: PromptAssembly): Promise<StepEndReason | null> {333    /* v8 ignore next -- private callers establish the running phase before executing a step */334    if (this.phase.kind !== 'running') throw new Error(`agent "${this.id}": step outside running phase`)335    const { turn, step, abort: { signal } } = this.phase336    signal.throwIfAborted()337    const system = renderPrompt(assembly)338339    while (true) {340      const { request, preparedCall } = await this.buildRequest(341        turn, step, assembly.tools, system, this.session.deriveMessages(), signal,342      )343      const assembler = new BlockAssembler()344      const chunkSeqs: number[] = []345      try {346        const stream = preparedCall?.stream(request) ?? this.loopCtx.llm.stream(request)347        signal.throwIfAborted()348        for await (const chunk of stream) {349          signal.throwIfAborted()350          chunkSeqs.push(this.session.append('assistant/chunk', { turn, step, chunk }).seq)351          assembler.push(chunk)352        }353        signal.throwIfAborted()354      } catch (error: unknown) {

372      const finish = assembler.finish373      if (finish.kind === 'error' || finish.kind === 'aborted') {374        const action = await this.dispatch.waterfall(375          'agent/request-error', {376            turn,377            step,378            provider: request.provider,379            failure: finish.failure,380            retryPolicy: preparedCall?.retryPolicy,381            signal,382          },383          () => Promise.resolve<RequestErrorAction>(undefined),384        )385        signal.throwIfAborted()386        if (action?.kind !== 'retry') {387          throw new LlmError(finish.failure.message, finish.failure.code, finish.failure)388        }389        continue390      }

410      if (finish.kind === 'max-tokens') return { kind: 'max-tokens' }411412      const toolCalls = message.content.filter(block => block.type === 'tool-call')413      if (toolCalls.length === 0) return { kind: 'completed' }414      const { concluded } = await executeToolCalls(415        this.loopCtx, turn, step, toolCalls, signal,416        context => this.inbox.splice('next-step', this.inbox.nextStep.length, 0, [context]),417      )418      return concluded ? { kind: 'completed' } : null

59export async function executeToolCalls(60  ctx: Context,61  turn: number,62  step: number,63  toolCalls: ToolCallBlock[],64  signal: AbortSignal,65  acceptContext: (context: UserMessage) => void,66): Promise<{ concluded: boolean }> {67  const agent = ctx.agents.requireInitiator()68  const { session } = agent6970  // Inputs are distinct because tools/execute wrappers may replace `exec.signal`.71  const planned: PlannedCall[] = toolCalls.map(block => ({72    block,73    exec: {74      callId: block.id,75      name: block.name,76      arguments: parseArguments(block.arguments),77      agent,78      signal,79    },80  }))8182  let next = 083  let concluded = false84  while (next < planned.length) {85    // Commit before classifying again so registry changes affect unstarted calls.86    // oxlint-disable-next-line typescript/no-non-null-assertion -- bounded by the loop condition87    const first = planned[next]!88    const mode = ctx.tools.executionMode(first.exec).kind89    const group = mode === 'parallel' ? planned.slice(next) : [first]90    const outcome = await runGroup(91      ctx, turn, step, group, mode, signal, acceptContext,92    )93    next += outcome.consumed94    concluded ||= outcome.concluded95    if (outcome.aborted) {96      for (const call of planned.slice(next)) appendSkippedToolCall(session, turn, step, call.block)97      return { concluded }98    }99  }100  return { concluded }101}

237  if (aborted) {238    // Started calls and accepted context settle first; every remaining model239    // call then receives an ordered synthetic result before the turn aborts.240    for (const call of group.slice(started)) appendSkippedToolCall(session, turn, step, call.block)241    return { consumed: group.length, aborted: true, concluded }242  }243  /* v8 ignore next -- unreachable: a non-aborted group commits every started call */244  if (committed !== started) throw new Error('tool-call scheduler: uncommitted settled calls')245  return { consumed: started, aborted: false, concluded }246}247248/** Append the durable call/result pair for a model call skipped after cancellation. */249function appendSkippedToolCall(session: Session, turn: number, step: number, block: ToolCallBlock): void {250  const callSeq = appendToolCall(session, turn, step, block)251  appendToolResult(session, turn, step, block, {252    content: [{ type: 'text', text: 'Error: tool call aborted before dispatch' }],253    isError: true,254    error: {255      message: 'tool call aborted before dispatch',256      info: { name: 'AbortError', code: TOOL_ABORTED_BEFORE_DISPATCH },257    },258  }, callSeq)259}

484    if (!this.requestHeaderLogged) {485      this.session.append('request/header', { header, reason: baseline === undefined ? 'initial' : 'resume' })486      this.requestHeaderLogged = true487    } else if (baseline === undefined || !headerEquals(baseline, header)) {488      this.session.append('request/header', { header, reason: 'change' })489    }

63## Turn flow6465A **step** is one model request plus the tools it calls. A **turn** is zero or more steps: it opens before its first input is claimed and closes once nothing is owed.

13Pure Cordis event taxonomy. The loop's extension points are typed events with deliberate dispatch modes:1415- **waterfall** (around-middleware) where plugins transform, short-circuit, recover, or wrap: `agent/pre-step`, `agent/request`, `agent/request-error`, `tools/pre-execute`, `tools/execute`, `tools/post-execute`, `llm/stream`, `system-prompt/assemble`.16- **serial** (awaited in listener order) for ordered checkpoints such as `agent/turn-stopping`.17- **parallel** (awaited fan-out) where every listener must get an independent chance: the `session/flush` durability checkpoint.18- **emit** (synchronous fire-and-forget) for notifications: inbox transitions, lifecycle, errors, and the contained immutable `tools/result` observation. Durable session events own turn and step boundaries.

217`agent/pre-step` receives one payload carrying the exclusive claimed batch (`messages`), the proposed step's coordinates (`turn`, `step`), and the current turn's cancellation `signal`. The initial proposal runs inside an open turn before any step; a tool continuation may submit an empty claimed batch between steps:218219It returns a `PreStepDecision`. Reject opens no step. Enter supplies the complete message batch appended after `step/start`; claimed messages omitted by the final decision remain removed, while input inserted after the claim stays pending:

108- **Model-visible ⟺ logged**: anything that reaches a model request must be reconstructable from the session log; a new model-visible input requires a session event.109- **Plugins, not loop changes**: new behavior goes on documented extension points; changing `agent-loop` requires updating docs/architecture.md.

187    /**188     * One message left the inbox inside its open turn. If the proposed step189     * is rejected, the claimed message ends here: it is neither discarded nor190     * re-emitted as a user/message, and the turn closes without a step.191     * @param payload.agent - the agent whose inbox changed.192     * @param payload.message - the claimed message.193     * @param payload.turn - the owning turn.194     * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent.195     * @mode emit196     */197    'agent/inbox/claimed'(this: Scoped<Agent>, payload: { agent: Agent; message: UserMessage; turn: number }): void

234  it('reject closes the claimed prompt turn without a step or model call', async () => {235    const adapter = new MockAdapter([textResponse('should not run')])236    const ctx = await harness(adapter)237    const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })238239    ctx.on('agent/pre-step', async (): Promise<PreStepDecision> => ({ kind: 'reject' }))240241    const reasons: TurnEndReason[] = []242    ctx.on('session/event', (_s, event: SessionEvent) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })243244    agent.followup(createUserMessage({ content: [{ type: 'text', text: 'do something' }], source: { kind: 'user' } }))245    await agent.whenIdle()246247    // the model was never called248    expect(adapter.requests).toHaveLength(0)249    const log = events(agent)250    expect(log.filter(e => e.type === 'turn/start' || e.type === 'turn/end').map(e => e.type))251      .toEqual(['turn/start', 'turn/end'])252    expect(log.some(e => e.type === 'user/message')).toBe(false)253    expect(log.some(e => e.type === 'step/start')).toBe(false)254    expect(reasons).toEqual([{ kind: 'blocked' }])255  })

318  it('preserves input staged after the blocked batch was claimed', async () => {319    const adapter = new MockAdapter([textResponse('retried')])320    const ctx = await harness(adapter)321    const agent = ctx.agentLoop.create(SessionId('blocked-pre-step-outbox'), { provider: 'mock', model: 'mock' })322    const entered = Promise.withResolvers<undefined>()323    const decision = Promise.withResolvers<PreStepDecision>()324    const disposeBlock = ctx.on('agent/pre-step', async () => {325      entered.resolve(undefined)326      return decision.promise327    })328329    const blockedIdle = waitForIdle(ctx, agent)330    send(agent, 'blocked prompt')331    await entered.promise332    agent.inject(createUserMessage({333      content: [{ type: 'text', text: 'staged context' }],334      source: { kind: 'plugin', plugin: 'test' },335    }))336    agent.steer(createUserMessage({ content: [{ type: 'text', text: 'staged steering' }], source: { kind: 'user' } }))337    decision.resolve({ kind: 'reject' })338    await blockedIdle339340    expect(agent.inbox.nextStep.map(message => message.content[0]))341      .toEqual([342        { type: 'text', text: 'staged context' },343        { type: 'text', text: 'staged steering' },344      ])345    expect(events(agent).filter(event => event.type === 'turn/start' || event.type === 'turn/end')346      .map(event => event.type)).toEqual(['turn/start', 'turn/end'])347    expect(adapter.requests).toEqual([])348349    disposeBlock()350    send(agent, 'resume')351    await waitForIdle(ctx, agent)352353    const staged = events(agent).filter(event =>354      event.type === 'user/message')355    expect(staged.map(event => event.type)).toEqual([356      'user/message',357      'user/message',358      'user/message',359    ])360    expect(JSON.stringify(adapter.requests[0]?.messages)).not.toContain('blocked prompt')361    expect(JSON.stringify(adapter.requests[0]?.messages)).toContain('staged context')362    expect(JSON.stringify(adapter.requests[0]?.messages)).toContain('staged steering')363  })

3This directory contains some official Claude Code plugins that extend functionality through custom commands, agents, and workflows. These are examples of what's possible with the Claude Code plugin system—many more plugins are available through community marketplaces.

249  /** The `agent(prompt, opts)` hook. */250  private async agent(rawPrompt: unknown, rawOpts: unknown): Promise<unknown> {251    this.throwIfCancelled()252    if (typeof rawPrompt !== 'string' || rawPrompt.length === 0) {253      throw new WorkflowError('agent() requires a non-empty prompt string', 'INVALID_ARGUMENT')254    }255    const opts = this.readAgentOptions(rawOpts)256    if (this.started >= this.limits.maxTotalAgents) {257      throw new WorkflowError(258        `this run reached its total agent cap (${this.limits.maxTotalAgents}) — a runaway-loop backstop; raise the applicable maxTotalAgents limit if the scale is intentional`,259        'AGENT_CAP',260      )261    }262    this.started += 1

74  async startAgent(request: ChildStartRequest): Promise<ChildHandle> {75    this.nextCallId += 176    const callId = this.nextCallId77    const entry: PendingChild = {78      started: Promise.withResolvers<string>(),79      settled: Promise.withResolvers<ChildResult>(),80      disposed: Promise.withResolvers<void>(),81    }82    // Containment: when asynchronous provider start fails (or83    // the run is torn down), the settled promise may never gain a consumer —84    // it must not surface as an unhandled rejection and kill the worker.85    entry.settled.promise.catch(() => { /* consumed: unconsumed child settlement after failed start */ })86    this.pending.set(callId, entry)87    this.post(WorkerToHostType.ChildStart, { callId, request })88    const childId = await entry.started.promise89    return new RpcChildHandle(this.post, callId, entry, childId)90  }