fix(agent): route loop dispatches through prebuilt fused dispatcher

Address review feedback on PR #1738: - ReactLoopAgent builds its AgentEventDispatch once in the constructor and routes every emit/serial/waterfall through it, so hot-path dispatches no longer allocate a carrier and dispatcher per call; the public carrier field is gone (fused dispatcher is private). - agentEvents accepts an optional prebuilt carrier. - The fused payload builder spreads the payload before the injected agent so a structurally acceptable payload carrying an agent field can never override the subject. - Regenerate doc graphs; re-record core + architecture + affected Agent Note translation pairs; add payload-object event contract Agent Note.
2026-08-06 13:44:23 +08:00
parent 23671848c3
commit 262d044642
23 changed files with 143 additions and 61 deletions
--- a/packages/core/agent-loop/src/agent.ts
+++ b/packages/core/agent-loop/src/agent.ts
@@ -7,6 +7,7 @@
 import type {
  Agent,
  AgentCancelCause,
+  AgentEventDispatch,
  AgentOptions,
  AgentStatus,
  CancelOptions,
@@ -14,7 +15,7 @@ import type {
  PreStepDecision,
  RequestErrorAction,
 } from '@deepseek-ai/dsh-agent'
-import { Inbox, agentCarrier, assembleContextFor, emitAgentEvent } from '@deepseek-ai/dsh-agent'
+import { Inbox, agentEvents, assembleContextFor } from '@deepseek-ai/dsh-agent'
 import type { GenerateOptions, LlmCallConfig, Message, PreparedLlmCall } from '@deepseek-ai/dsh-llm'
 import {
  BlockAssembler,
@@ -24,7 +25,7 @@ import {
  errorChain,
  markAgentLoopRequest,
 } from '@deepseek-ai/dsh-llm'
-import type { Scope, Scoped } from '@deepseek-ai/dsh-scope'
+import type { Scope } from '@deepseek-ai/dsh-scope'
 import { createScope } from '@deepseek-ai/dsh-scope'
 import type { EpochHeader, RequestContext, Session, SessionId, TurnEndReason, UserMessage } from '@deepseek-ai/dsh-session'
 import { canonicalHeader, headerEquals } from '@deepseek-ai/dsh-session'
@@ -69,8 +70,8 @@ export class ReactLoopAgent implements Agent {
  readonly scope: Scope
  readonly ctx: Context

-  /** Fused scope carrier, built once in the constructor for every dispatch. */
-  readonly carrier: Scoped<Agent>
+  /** Fused dispatcher, built once in the constructor so hot-path dispatches never allocate. */
+  private readonly dispatch: AgentEventDispatch

  /** Whether this loop instance has appended its initial/resume request anchor. */
  private requestHeaderLogged = false
@@ -82,11 +83,11 @@ export class ReactLoopAgent implements Agent {
    public readonly options: AgentOptions,
    public readonly session: Session,
  ) {
-    this.carrier = agentCarrier(this)
+    this.dispatch = agentEvents(loopCtx, this)
    this.inbox = new Inbox(session, {
-      inserted: (message) => { emitAgentEvent(loopCtx, this, 'agent/inbox/inserted', { message }) },
-      discarded: (message) => { emitAgentEvent(loopCtx, this, 'agent/inbox/discarded', { message }) },
-      claimed: (message, turn) => { emitAgentEvent(loopCtx, this, 'agent/inbox/claimed', { message, turn }) },
+      inserted: (message) => { this.dispatch.emit('agent/inbox/inserted', { message }) },
+      discarded: (message) => { this.dispatch.emit('agent/inbox/discarded', { message }) },
+      claimed: (message, turn) => { this.dispatch.emit('agent/inbox/claimed', { message, turn }) },
    })
    const lastTurn = session.events.findLast(event => event.type === 'turn/start')?.data.turn ?? 0
    this.phase = { kind: 'idle', lastTurn }
@@ -105,7 +106,7 @@ export class ReactLoopAgent implements Agent {
    this.phase = next
    const status = this.status
    if (status !== previousStatus) {
-      emitAgentEvent(this.loopCtx, this, 'agent/status', { status })
+      this.dispatch.emit('agent/status', { status })
    }
  }

@@ -183,7 +184,7 @@ export class ReactLoopAgent implements Agent {
  private throwError(error: unknown): never {
    const turn = this.phase.kind === 'running' ? this.phase.turn : this.phase.lastTurn
    const step = this.phase.kind === 'running' ? this.phase.step : 0
-    emitAgentEvent(this.loopCtx, this, 'agent/error', { turn, step, error })
+    this.dispatch.emit('agent/error', { turn, step, error })
    throw error
  }

@@ -209,8 +210,8 @@ export class ReactLoopAgent implements Agent {
    signal.throwIfAborted()
    const sections = renderContextSections(assembly)
    const context = this.runtimeContext.project(joinContextSections(sections), sections)
-    const decision = await this.loopCtx.waterfall(
-      this.carrier, 'agent/pre-step', { agent: this, messages: claimed, ...position, signal },
+    const decision = await this.dispatch.waterfall(
+      'agent/pre-step', { messages: claimed, ...position, signal },
      (): Promise<PreStepDecision> => Promise.resolve<PreStepDecision>({
        kind: 'enter',
        messages: context === undefined ? claimed : [...claimed, context],
@@ -271,7 +272,7 @@ export class ReactLoopAgent implements Agent {
        }
        signal.throwIfAborted()
        if (turnEnds && this.inbox.nextStep.length === 0) {
-          await this.loopCtx.serial(this.carrier, 'agent/turn-stopping', { agent: this, turn, signal })
+          await this.dispatch.serial('agent/turn-stopping', { turn, signal })
          signal.throwIfAborted()
        }
        if (turnEnds && this.inbox.nextStep.length === 0) break
@@ -328,9 +329,8 @@ export class ReactLoopAgent implements Agent {
      signal.throwIfAborted()
      const finish = assembler.finish
      if (finish.kind === 'error' || finish.kind === 'aborted') {
-        const action = await this.loopCtx.waterfall(
-          this.carrier, 'agent/request-error', {
-            agent: this,
+        const action = await this.dispatch.waterfall(
+          'agent/request-error', {
            turn,
            step,
            provider: request.provider,
@@ -412,8 +412,8 @@ export class ReactLoopAgent implements Agent {
          ...maxTokens === undefined ? {} : { maxTokens },
        },
    ))
-    const proposedConfig = await this.loopCtx.waterfall(
-      this.carrier, 'agent/request', { agent: this, turn, step, signal },
+    const proposedConfig = await this.dispatch.waterfall(
+      'agent/request', { turn, step, signal },
      () => Promise.resolve(seedConfig),
    )
    signal.throwIfAborted()
--- a/packages/core/agent/README.i18n.yaml
+++ b/packages/core/agent/README.i18n.yaml
@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write packages/core/agent/README.md
-README.md: c3d6e6c24480894b6059417c1ab89db7aa0d7fa2
-README.zh.md: 16ee8f5e6c483555839b0c3ab174e2e2356b1359
+README.md: 2a69ab380eaad3929e27039582807037969eba64
+README.zh.md: 176f3f75cf0f6e3309b2f5d34afb4d562105608e
--- a/packages/core/agent/README.md
+++ b/packages/core/agent/README.md
@@ -50,7 +50,7 @@ Agent *creation* is provided by the plugin implementing `AgentFactory` (`dsh-age

 The lifecycle edges have two important local caveats. `agent/created` runs after scoped setup and after both session and agent registry entries exist. Setup is trusted composition-only code; the immediately following non-vetoing `agent/session-start` notification is the first supported startup injection point. `agent/disposed` always means the exact agent has left the registry. AgentLoop emits it after its driver is quiescent, while ordered teardown may still be detaching the session and unwinding the scope; custom agents registered directly own any stronger driver-ordering contract themselves.

-Most interception points are cooperative waterfalls. `agent/pre-step` receives the exclusive claimed `UserMessage[]` plus a `PreStepContext` containing the proposed `turn`, `step`, and cancellation `signal`; its batch may be empty when tools already require another request. Other turn-scoped asynchronous seams receive their explicit `AbortSignal` positionally. Listeners may cooperate with a signal but must not retain it as authority over another turn. `agent/request-error` is the failed-model-request recovery waterfall: it receives request coordinates, normalized failure facts, the serving registration's retry policy when available, and the signal. A listener returns `{ kind: 'retry' }` without calling `next()` when it owns recovery. `agent/turn-stopping` runs before an otherwise completed turn closes. The [explicit-cancellation decision](../../../.agents/notes/implemented/architecture/2026-07-16-explicit-turn-cancellation.md) owns signal lifetime; the [agent-scope runtime-design Agent Note](../../../.agents/notes/implemented/architecture/2026-07-12-agent-scope-runtime-design.md#three-execution-boundaries-are-deliberately-one-way) owns scoped dispatch and terminal settlement.
+Most interception points are cooperative waterfalls. `agent/pre-step` receives a payload carrying the subject `agent`, the exclusive claimed `UserMessage[]`, and the proposed `turn`, `step`, and cancellation `signal`; its batch may be empty when tools already require another request. Agent-scoped turn seams carry their explicit `AbortSignal` in the payload; the remaining turn-scoped seams receive it through their request value. Listeners may cooperate with a signal but must not retain it as authority over another turn. `agent/request-error` is the failed-model-request recovery waterfall: it receives request coordinates, normalized failure facts, the serving registration's retry policy when available, and the signal. A listener returns `{ kind: 'retry' }` without calling `next()` when it owns recovery. `agent/turn-stopping` runs before an otherwise completed turn closes. The [explicit-cancellation decision](../../../.agents/notes/implemented/architecture/2026-07-16-explicit-turn-cancellation.md) owns signal lifetime; the [agent-scope runtime-design Agent Note](../../../.agents/notes/implemented/architecture/2026-07-12-agent-scope-runtime-design.md#three-execution-boundaries-are-deliberately-one-way) owns scoped dispatch and terminal settlement.

 `PreStepDecision` is either `{ kind: 'reject' }` or `{ kind: 'enter', messages }`. The enter branch is the complete identified, frozen batch for the proposed step. A listener that wraps downstream entry preserves that batch unless it intentionally replaces it; additions follow the waterfall's natural return order. Claiming already removed the offered messages from the inbox, so rejection does not retain them. Messages inserted after the claim remain pending for a later boundary.

--- a/packages/core/agent/README.zh.md
+++ b/packages/core/agent/README.zh.md
@@ -50,7 +50,7 @@ Agent *创建* 由实现 `AgentFactory` 的插件（`dsh-agent-loop`）提供，

 生命周期边有两个重要的本地注意事项。`agent/created` 在作用域 setup 之后、会话与 agent 注册表条目都存在之后运行。Setup 是受信任、仅用于组合的代码；紧随其后且不可 veto 的 `agent/session-start` 通知是第一个受支持的启动注入点。`agent/disposed` 始终表示确切 agent 已离开注册表。AgentLoop 在其驱动器完全停稳后发出该事件，而有序 teardown 此时可能仍在分离会话并撤销作用域；直接注册的自定义 agent 自行拥有任何更强的驱动器顺序契约。

-大多数拦截点都是协作式 waterfall（瀑布式事件）。`agent/pre-step` 接收独占的已领取 `UserMessage[]`，以及包含拟进入 `turn`、`step` 与取消 `signal` 的 `PreStepContext`；当工具已经要求继续请求时，该批次可以为空。其他轮次作用域异步 seam 仍按位置接收显式 `AbortSignal`。监听器可以配合信号，但不得将它保留为控制另一轮次的权限。`agent/request-error` 是失败模型请求的恢复 waterfall：它接收请求坐标、规范化失败事实、可用时提供服务的注册项重试策略以及信号。拥有恢复权的监听器返回 `{ kind: 'retry' }` 且不调用 `next()`。`agent/turn-stopping` 在本可完成的轮次关闭前运行。信号生命周期由[显式取消决策](../../../.agents/notes/implemented/architecture/2026-07-16-explicit-turn-cancellation.md)拥有；作用域分发与终止结算由 [agent 作用域 runtime 设计 Agent Note（agent 决策记录）](../../../.agents/notes/implemented/architecture/2026-07-12-agent-scope-runtime-design.md#three-execution-boundaries-are-deliberately-one-way)拥有。
+大多数拦截点都是协作式 waterfall（瀑布式事件）。`agent/pre-step` 接收一个 payload，携带主体 `agent`、独占的已领取 `UserMessage[]` 以及拟进入的 `turn`、`step` 与取消 `signal`；当工具已经要求继续请求时，该批次可以为空。agent 作用域轮次 seam 在 payload 中携带显式 `AbortSignal`；其余轮次作用域 seam 通过其请求值接收它。监听器可以配合信号，但不得将它保留为控制另一轮次的权限。`agent/request-error` 是失败模型请求的恢复 waterfall：它接收请求坐标、规范化失败事实、可用时提供服务的注册项重试策略以及信号。拥有恢复权的监听器返回 `{ kind: 'retry' }` 且不调用 `next()`。`agent/turn-stopping` 在本可完成的轮次关闭前运行。信号生命周期由[显式取消决策](../../../.agents/notes/implemented/architecture/2026-07-16-explicit-turn-cancellation.md)拥有；作用域分发与终止结算由 [agent 作用域 runtime 设计 Agent Note（agent 决策记录）](../../../.agents/notes/implemented/architecture/2026-07-12-agent-scope-runtime-design.md#three-execution-boundaries-are-deliberately-one-way)拥有。

 `PreStepDecision` 要么是 `{ kind: 'reject' }`，要么是 `{ kind: 'enter', messages }`。enter 分支是拟进入步骤的完整、带标识且冻结的批次。包装下游 enter 的监听器会保留该批次，除非有意替换它；新增消息遵循 waterfall 的自然返回顺序。领取操作已经把候选消息从 inbox 删除，因此 reject 不会保留它们；领取后插入的消息仍等待后续边界。

--- a/packages/core/agent/src/dispatch.ts
+++ b/packages/core/agent/src/dispatch.ts
@@ -1,7 +1,8 @@
 /**
- * Agent-scoped dispatch and prompt assembly helpers. Ordinary events use the
- * fused dispatcher so subject and scope key cannot diverge; registry lifecycle
- * code instead captures one stable carrier for both edges.
+ * Agent-scoped dispatch and prompt assembly helpers. The fused dispatcher
+ * {@link agentEvents} couples the agent subject to its scope carrier, so the
+ * scope key and the payload's `agent` cannot diverge; repeat dispatchers (the
+ * loop driver) build it once in the agent's constructor and reuse it.
 * @module @deepseek-ai/dsh-agent/dispatch
 */

@@ -83,9 +84,10 @@ export interface AgentEventDispatch {
 /**
 * Build the fused scope carrier for one agent subject.
 *
- * The carrier is a stateless routing object; callers that dispatch repeatedly
- * for the same agent (the loop driver) build it once in the agent's
- * constructor and reuse it, so hot-path dispatches never allocate.
+ * The carrier is a stateless routing object. {@link agentEvents} accepts an
+ * existing carrier, so callers that dispatch repeatedly for the same agent
+ * (the loop driver) build it once in the agent's constructor and reuse it,
+ * keeping hot-path dispatches allocation-free.
 * @param agent - the subject agent and scope key.
 * @returns the carrier passed as the event dispatcher `this` value.
 */
@@ -97,10 +99,12 @@ export function agentCarrier(agent: Agent): Scoped<Agent> {
 * Build a dispatcher that couples the agent subject to its scope carrier.
 * @param ctx - the context to dispatch through (any context of the app).
 * @param agent - the subject agent; also the scope-carrier key.
+ * @param carrier - the scope carrier to dispatch through; defaults to
+ * {@link agentCarrier} for the agent. Pass a constructor-built carrier to
+ * avoid rebuilding it for every dispatch.
 * @returns the fused dispatcher.
 */
-export function agentEvents(ctx: Context, agent: Agent): AgentEventDispatch {
-  const carrier = agentCarrier(agent)
+export function agentEvents(ctx: Context, agent: Agent, carrier: Scoped<Agent> = agentCarrier(agent)): AgentEventDispatch {
  // The ordinary dispatch methods forward through Cordis' variadic mixins. The
  // fused (carrier, name, payload, ...rest) tuple is provably a valid argument
  // list for the matching thisArg overload, but TypeScript cannot relate the
@@ -108,8 +112,10 @@ export function agentEvents(ctx: Context, agent: Agent): AgentEventDispatch {
  // tuple — hence one contained, shape-preserving cast per method.
  const fused = <K extends AgentSubjectEvent>(payload: PayloadRest<K>): PayloadOf<K> =>
    // The dispatcher owns the subject injection; callers pass PayloadRest, so
-    // the fused record is exactly the declared payload.
-    ({ agent, ...payload } as PayloadOf<K>)
+    // the fused record is exactly the declared payload. The spread comes
+    // first, so a structurally acceptable payload that happens to carry an
+    // `agent` field can never override the injected subject.
+    ({ ...payload, agent } as PayloadOf<K>)
  return {
    emit(name, payload) {
      // Cordis emit invokes callbacks through Array.map: one synchronous throw
--- a/packages/core/agent/tests/agent.spec.ts
+++ b/packages/core/agent/tests/agent.spec.ts
@@ -11,6 +11,7 @@ import type {
  Agent,
  AgentCancelCause,
  AgentFactory,
+  AgentStatus,
  CreateAgentOptions,
  ResumeAgentOptions,
 } from '@deepseek-ai/dsh-agent'
@@ -305,6 +306,21 @@ describe('agentEvents()', () => {

    expect(heard).toEqual([{ agent, turn: 3, signal }])
  })
+
+  it('injects the fused subject even when the payload carries a conflicting agent field', async () => {
+    const ctx = new Context()
+    const agent = stubAgent('fused-subject')
+    const other = stubAgent('payload-agent')
+    const heard: Agent[] = []
+    ctx.on('agent/status', ({ agent: subject }) => void heard.push(subject))
+    // A structurally acceptable payload may carry an extra `agent` field; the
+    // dispatcher's injected subject must win over it.
+    const payload: { status: AgentStatus; agent: Agent } = { status: 'running', agent: other }
+
+    agentEvents(ctx, agent).emit('agent/status', payload)
+
+    expect(heard).toEqual([agent])
+  })
 })

 describe('explicit cancellation contract', () => {