refactor(core): use initiator in loop internals

2026-07-19 14:06:47 +08:00
parent e0b5cadd2a
commit f21ed1bfa0
7 changed files with 28 additions and 29 deletions
--- a/packages/core/agent-loop/README.md
+++ b/packages/core/agent-loop/README.md
@@ -50,7 +50,7 @@ The concrete `Agent` class, its `Inbox`, `runLoop`, and instance-bound publicati

 ### Loop lifecycle (`loop.ts`)

-The driver owns one agent for its lifetime and runs inside `ctx.agents.withInitiator(agent, ...)`, so process-local asynchronous continuations can recover the initiating Agent. Creation, persistence load, and unpublished setup stay outside the driver boundary; explicit Agent fields remain authoritative at service, worker, process, persistence, and wire boundaries. The [agent service](../agent/README.md#initiating-agent-scope) owns propagation, teardown, and detached-work rules.
+The driver owns one agent for its lifetime and runs inside `ctx.agents.withInitiator(agent, ...)`, so package-private loop, turn, step, and tool-call helpers recover the exact Agent from `ctx.agents` instead of forwarding the concrete driver through their signatures. Creation, persistence load, and unpublished setup stay outside the driver boundary; explicit Agent fields remain authoritative at service, worker, process, persistence, and wire boundaries. The [agent service](../agent/README.md#initiating-agent-scope) owns propagation, teardown, and detached-work rules.

 Every provider call that reaches a successful finish appends exactly one `assistant/message` completion anchor, including content-less calls and `max-tokens` finishes. A successful `agent/step-result` stores its transformed content; a rejected result records empty content before the original failure continues. The anchor retains exact chunk provenance (`[]` for a stream with no chunks) and usage when available, while empty content stays out of derived message history.

--- a/packages/core/agent-loop/src/agent.ts
+++ b/packages/core/agent-loop/src/agent.ts
@@ -387,7 +387,7 @@ export class ReactLoopAgent implements Agent {
  [startDriver](): void {
    if (this._status === 'disposed') return
    this.driverStarted = true
-    this.done = this.loopCtx.agents.withInitiator(this, () => runLoop(this.loopCtx, this, {
+    this.done = this.loopCtx.agents.withInitiator(this, () => runLoop(this.loopCtx, {
      inbox: this.#inbox,
      maxParallelToolCalls: this.maxParallelToolCalls,
      setStatus: (status) => { this.setStatus(status) },
--- a/packages/core/agent-loop/src/loop.ts
+++ b/packages/core/agent-loop/src/loop.ts
@@ -19,7 +19,6 @@ 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'

 /** An Error with an optional machine-readable code (e.g., from LlmError or a throwing plugin). */
@@ -96,11 +95,13 @@ export interface LoopHandle {
 /**
 * Drive queued batches as durable turns until disposal. Plugin failures end the
 * current turn without terminating the driver.
- * @param ctx - the plugin context the loop reaches events (agent/…, session/flush) and services (systemPrompt, llm, tools) through.
- * @param agent - the agent this invocation drives for its whole lifetime (its inbox, session, and options).
+ * @param ctx - the plugin context the loop reaches its initiating Agent,
+ * events (agent/…, session/flush), and services (systemPrompt, llm, tools)
+ * through.
 * @param handle - the bridge to the agent's mutable state: status/abort setters plus the disposal and cancel-marker reads.
 */
-export async function runLoop(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle): Promise<void> {
+export async function runLoop(ctx: Context, handle: LoopHandle): Promise<void> {
+  const agent = ctx.agents.requireInitiator()
  // Per-instance prefix and request-header state; conversation history remains in the session log.
  const transmission = createTransmissionLog()

@@ -138,7 +139,7 @@ export async function runLoop(ctx: Context, agent: ReactLoopAgent, handle: LoopH
    const turn = lastTurnNumber(session) + 1
    let terminalStopped = false
    try {
-      terminalStopped = await runTurn(ctx, events, agent, handle, turn, transmission)
+      terminalStopped = await runTurn(ctx, events, handle, turn, transmission)
    } catch (error: unknown) {
      // Pre-turn failure has no durable boundary to close; report it without appending outside a turn.
      const err = toError(error)
@@ -161,8 +162,9 @@ export async function runLoop(ctx: Context, agent: ReactLoopAgent, handle: LoopH
 }

 async function runTurn(
-  ctx: Context, events: AgentEventDispatch, agent: ReactLoopAgent, handle: LoopHandle, turn: number, transmission: TransmissionLog,
+  ctx: Context, events: AgentEventDispatch, handle: LoopHandle, turn: number, transmission: TransmissionLog,
 ): Promise<boolean> {
+  const agent = ctx.agents.requireInitiator()
  const { session } = agent

  // Drain before opening the turn, but append only after `turn/start`.
@@ -262,7 +264,7 @@ async function runTurn(

      // Steering from the previous round's continuation listeners joins before
      // the request.
-      drainSteering(agent, handle.inbox, turn)
+      drainSteering(session, handle.inbox, turn)

      // The step's AbortController exists BEFORE any async pre-step work so a
      // dispose() or cancel() — in a synchronous turn-start listener or an
@@ -336,7 +338,7 @@ async function runTurn(
      let stepOutcome: { hadToolCalls: boolean; finish: FinishReason } | { error: Error }
      try {
        stepOutcome = await runStep(
-          ctx, events, agent, handle, turn, step, assembly, fullSystemPrompt, boundaryMessages, transmission, abort.signal)
+          ctx, events, handle, turn, step, assembly, fullSystemPrompt, boundaryMessages, transmission, abort.signal)
      } catch (error: unknown) {
        stepOutcome = { error: toError(error) }
      } finally {
@@ -365,7 +367,7 @@ async function runTurn(
      if (stepReason) reason = stepReason

      // Steering that arrived during streaming/tool execution.
-      const steered = drainSteering(agent, handle.inbox, turn)
+      const steered = drainSteering(session, handle.inbox, turn)

      closeStep()

@@ -455,10 +457,10 @@ async function runTurn(
 }

 /** Drain the steering queue into the session. Returns whether any arrived. */
-function drainSteering(agent: ReactLoopAgent, inbox: Inbox, turn: number): boolean {
+function drainSteering(session: Session, inbox: Inbox, turn: number): boolean {
  const messages = inbox.drainSteering()
  for (const message of messages) {
-    agent.session.append('steering/message', { turn, content: message.content, source: message.source }, { surfaceOp: 'append' })
+    session.append('steering/message', { turn, content: message.content, source: message.source }, { surfaceOp: 'append' })
  }
  return messages.length > 0
 }
@@ -472,7 +474,6 @@ function drainSteering(agent: ReactLoopAgent, inbox: Inbox, turn: number): boole
 async function runStep(
  ctx: Context,
  events: AgentEventDispatch,
-  agent: ReactLoopAgent,
  handle: LoopHandle,
  turn: number,
  step: number,
@@ -482,6 +483,7 @@ async function runStep(
  transmission: TransmissionLog,
  signal: AbortSignal,
 ): Promise<{ hadToolCalls: boolean; finish: FinishReason }> {
+  const agent = ctx.agents.requireInitiator()
  const { session, options } = agent

  // Seed the first request from agent options and later requests from the logged header;
@@ -567,7 +569,7 @@ async function runStep(
  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,
+      ctx, turn, step, toolCalls, signal, handle.maxParallelToolCalls, acceptContext,
    )
    return { hadToolCalls: true, finish: assembler.finish }
  })
--- a/packages/core/agent-loop/src/tool-calls.ts
+++ b/packages/core/agent-loop/src/tool-calls.ts
@@ -14,7 +14,6 @@ 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 {
@@ -35,7 +34,6 @@ interface Slot {
 * 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.
@@ -45,7 +43,6 @@ interface Slot {
 */
 export async function executeToolCalls(
  ctx: Context,
-  agent: ReactLoopAgent,
  turn: number,
  step: number,
  toolCalls: ToolCallBlock[],
@@ -53,7 +50,7 @@ export async function executeToolCalls(
  maxParallel: number,
  acceptContext: (context: HookContext) => void,
 ): Promise<void> {
-  const { session } = agent
+  const agent = ctx.agents.requireInitiator()

  // Inputs are distinct because tools/execute wrappers may replace `exec.signal`.
  const planned: PlannedCall[] = toolCalls.map(block => ({
@@ -74,7 +71,7 @@ export async function executeToolCalls(
    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)
+    next += await runGroup(ctx, turn, step, group, mode, signal, maxParallel, acceptContext)
  }
 }

@@ -96,7 +93,6 @@ function parseArguments(raw: string): unknown {
 */
 async function runGroup(
  ctx: Context,
-  session: Session,
  turn: number,
  step: number,
  group: PlannedCall[],
@@ -105,6 +101,7 @@ async function runGroup(
  maxParallel: number,
  acceptContext: (context: HookContext) => void,
 ): Promise<number> {
+  const { session } = ctx.agents.requireInitiator()
  /* 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)