Chương 3: Vòng lặp Agent

Chương 2 đã tách model transport, Agent runtime và sản phẩm coding. Agent Loop là phần chuyển động bên trong kiến trúc đó. Chương này bắt đầu từ lý do cần vòng lặp, rồi theo một message qua Pi 0.85.0: chuẩn bị context, streaming, thực thi Tool, chỉ dẫn trong hàng chờ, termination, event và thời điểm run settle hoàn toàn.

1. Mở đầu: ba cách dùng LLM

Mức quyền quyết định được giao cho model phân biệt direct call, Workflow và Agent Loop. Cả ba có thể dùng cùng provider và model; control flow của chúng khác nhau.

Kiểu 1: gọi trực tiếp, “model, hãy trả lời”

Direct call gửi một context đã chuẩn bị và nhận một response:

user input -> dựng Context -> models.streamSimple() -> AssistantMessage hoàn chỉnh

Entry point Pi AI hiện tại là một collection Models. Ví dụ đầy đủ này đăng ký một provider rồi gọi model một lần:

import { createModels, type Context } from "@earendil-works/pi-ai";
import { anthropicProvider } from "@earendil-works/pi-ai/providers/anthropic";

const models = createModels();
models.setProvider(anthropicProvider());

const model = models.getModel("anthropic", "claude-sonnet-4-6");
if (!model) throw new Error("Model not found");

const context: Context = {
  systemPrompt: "You are a translation assistant.",
  messages: [
    {
      role: "user",
      content: "Translate this TypeScript function into Python.",
      timestamp: Date.now(),
    },
  ],
};

const stream = models.streamSimple(model, context);
const response = await stream.result();
console.log(response.content);

Việc chính của ứng dụng là dựng prompt và xử lý response. Dịch, extraction, classification và câu hỏi có phạm vi hẹp thường phù hợp với cách này.

Kiểu 2: Workflow, “làm theo các bước do ứng dụng định sẵn”

Workflow gọi model nhiều lần, nhưng code ứng dụng cố định thứ tự và quyết định một bước đã đạt yêu cầu hay chưa:

input
  -> model: trích yêu cầu
  -> code: kiểm tra các field bắt buộc
  -> model: viết bản thay đổi
  -> code: chạy test
  -> model: giải thích lỗi hoặc viết summary

Model đưa ra phán đoán bên trong từng stage. Ứng dụng sở hữu state machine. Pipeline tài liệu, RAG, review gate và approval flow lặp lại được thường cần tính dự đoán đó.

Kiểu 3: Agent Loop, “tự chọn thao tác tiếp theo”

Agent cung cấp một tập Tool và để mỗi assistant response chọn thao tác kế tiếp:

user: "Giải thích vì sao test này fail"
  -> model yêu cầu read(test file)
  -> ứng dụng chạy read và thêm ToolResultMessage
  -> model yêu cầu grep(symbol)
  -> ứng dụng chạy grep và thêm ToolResultMessage
  -> model trả lời mà không có ToolCall
  -> run chạm một boundary ổn định

Ứng dụng vẫn kiểm soát Tool nào được cung cấp, permission, validation, stop hook, queue và error policy. Model chỉ chọn trong các thao tác đã được cho phép. Pi lặp lại model turn và Tool turn cho tới một exit boundary tường minh.

Sự phân quyền này có tác dụng trực tiếp. Model không thể chạy một function bất kỳ chỉ bằng cách gọi tên nó: Agent Core resolve tên trong AgentContext.tools, normalize và validate argument, rồi có thể block call trước khi code sản phẩm chạy. Model cũng không thể giữ process sống chỉ bằng một câu “hãy tiếp tục”. Turn kế tiếp cần một Tool batch, steering message hoặc follow-up message được runtime chấp nhận.

2. Hai khái niệm đầu tiên: Trace và Turn

Tên event public của Pi làm rõ sự khác biệt. Một run nằm giữa agent_start và agent_end. Một Turn nằm giữa turn_start và turn_end.

Trace: một run hoàn chỉnh

Chương này dùng Trace để chỉ toàn bộ run do một lần gọi Agent.prompt() hoặc Agent.continue() khởi động. “Trace” là thuật ngữ giảng giải ở đây, không phải type được Pi export.

Trace
├─ agent_start
├─ Turn 1: assistant yêu cầu read + grep; cả hai Tool settle
├─ Turn 2: assistant yêu cầu edit; Tool settle
├─ Turn 3: assistant trả lời mà không có ToolCall
└─ agent_end

Trace cũng có thể kết thúc sau một Turn, khi provider gặp hard failure, sau shouldStopAfterTurn hoặc tại deferred-response boundary. Các subscriber được Agent await vẫn thuộc quá trình settlement dù event agent_end đã được phát.

Turn: một assistant response cùng Tool batch của nó

Một Turn chứa đúng một assistant model response và mọi Tool call mà Pi chấp nhận từ response đó. Ba Tool call chạy parallel vẫn nằm trong một Turn:

turn_start
  -> một lần gọi streamFn(model, context, options)
  -> một AssistantMessage hoàn chỉnh
  -> không hoặc nhiều Tool execution từ message đó
  -> không hoặc nhiều ToolResultMessage
turn_end

Model call tiếp theo mở Turn khác. User prompt đầu tiên được phát bên trong Turn đầu, trước khi assistant streaming bắt đầu. Steering và follow-up message cũng được phát khi vòng lặp inject chúng trước một assistant response sau đó.

turn_end.toolResults chứa các Tool result message đã finalize cho assistant response ấy. Turn không có call mang một array rỗng. Response có nhiều call vẫn chỉ phát một turn_end sau khi cả batch settle; Pi không đóng rồi mở lại Turn quanh từng Tool. Boundary này cho session persistence và telemetry một đơn vị ổn định mà vẫn giữ event tiến độ chi tiết của từng Tool.

Quan hệ giữa Trace và Turn

một Trace
│
├─ Turn 1
│  ├─ AssistantMessage: ToolCall(read), ToolCall(grep)
│  └─ ToolResultMessage(read), ToolResultMessage(grep)
│
├─ Turn 2
│  ├─ AssistantMessage: ToolCall(edit)
│  └─ ToolResultMessage(edit)
│
└─ Turn 3
   └─ AssistantMessage: text, không có ToolCall

Phân biệt này tránh hai lỗi đếm phổ biến: coi từng Tool trong một parallel batch là một Turn, hoặc coi cả prompt nhiều Turn là một Turn duy nhất.

3. Toàn cảnh: hành trình message và vòng lặp

Hành trình đầy đủ

Đường đi đầy đủ của agent.prompt("Read src/main.ts and explain it") là:

string input
  -> UserMessage đã normalize
  -> agent_start, turn_start, user message_start/message_end
  -> transformContext(AgentMessage[])
  -> convertToLlm(AgentMessage[]) -> Message[]
  -> streamFn(model, Context, options)
  -> assistant message_start/message_update*/message_end
  -> chọn ToolCall block từ AssistantMessage.content
  -> Tool preflight và execution
  -> event ToolResultMessage và append transcript
  -> turn_end
  -> shouldStopAfterTurn(Turn đã hoàn tất)
  -> drain steering queue
  -> nếu cần inner Turn khác: prepareNextTurn
  -> nếu lần poll trước rỗng: refresh steering queue sau preparation
  -> Turn khác, outer loop follow-up hoặc agent_end

Trong một Tool turn bình thường, transcript tăng theo thứ tự hội thoại. Artifact này chỉ hiển thị các field được chọn, không phải protocol object đầy đủ:

[
  { "role": "user", "content": "Read src/main.ts" },
  {
    "role": "assistant",
    "content": [{ "type": "toolCall", "id": "call_1", "name": "read" }]
  },
  {
    "role": "toolResult",
    "toolCallId": "call_1",
    "toolName": "read",
    "isError": false
  }
]

AgentState.streamingMessage expose assistant message tạm thời trong lúc nó được dựng. AgentState.pendingToolCalls theo dõi Tool call ID giữa tool_execution_start và tool_execution_end. Message hoàn chỉnh đi vào AgentState.messages tại message_end.

Vòng lặp còn giữ newMessages, một collector cục bộ của run được raw stream trả về và gắn vào agent_end. Với prompt run, collector bắt đầu bằng input message; với continuation run, nó bắt đầu rỗng. Sau đó nó nhận assistant output, Tool result và queue message đã inject. Vì vậy session bên ngoài có thể phân biệt context cũ với artifact sinh trong invocation hiện tại khi quyết định lưu hay render gì.

Điều gì làm vòng lặp chạy tiếp, và điều gì kết thúc nó

Implementation cũ dễ bị tóm tắt quá mức thành “kiểm tra stopReason”. Pi 0.85.0 dùng nhiều mảnh state:

assistant response
  ├─ error / aborted ------------------------------> hard exit
  ├─ ToolCall block -------------------------------> Tool batch
  │    ├─ batch không terminate -------------------> tự động mở Turn tiếp
  │    └─ mọi result terminate=true ---------------> không tự tiếp tục vì Tool
  ├─ shouldStopAfterTurn=true ---------------------> graceful exit trước queue
  ├─ steering message -----------------------------> Turn tiếp trong inner loop
  ├─ follow-up message tại boundary ổn định -------> mở lại inner loop
  └─ không có điều kiện trên ----------------------> agent_end

StopReason vẫn ghi lại lý do provider streaming kết thúc:

pending là giá trị khởi tạo hoặc tạm thời khi một số provider stream còn chạy. Nó không phải final reason thành công của event done. Final message có deferred mang một DeferredHandle; host phải fetch hoặc cancel qua Models.fetchDeferred() hay Models.cancelDeferred() bên ngoài Agent Loop này.

Vì vậy báo cáo termination phải nêu cả provider result lẫn runtime state. “Model trả về stop” chưa đủ nếu steering instruction đã nằm trong queue. “Một Tool trả terminate: true” chưa đủ nếu result khác trong cùng batch không trả. “Stream đã kết thúc” chưa đủ khi final message là deferred và host vẫn phải xử lý nó. Điều kiện hoàn tất quan sát được thuộc về toàn bộ run, không thuộc một field trên một message.

Một rule dẫn dắt automatic continuation

Quyết định cốt lõi dựa vào content và state của Tool đã finalize, không dựa vào một string:

// Abridged from packages/agent/src/agent-loop.ts at 107d79f1.
const toolCalls = message.content.filter((part) => part.type === "toolCall");
hasMoreToolCalls = false;

if (toolCalls.length > 0) {
  const batch =
    message.stopReason === "length"
      ? await failToolCallsFromTruncatedMessage(toolCalls, emit)
      : await executeToolCalls(currentContext, message, config, signal, emit);

  hasMoreToolCalls = !batch.terminate;
}

Reason toolUse mà không có ToolCall block không ép mở Turn khác. Ngược lại, Tool block hợp lệ điều khiển execution trong khi runtime vẫn phải xử lý riêng length, abort, hook và queue. Sau Tool execution, batch termination chỉ tắt automatic Tool continuation; steering hoặc follow-up vẫn có thể nối dài run.

Vòng lặp tối thiểu: mẫu số chung

Có thể dạy Agent loop nhỏ nhất mà chưa cần hook riêng của Pi. Đoạn sau là pseudocode, không phải API để copy:

// Pseudocode
while (true) {
  const assistant = await callModel(messages, tools);
  messages.push(assistant);

  const calls = getCompleteToolCalls(assistant);
  if (calls.length === 0) break;

  const results = await executeAllowedTools(calls);
  messages.push(...results);
}

Đó là nhịp ReAct: model reason thành một action, ứng dụng observe action bằng cách chạy Tool, rồi observation quay lại dưới dạng Tool result. Code production cần thêm cancellation, validation, event ordering, queue policy, custom message và settlement guarantee.

Mọi đường thoát

Với message deferred không có Tool, loop phát turn_end rồi chạy shouldStopAfterTurn trên snapshot của Turn đã hoàn tất. Nếu hook trả truthy, loop phát agent_end ngay. Nếu không, loop poll steering. Steering message được trả về sẽ yêu cầu inner Turn khác, với preparation chạy trước khi message được inject; nếu steering rỗng, loop kiểm tra follow-up tại stable boundary. Cả hai lần poll đều không fetch hay cancel DeferredHandle; việc đó vẫn thuộc về host.

Agent.abort() signal provider request và Tool callback đang hoạt động. Cancellation phía provider thường thành một assistant message aborted. Nếu signal đến trong Tool processing, Tool đã start nhận signal; sequential preparation dừng sau khi quan sát abort, còn provider boundary tiếp theo nhận signal đã aborted. Tool phải tôn trọng signal thì cancellation mới kịp thời.

4. Đi qua source: loop nền và lớp của coding-agent

Loop tái sử dụng hiện nằm trong @earendil-works/pi-agent-core. Sản phẩm coding không duy trì một loop private khác. Nó dựng Agent, cung cấp wrapper StreamFn, convert message riêng của coding, refresh model/system prompt/Tool giữa các Turn, map terminal input thành steering hoặc follow-up, rồi lưu state được phát qua event.

Kernel khái niệm vẫn ngắn:

// Pseudocode: conceptual kernel only.
for (;;) {
  const assistant = await streamAssistant(context);
  const batch = await executeCompleteToolCalls(assistant);
  if (!batch.needsAnotherModelTurn) break;
}

Coding Agent phủ thêm những gì

Đây là một correction so với source tour cũ: steering, follow-up, turn hook và parallel Tool scheduling là tính năng Agent Core tại revision đã pin. Coding Agent đưa product policy vào qua các extension point đó.

4.1 Entry: Agent, agentLoop() và agentLoopContinue()

Phần lớn ứng dụng nên đi qua Agent. Model stream function hiện tại phải được bind với instance Models của nó:

import { Agent } from "@earendil-works/pi-agent-core";
import { createModels } from "@earendil-works/pi-ai";
import { anthropicProvider } from "@earendil-works/pi-ai/providers/anthropic";

const models = createModels();
models.setProvider(anthropicProvider());
const model = models.getModel("anthropic", "claude-sonnet-4-6");
if (!model) throw new Error("Model not found");

const agent = new Agent({
  initialState: {
    systemPrompt: "Inspect code before making a claim.",
    model,
    tools: [],
  },
  streamFn: models.streamSimple.bind(models),
});

await agent.prompt("Explain the build scripts in package.json.");

Hai low-level function được export trả về EventStream<AgentEvent, AgentMessage[]>:

function agentLoop(
  prompts: AgentMessage[],
  context: AgentContext,
  config: AgentLoopConfig,
  signal: AbortSignal | undefined,
  streamFn: StreamFn,
): EventStream<AgentEvent, AgentMessage[]>;

function agentLoopContinue(
  context: AgentContext,
  config: AgentLoopConfig,
  signal: AbortSignal | undefined,
  streamFn: StreamFn,
): EventStream<AgentEvent, AgentMessage[]>;

agentLoopContinue() yêu cầu context không rỗng và message cuối không phải assistant message. Sau convertToLlm, tail phía provider phải là user hoặc toolResult. Path này dành cho một continuation đã được chuẩn bị; nó không tự tạo retry prompt.

Low-level caller còn phải tự sở hữu transcript persistence. agentLoop() tạo working message array và trả artifact sinh trong invocation qua stream.result(); AgentContext truyền vào không thay thế cho một Agent có state. Caller muốn chạy low-level lần độc lập khác phải tự merge result vào context của mình. Vì thế wrapper Agent là default an toàn hơn.

Low-level prompt input có shape AgentMessage bình thường:

const prompts: AgentMessage[] = [
  {
    role: "user",
    content: "Inspect package.json.",
    timestamp: Date.now(),
  },
];

Loop nhận một context snapshot:

const context: AgentContext = {
  systemPrompt: "Be precise.",
  messages: [],
  tools: [],
};

Behavior của nó đến từ callback và stream option:

const config: AgentLoopConfig = {
  model,
  convertToLlm: (messages) =>
    messages.filter(
      (message) =>
        message.role === "user" ||
        message.role === "assistant" ||
        message.role === "toolResult",
    ),
  toolExecution: "parallel",
};

Agent public tạo snapshot của system prompt, message và Tool, chạy runAgentLoop hoặc runAgentLoopContinue, rồi reduce event ngược về AgentState đang live.

4.2 Bộ xương runLoop(): lõi trước, lớp ngoài sau

Lõi: inner loop

Điều kiện inner loop ở revision đã pin gồm cả automatic Tool continuation và message được inject:

// Faithfully abridged from packages/agent/src/agent-loop.ts.
let lastCompletedTurn: PrepareNextTurnContext | undefined;

while (hasMoreToolCalls || pendingMessages.length > 0) {
  if (lastCompletedTurn) {
    const nextTurnSnapshot = await config.prepareNextTurn?.(lastCompletedTurn);
    if (nextTurnSnapshot) {
      currentContext = nextTurnSnapshot.context ?? currentContext;
      config = {
        ...config,
        model: nextTurnSnapshot.model ?? config.model,
        reasoning:
          nextTurnSnapshot.thinkingLevel === undefined
            ? config.reasoning
            : nextTurnSnapshot.thinkingLevel === "off"
              ? undefined
              : nextTurnSnapshot.thinkingLevel,
      };
    }
    if (pendingMessages.length === 0) {
      pendingMessages = (await config.getSteeringMessages?.()) || [];
    }
    await emit({ type: "turn_start" });
  }

  if (pendingMessages.length > 0) {
    for (const pendingMessage of pendingMessages) {
      await emit({ type: "message_start", message: pendingMessage });
      await emit({ type: "message_end", message: pendingMessage });
      currentContext.messages.push(pendingMessage);
      newMessages.push(pendingMessage);
    }
    pendingMessages = [];
  }

  const message = await streamAssistantResponse(
    currentContext,
    config,
    signal,
    emit,
    streamFunction,
  );
  newMessages.push(message);

  if (message.stopReason === "error" || message.stopReason === "aborted") {
    await emit({ type: "turn_end", message, toolResults: [] });
    await emit({ type: "agent_end", messages: newMessages });
    return;
  }

  const toolCalls = message.content.filter((part) => part.type === "toolCall");
  const toolResults: ToolResultMessage[] = [];
  hasMoreToolCalls = false;
  if (toolCalls.length > 0) {
    const executedToolBatch =
      message.stopReason === "length"
        ? await failToolCallsFromTruncatedMessage(toolCalls, emit)
        : await executeToolCalls(currentContext, message, config, signal, emit);
    toolResults.push(...executedToolBatch.messages);
    hasMoreToolCalls = !executedToolBatch.terminate;
    for (const result of toolResults) {
      currentContext.messages.push(result);
      newMessages.push(result);
    }
  }

  await emit({ type: "turn_end", message, toolResults });
  lastCompletedTurn = {
    message,
    toolResults,
    context: currentContext,
    newMessages,
  };

  if (await config.shouldStopAfterTurn?.(lastCompletedTurn)) {
    await emit({ type: "agent_end", messages: newMessages });
    return;
  }

  pendingMessages = (await config.getSteeringMessages?.()) || [];
}

hasMoreToolCalls bắt đầu bằng true để assistant response đầu tiên chạy dù không có pending message. Mỗi iteration tiếp theo ứng với một Turn mới. Bản rút gọn bỏ phần thân chi tiết của helper nhưng giữ nguyên thứ tự control flow trong file đã pin.

Lớp ngoài: queue shell và stateful wrapper

Có hai shell quanh kernel:

Agent wrapper
  ├─ public state có thể thay đổi
  ├─ subscriber được await
  ├─ AbortController và settlement promise
  └─ object steering/follow-up queue
       |
       └─ runLoop outer while(true)
            ├─ inner while(Tool continuation || pending message)
            └─ khi inner dừng: drain follow-up queue hoặc break

Outer while (true) không phải một model algorithm khác. Nó chỉ mở lại inner loop khi follow-up message tồn tại đúng lúc Agent sắp dừng.

4.3 Inject steering

Ứng dụng enqueue một AgentMessage hoàn chỉnh:

agent.steer({
  role: "user",
  content: "Read the test fixture instead of production data.",
  timestamp: Date.now(),
});

Queue không ngắt provider stream đang hoạt động hoặc Tool đang chạy. Agent Core poll steering trước inner-loop iteration đầu tiên và sau mỗi Turn hoàn chỉnh có stop hook trả falsy. Nếu Tool continuation hoặc lần poll sau Turn đó yêu cầu iteration khác, loop chạy prepareNextTurn; khi lần poll trước rỗng, loop poll thêm một lần sau preparation để steering được queue trong một hook chạy lâu có thể vào Turn kế tiếp:

if (pendingMessages.length > 0) {
  for (const message of pendingMessages) {
    await emit({ type: "message_start", message });
    await emit({ type: "message_end", message });
    currentContext.messages.push(message);
    newMessages.push(message);
  }
  pendingMessages = [];
}

one-at-a-time drain message cũ nhất ở mỗi lần poll. all drain cả queue. Vì việc poll diễn ra tại Turn boundary, “steering” có nghĩa là ưu tiên Turn kế tiếp, không phải preempt giữa lúc Tool chạy.

4.4 streamAssistantResponse(): boundary của model

Pha A: transform Agent context

Hook đầu tiên làm việc hoàn toàn trong domain application message giàu thông tin hơn:

let messages = context.messages;
if (config.transformContext) {
  messages = await config.transformContext(messages, signal);
}

Coding Agent dùng stage này để Extension transform context. Compaction hoặc retrieval cũng có thể trả một AgentMessage[] khác. Contract quy định hook không được throw; khi thất bại, nó phải trả message ban đầu hoặc một fallback an toàn.

AgentMessage[] bền vững
  -> transformContext()
  -> AgentMessage[] dành riêng cho Turn

Array trả về chỉ chuẩn bị một request. Nó không thay transcript bền vững trừ khi application policy chủ động cập nhật state ở bước riêng.

Pha B: convert AgentMessage thành Message

convertToLlm là bước bắt buộc tại low-level boundary:

const llmMessages = await config.convertToLlm(messages);

Converter mặc định của Agent giữ role user, assistant và toolResult. Coding Agent còn map bashExecution, custom, branchSummary và compactionSummary thành user message; Bash message bị đánh dấu exclude sẽ bị lọc:

// Faithfully abridged from packages/coding-agent/src/core/messages.ts.
switch (m.role) {
  case "bashExecution":
    if (m.excludeFromContext) return undefined;
    return {
      role: "user",
      content: [{ type: "text", text: bashExecutionToText(m) }],
      timestamp: m.timestamp,
    };
  case "custom": {
    const content =
      typeof m.content === "string"
        ? [{ type: "text" as const, text: m.content }]
        : m.content;
    return { role: "user", content, timestamp: m.timestamp };
  }
  case "branchSummary":
    return {
      role: "user",
      content: [
        {
          type: "text" as const,
          text: BRANCH_SUMMARY_PREFIX + m.summary + BRANCH_SUMMARY_SUFFIX,
        },
      ],
      timestamp: m.timestamp,
    };
  case "compactionSummary":
    return {
      role: "user",
      content: [
        {
          type: "text" as const,
          text:
            COMPACTION_SUMMARY_PREFIX + m.summary + COMPACTION_SUMMARY_SUFFIX,
        },
      ],
      timestamp: m.timestamp,
    };
  case "user":
  case "assistant":
  case "toolResult":
    return m;
}

bashExecutionToText() cùng hai cặp hằng prefix/suffix cho summary được khai báo trong chính file đã pin. Đoạn trích giữ đủ bốn nhánh role riêng của coding thay vì dùng helper chuyển đổi không tồn tại.

Type transition này cố ý làm mất thông tin:

AgentMessage[]                           Message[]
├─ user ------------------------------> user
├─ assistant -------------------------> assistant
├─ toolResult ------------------------> toolResult
├─ compactionSummary -----------------> user summary
├─ bashExecution(excluded) -----------> bị loại
└─ custom ----------------------------> user content

Provider adapter không cần hiểu message type của storage hoặc UI trong Coding Agent.

Thứ tự hai hook là một phần của contract. transformContext có thể xử lý application-only type trước khi dữ liệu bị loại. Sau đó convertToLlm mới projection lần cuối sang provider union. Nếu đảo thứ tự, compaction hoặc Extension logic sẽ không thấy message không được gửi thẳng cho model nhưng vẫn mang state hữu ích của ứng dụng.

Pha C: dựng Context và gọi model đã chọn

Loop tạo một provider-facing wrapper mới cho mỗi Turn:

// Faithfully abridged from packages/agent/src/agent-loop.ts at 107d79f1.
const llmContext: Context = {
  systemPrompt: context.systemPrompt,
  messages: llmMessages,
  tools: context.tools,
};

Nó resolve API key hiện hành rồi gọi function đã inject:

// Faithfully abridged from packages/agent/src/agent-loop.ts at 107d79f1.
const response = await streamFunction(config.model, llmContext, {
  ...config,
  apiKey: resolvedApiKey,
  signal,
});

Với ứng dụng Agent Core thông thường, hãy truyền method của model collection hiện tại kèm receiver:

const streamFn = models.streamSimple.bind(models);

Coding Agent bọc cùng contract thay vì truyền Models trực tiếp:

// Faithfully abridged from packages/coding-agent/src/core/sdk.ts.
streamFn: async (model, context, options) => {
  const providerRetrySettings = settingsManager.getProviderRetrySettings();
  const httpIdleTimeoutMs = settingsManager.getHttpIdleTimeoutMs();
  const effectiveTimeoutMs =
    httpIdleTimeoutMs === 0 ? 2147483647 : httpIdleTimeoutMs;
  const timeoutMs =
    options?.timeoutMs ?? providerRetrySettings.timeoutMs ?? effectiveTimeoutMs;
  const websocketConnectTimeoutMs =
    options?.websocketConnectTimeoutMs ??
    settingsManager.getWebSocketConnectTimeoutMs();
  const headerRunner = extensionRunnerRef.current;

  return modelRuntime.streamSimple(model, context, {
    ...options,
    timeoutMs,
    websocketConnectTimeoutMs,
    maxRetries: options?.maxRetries ?? providerRetrySettings.maxRetries,
    maxRetryDelayMs:
      options?.maxRetryDelayMs ?? providerRetrySettings.maxRetryDelayMs,
    transformHeaders: async (requestHeaders) => {
      const headers = mergeProviderAttributionHeaders(
        model,
        settingsManager,
        options?.sessionId,
        requestHeaders,
      );
      return headerRunner?.hasHandlers("before_provider_headers")
        ? headerRunner.emitBeforeProviderHeaders(headers ?? {})
        : (headers ?? {});
    },
  });
},

Scope bao quanh trong sdk.ts cung cấp settingsManager, extensionRunnerRef, mergeProviderAttributionHeaders và modelRuntime. Wrapper áp dụng setting timeout và retry, provider attribution cùng Extension header hook. Việc resolve credential vẫn thuộc Agent Loop qua getApiKey; loop chỉ nhìn thấy StreamFn.

Provider adapter sở hữu cách serialize cache control. Việc dựng lại object Context nhỏ không tự quyết định cache hit; content provider nhìn thấy và cache semantics của provider mới quyết định.

Pha D: stream và thay assistant message tại chỗ

streamAssistantResponse() dành một slot transcript ở event start, thay slot đó bằng từng partial, rồi thay lần cuối bằng message hoàn chỉnh:

// Faithfully abridged from packages/agent/src/agent-loop.ts at 107d79f1.
case "start":
  partialMessage = event.partial;
  context.messages.push(partialMessage);
  await emit({ type: "message_start", message: { ...partialMessage } });
  break;

case "text_delta":
case "toolcall_delta":
case "thinking_delta":
  partialMessage = event.partial;
  context.messages[context.messages.length - 1] = partialMessage;
  await emit({ type: "message_update", message: { ...partialMessage }, assistantMessageEvent: event });
  break;

case "done":
case "error":
  finalMessage = await response.result();
  context.messages[context.messages.length - 1] = finalMessage;
  await emit({ type: "message_end", message: finalMessage });
  return finalMessage;

Switch thật còn xử lý event *_start và *_end. Nếu stream hoàn tất mà không phát start, implementation append final message và tự phát message_start trước message_end.

start          messages[last] = AssistantMessage rỗng/ban đầu
text_delta     messages[last] = AssistantMessage partial mới hơn
toolcall_end   messages[last] = partial có ToolCall hoàn chỉnh
done/error     messages[last] = AssistantMessage cuối

Một slot tránh lưu mỗi token delta thành conversation message. Subscriber vẫn nhận từng typed update để render.

4.5 Kiểm tra stop và termination

Hard-stop check chạy trước bước chọn Tool:

// Faithfully abridged from packages/agent/src/agent-loop.ts at 107d79f1.
if (message.stopReason === "error" || message.stopReason === "aborted") {
  await emit({ type: "turn_end", message, toolResults: [] });
  await emit({ type: "agent_end", messages: newMessages });
  return;
}

Với mọi final reason khác, loop kiểm tra Tool block thật. length là safety branch riêng: argument có thể parse được nhưng chưa đầy đủ, nên Pi phát failed Tool result cho từng call và không thực thi call nào. Sau một Turn bình thường, shouldStopAfterTurn nhận context của Turn đã hoàn tất trước, rồi đến lần poll steering thông thường. Chỉ inner loop đang tiếp tục mới chạy prepareNextTurn và có thể poll steering lần hai trước turn_start kế tiếp.

4.6 Thực thi Tool call

Hai mode giữ conversation order theo cách khác nhau:

Nếu bất kỳ Tool được gọi nào khai báo executionMode: "sequential", toàn bộ assistant batch chạy sequential. Preflight resolve Tool, áp dụng prepareArguments, validate schema rồi gọi beforeToolCall:

// Faithfully abridged from packages/agent/src/agent-loop.ts at 107d79f1.
const preparedToolCall = prepareToolCallArguments(tool, toolCall);
const validatedArgs = validateToolArguments(tool, preparedToolCall);
const beforeResult = await config.beforeToolCall?.(
  { assistantMessage, toolCall, args: validatedArgs, context: currentContext },
  signal,
);

Tool không tồn tại, argument sai, preflight code throw, call bị block và abort đã được quan sát đều trở thành immediate error result. afterToolCall chỉ chạy sau khi một Tool được phép đã thực thi; hook có thể thay content, details, usage, isError hoặc terminate trước final event:

// Faithfully abridged from packages/agent/src/agent-loop.ts at 107d79f1.
const afterResult = await config.afterToolCall?.(
  {
    assistantMessage,
    toolCall,
    args,
    result,
    isError,
    context: currentContext,
  },
  signal,
);

result = {
  ...result,
  content: afterResult?.content ?? result.content,
  details: afterResult?.details ?? result.details,
  usage: afterResult?.usage ?? result.usage,
  terminate: afterResult?.terminate ?? result.terminate,
};
isError = afterResult?.isError ?? isError;

Với mỗi call đã finalize, Pi phát tool_execution_end rồi một cặp message_start/message_end cho ToolResultMessage đã normalize. Trong batch không bị abort, mỗi call có một result. Nếu abort được quan sát khi đang prepare batch, các source call phía sau có thể chưa từng start.

Batch termination dùng every:

// Faithfully abridged from packages/agent/src/agent-loop.ts at 107d79f1.
const terminate =
  finalizedCalls.length > 0 &&
  finalizedCalls.every((entry) => entry.result.terminate === true);

Mixed batch vẫn tiếp tục. Result bị beforeToolCall block chỉ tham gia termination khi đặt cả block: true và terminate: true. afterToolCall có thể thêm hoặc bỏ termination hint của executed result.

Parallel mode cố ý tách preparation khỏi execution. Pi prepare call theo source order của assistant và ghi immediate failure trước khi launch đồng thời các call được phép. Vì thế tool_execution_end có thể phản ánh completion order thật, còn Tool result message đợi mọi work đã launch settle rồi quay về source order. Model nhìn thấy transcript deterministic dù UI cho thấy Tool này hoàn thành trước Tool kia.

Flag terminate chỉ tồn tại trong runtime. createToolResultMessage() copy content, details, usage, tên Tool mới được thêm, error state và field nhận dạng, nhưng không copy terminate. Provider request kế tiếp không nhận một termination field ngoài protocol. Agent Core tiêu thụ hint này khi quyết định có cần automatic continuation hay không.

4.7 turn_end, hook, event và steering

Thứ tự sau Turn được cố định:

turn_end
  -> lưu { message, toolResults, context, newMessages }
  -> shouldStopAfterTurn(snapshot của Turn đã hoàn tất)
  -> nếu true: agent_end
  -> nếu không: getSteeringMessages()
  -> nếu inner loop tiếp tục: prepareNextTurn(snapshot đã lưu)
  -> áp dụng context/model/thinkingLevel được trả về
  -> nếu lần poll sau Turn rỗng: getSteeringMessages() lần nữa
  -> turn_start

prepareNextTurn không tự ép mở Turn khác. Nó chạy ở đầu một iteration vốn đã được Tool continuation hoặc pending message yêu cầu. Coding Agent cài prepareNextTurnWithContext để refresh system prompt, Tool registry, model đã chọn và thinking level từ session state đang live.

Event path của một Tool Turn là:

Agent.processEvents() reduce state trước khi gọi listener. Listener được await theo thứ tự subscription, nên assistant message_end là một barrier: beforeToolCall nhìn thấy Agent.state.messages đã chứa assistant request. Raw agentLoop() stream giữ event order nhưng không biến asynchronous consumer work thành producer barrier.

Settlement kéo dài qua thời điểm phát event. agent_end bảo đảm loop không phát event sau đó, nhưng Agent.state.isStreaming vẫn là true khi listener agent_end được await. Chỉ finishRun() mới xóa streaming state và pending Tool ID, resolve waitForIdle() rồi gỡ active run. Nhờ vậy subscriber có thể flush session hoặc telemetry buffer trước khi await agent.prompt() trả về.

4.8 Quay lại đầu vòng lặp

// Faithfully abridged from packages/agent/src/agent-loop.ts at 107d79f1.
while (hasMoreToolCalls || pendingMessages.length > 0) {
  // one assistant response and its Tool batch
}

Automatic continuation đến từ Tool batch không terminate. Steering continuation đến từ lần poll sau Turn. Nếu Tool continuation đã yêu cầu iteration khác và lần poll đó rỗng, preparation sẽ poll steering thêm một lần trước turn_start. Nếu không có điều kiện continuation nào, inner loop kết thúc. shouldStopAfterTurn có thể thoát trước mọi quyết định queue sau Turn này.

4.9 Outer loop follow-up

Tại boundary ổn định, Agent Core chỉ poll follow-up queue:

// Faithfully abridged from packages/agent/src/agent-loop.ts at 107d79f1.
const followUpMessages = (await config.getFollowUpMessages?.()) || [];
if (followUpMessages.length > 0) {
  pendingMessages = followUpMessages;
  continue;
}
break;

Outer continue quay lại inner loop, nơi follow-up message nhận message event bình thường trước assistant call tiếp theo. Chúng vẫn thuộc cùng run agent_start/agent_end. Path hard error/aborted hoặc shouldStopAfterTurn return trước lần poll này.

Agent.continue() là public action riêng sau khi một run đã settle. Khi tail là user hoặc Tool result, nó bắt đầu run mới từ transcript state có sẵn. Khi tail là assistant, nó có thể dùng steering hoặc follow-up message đã nằm trong queue; nếu cả hai queue rỗng thì call bị reject. Mọi path được chấp nhận đều phát agent_start mới. Không nên nhầm nó với outer loop kéo dài run hiện tại.

4.10 Steering và follow-up

Coding Agent map input nhập khi streaming vào một trong hai queue và expose mode qua setting. Steering phù hợp với “hãy dùng fixture thay thế”. Follow-up phù hợp với “sau đó hãy tóm tắt diff”.

5. Tổng kết: ba thiết kế vòng lặp cần giữ

1. ReAct là nhịp cốt lõi

Reason trong AssistantMessage
  -> Act qua ToolCall
  -> Observe qua ToolResultMessage
  -> Reason lần nữa

Một Turn chứa một assistant response cùng Tool batch của nó. Một run có thể chứa nhiều Turn.

2. Termination là quyết định trên state

stopReason mô tả cách provider hoàn tất, nhưng control còn phụ thuộc Tool block, an toàn khi call bị cắt, terminate trên toàn batch, shouldStopAfterTurn, steering, follow-up, abort, error và quyền sở hữu deferred response. Runtime chỉ kết thúc tại boundary đã định nghĩa; nó không yêu cầu model chứng nhận task đã hoàn tất.

3. Giữ kernel nhỏ và đặt product policy thành lớp ngoài

Cách tách này cho phép một domain Agent nhỏ dùng Agent trực tiếp, trong khi coding assistant đầy đủ giữ policy phong phú hơn trong AgentSession và Extension.

6. Trạm tiếp theo

Chương 4 mở boundary StreamFn: model collection, provider registration, request conversion, normalized streaming event và error handling.

Version boundary: phần walkthrough này theo Pi 0.85.0 tại commit 107d79f11072bbc8a3a757ed7fd69596bee7d68c và Node.js >=22.19.0.