云原生 Agent 开发（一）：从 LLM 响应到 Agent 会话

type ToolCall = {
  callId: string;
  name: string;
  argumentsText: string;
};

type ModelResult = {
  outcome: "completed" | "incomplete" | "failed";
  output: ModelOutput;          // 完整结构与 provider 续接材料
  toolCalls: ToolCall[];         // 已定稿的应用函数调用
};

async function runTask(context: Context, maxSteps: number) {
  for (let step = 0; step < maxSteps; step++) {
    const result: ModelResult = await model.collect(
      context.buildRequest(),
    );
    if (result.outcome !== "completed") {
      return {
        kind: "generation_stopped",
        outcome: result.outcome,
        output: result.output,
      };
    }
    context.appendModelOutput(result.output);
    if (result.toolCalls.length === 0) {
      return { kind: "finished", output: result.output };
    }

    for (const call of result.toolCalls) {
      const toolResult = await tools.executeValidated(call);
      // toolResult 保留 callId；可报告的工具错误也是一种结果
      context.appendToolResult(toolResult);
    }
  }
  return { kind: "budget_exhausted" };
}

type BaseModel[M messageType] interface {
  Generate(ctx context.Context, input []M, opts ...Option) (M, error)
  Stream(ctx context.Context, input []M, opts ...Option) (*schema.StreamReader[M], error)
}

type AgenticModel = BaseModel[*schema.AgenticMessage]

func (r *streamReceiver) functionCallArgumentsDeltaEventToContentBlock(ev responses.ResponseFunctionCallArgumentsDeltaEvent) *schema.ContentBlock {
  meta := &schema.StreamingMeta{
    Index: r.getBlockIndex(makeFunctionToolCallIndexKey(ev.OutputIndex)),
  }
  block := schema.NewContentBlockChunk(&schema.FunctionToolCall{
    Arguments: ev.Delta,
  }, meta)

  setItemID(block, ev.ItemID)

  return block
}

func concatFunctionToolCalls(calls []*FunctionToolCall) (*FunctionToolCall, error) {
  if len(calls) == 0 {
    return nil, fmt.Errorf("no function tool call found")
  }
  if len(calls) == 1 {
    return calls[0], nil
  }

  ret := &FunctionToolCall{}

  for _, c := range calls {
    if c == nil {
      continue
    }

    if ret.CallID == "" {
      ret.CallID = c.CallID
    } else if c.CallID != "" && c.CallID != ret.CallID {
      return nil, fmt.Errorf("expected call ID '%s' for function tool call, but got '%s'", ret.CallID, c.CallID)
    }

    if ret.Name == "" {
      ret.Name = c.Name
    } else if c.Name != "" && c.Name != ret.Name {
      return nil, fmt.Errorf("expected tool name '%s' for function tool call, but got '%s'", ret.Name, c.Name)
    }

    ret.Arguments += c.Arguments
  }

  return ret, nil
}

func (w *typedStateModelWrapper[M]) Stream(ctx context.Context, _ []M, opts ...model.Option) (*schema.StreamReader[M], error) {
  var (
    stateMessages          []M
    stateToolInfos         []*schema.ToolInfo
    stateDeferredToolInfos []*schema.ToolInfo
  )
  _ = compose.ProcessState(ctx, func(_ context.Context, st *typedState[M]) error {
    stateMessages = st.Messages
    stateToolInfos = st.ToolInfos
    stateDeferredToolInfos = st.DeferredToolInfos
    return nil
  })

  // Backfill: old checkpoints or fresh starts have nil ToolInfos.
  // Use compose-level tools from opts (which always reflects the latest bc.toolInfos)
  // rather than w.toolInfos (which may be stale if the graph was reused).
  if stateToolInfos == nil {
    composeLevelOpts := model.GetCommonOptions(&model.Options{}, opts...)
    if composeLevelOpts.Tools != nil {
      stateToolInfos = composeLevelOpts.Tools
    } else {
      stateToolInfos = w.toolInfos
    }
  }

  state := &TypedChatModelAgentState[M]{
    Messages:          stateMessages,
    ToolInfos:         stateToolInfos,
    DeferredToolInfos: stateDeferredToolInfos,
  }

  if msgState, ok := any(state).(*ChatModelAgentState); ok {
    for _, m := range w.middlewares {
      if m.BeforeChatModel != nil {
        if err := m.BeforeChatModel(ctx, msgState); err != nil {
          return nil, err
        }
      }
    }
  }

  baseOpts := &model.Options{Tools: w.toolInfos}
  commonOpts := model.GetCommonOptions(baseOpts, opts...)
  mc := &TypedModelContext[M]{Tools: commonOpts.Tools, ModelRetryConfig: w.modelRetryConfig, cancelContext: w.cancelContext}
  for _, handler := range w.handlers {
    var err error
    ctx, state, err = handler.BeforeModelRewriteState(ctx, state, mc)
    if err != nil {
      return nil, err
    }
  }

  // Persist state (including tool infos) after BeforeModelRewriteState.
  _ = compose.ProcessState(ctx, func(_ context.Context, st *typedState[M]) error {
    st.Messages = state.Messages
    st.ToolInfos = state.ToolInfos
    st.DeferredToolInfos = state.DeferredToolInfos
    return nil
  })

  // Derive model options from state. Append after caller opts so state takes precedence
  // (model.GetCommonOptions applies left-to-right, last wins).
  // Use explicit copy to avoid mutating the caller's opts slice.
  derivedOpts := make([]model.Option, len(opts), len(opts)+2)
  copy(derivedOpts, opts)
  derivedOpts = append(derivedOpts, model.WithTools(state.ToolInfos))
  if state.DeferredToolInfos != nil {
    derivedOpts = append(derivedOpts, model.WithDeferredTools(state.DeferredToolInfos))
  }

  wrappedEndpoint := w.wrapStreamEndpoint(w.inner.Stream)
  stream, err := wrappedEndpoint(ctx, state.Messages, derivedOpts...)
  if err != nil {
    return nil, err
  }
  result, err := concatMessageStream(stream)
  if err != nil {
    return nil, err
  }

  // Re-read State.Messages after Stream completes: same rationale as in Generate above.
  if w.modelRetryConfig != nil && w.modelRetryConfig.ShouldRetry != nil {
    _ = compose.ProcessState(ctx, func(_ context.Context, st *typedState[M]) error {
      state.Messages = st.Messages
      return nil
    })
  }

  state.Messages = append(state.Messages, result)

  for _, handler := range w.handlers {
    ctx, state, err = handler.AfterModelRewriteState(ctx, state, mc)
    if err != nil {
      return nil, err
    }
  }

  if msgState, ok := any(state).(*ChatModelAgentState); ok {
    for _, m := range w.middlewares {
      if m.AfterChatModel != nil {
        if err := m.AfterChatModel(ctx, msgState); err != nil {
          return nil, err
        }
      }
    }
  }

  // Persist state (including tool infos) after AfterModelRewriteState.
  _ = compose.ProcessState(ctx, func(_ context.Context, st *typedState[M]) error {
    st.Messages = state.Messages
    st.ToolInfos = state.ToolInfos
    st.DeferredToolInfos = state.DeferredToolInfos
    return nil
  })

  if len(state.Messages) == 0 {
    return nil, errors.New("no messages left in state after model call")
  }
  return schema.StreamReaderFromArray([]M{state.Messages[len(state.Messages)-1]}), nil
}

func (a *AgenticToolsNode) Invoke(ctx context.Context, input *schema.AgenticMessage, opts ...ToolsNodeOption) ([]*schema.AgenticMessage, error) {
  result, err := a.inner.Invoke(ctx, agenticMessageToToolCallMessage(input), opts...)
  if err != nil {
    return nil, err
  }
  return toolMessageToAgenticMessage(result), nil
}

func agenticMessageToToolCallMessage(input *schema.AgenticMessage) *schema.Message {
  var tc []schema.ToolCall
  for _, block := range input.ContentBlocks {
    if block.Type != schema.ContentBlockTypeFunctionToolCall || block.FunctionToolCall == nil {
      continue
    }
    tc = append(tc, schema.ToolCall{
      ID: block.FunctionToolCall.CallID,
      Function: schema.FunctionCall{
        Name:      block.FunctionToolCall.Name,
        Arguments: block.FunctionToolCall.Arguments,
      },
      Extra: block.Extra,
    })
  }
  return &schema.Message{
    Role:      schema.Assistant,
    ToolCalls: tc,
  }
}

func toolMessageToAgenticMessage(input []*schema.Message) []*schema.AgenticMessage {
  results := make([]*schema.AgenticMessage, len(input))
  for i, m := range input {
    if msg, ok := toolSearchResultMessageToAgenticMessage(m, nil); ok {
      results[i] = msg
      continue
    }

    ftr := &schema.FunctionToolResult{
      CallID: m.ToolCallID,
      Name:   m.ToolName,
    }
    if len(m.UserInputMultiContent) > 0 {
      ftr.Content = messageInputPartsToFunctionToolBlocks(m.UserInputMultiContent)
    } else if m.Content != "" {
      ftr.Content = []*schema.FunctionToolResultContentBlock{
        newFuncToolResultContentBlock(&schema.UserInputText{Text: m.Content}),
      }
    }
    results[i] = &schema.AgenticMessage{
      Role: schema.AgenticRoleTypeUser,
      ContentBlocks: []*schema.ContentBlock{{
        Type:               schema.ContentBlockTypeFunctionToolResult,
        FunctionToolResult: ftr,
        Extra:              m.Extra,
      }},
      Extra: m.Extra,
    }
  }
  return results
}

export async function processResponsesStream<TApi extends Api>(
  openaiStream: AsyncIterable<ResponseStreamEvent>,
  output: AssistantMessage,
  stream: AssistantMessageEventStream,
  model: Model<TApi>,
  options?: OpenAIResponsesStreamOptions,
): Promise<void> {
  let sawTerminalResponseEvent = false;
  const outputSlots = new Map<number, ResponsesOutputSlot>();
  const reasoningBlocksById = new Map<string, ThinkingContent>();
  const applyMessagePhaseStopReason = (item: ResponseOutputItem): void => {
    if (item.type === "message" && item.phase === "final_answer") {
      output.stopReason = "stop";
    }
  };
  const getSlot = <TType extends ResponsesOutputSlot["type"]>(
    outputIndex: number,
    type: TType,
  ): Extract<ResponsesOutputSlot, { type: TType }> | undefined => {
    const slot = outputSlots.get(outputIndex);
    return slot?.type === type ? (slot as Extract<ResponsesOutputSlot, { type: TType }>) : undefined;
  };
  const pushToolCallDelta = (slot: ToolCallOutputSlot, delta: string | undefined): void => {
    if (delta === undefined) return;
    stream.push({
      type: "toolcall_delta",
      contentIndex: slot.contentIndex,
      delta,
      partial: output,
    });
  };
  const createSlot = (outputIndex: number, item: ResponseOutputItem): ResponsesOutputSlot | undefined => {
    if (item.type === "reasoning") {
      const block: ThinkingContent = { type: "thinking", thinking: "" };
      output.content.push(block);
      const slot = {
        type: "thinking",
        block,
        contentIndex: output.content.length - 1,
      } satisfies ResponsesOutputSlot;
      outputSlots.set(outputIndex, slot);
      stream.push({ type: "thinking_start", contentIndex: slot.contentIndex, partial: output });
      return slot;
    }
    if (item.type === "message") {
      applyMessagePhaseStopReason(item);
      const block: TextContent = { type: "text", text: "" };
      output.content.push(block);
      const slot = { type: "text", block, contentIndex: output.content.length - 1 } satisfies ResponsesOutputSlot;
      outputSlots.set(outputIndex, slot);
      stream.push({ type: "text_start", contentIndex: slot.contentIndex, partial: output });
      return slot;
    }
    if (item.type === "function_call") {
      const block: StreamingToolCall = {
        type: "toolCall",
        id: `${item.call_id}|${item.id}`,
        name: item.name,
        arguments: {},
        ...(item.namespace !== undefined ? { namespace: item.namespace } : {}),
        partialJson: item.arguments || "",
      };
      output.content.push(block);
      const slot = {
        type: "toolCall",
        block,
        contentIndex: output.content.length - 1,
      } satisfies ResponsesOutputSlot;
      outputSlots.set(outputIndex, slot);
      stream.push({ type: "toolcall_start", contentIndex: slot.contentIndex, partial: output });
      return slot;
    }
    if (item.type === "custom_tool_call") {
      const inputProperty = options?.grammarToolInputProperties?.get(item.name) ?? "input";
      const input = item.input || "";
      const block: StreamingToolCall = {
        type: "toolCall",
        id: `${item.call_id}|${item.id}`,
        name: item.name,
        arguments: { [inputProperty]: input },
        ...(item.namespace !== undefined ? { namespace: item.namespace } : {}),
        customInput: {
          property: inputProperty,
          jsonBuffer: { input: "", started: false, closed: false },
        },
      };
      output.content.push(block);
      const slot = {
        type: "toolCall",
        block,
        contentIndex: output.content.length - 1,
      } satisfies ResponsesOutputSlot;
      outputSlots.set(outputIndex, slot);
      stream.push({ type: "toolcall_start", contentIndex: slot.contentIndex, partial: output });
      return slot;
    }
    return undefined;
  };
  const getOrCreateSlot = (outputIndex: number, item: ResponseOutputItem): ResponsesOutputSlot | undefined => {
    return outputSlots.get(outputIndex) ?? createSlot(outputIndex, item);
  };
  // Azure OpenAI can omit reasoning.encrypted_content from response.output_item.done
  // and provide it only in response.completed.response.output. Backfill the
  // persisted reasoning signature from the terminal response to keep store:false
  // multi-turn replay stateless. See https://github.com/earendil-works/pi/issues/6409.
  const backfillReasoningSignatures = (responseOutput: ResponseOutputItem[]): void => {
    for (const item of responseOutput) {
      if (item.type !== "reasoning" || !item.encrypted_content) continue;
      const block = reasoningBlocksById.get(item.id);
      if (!block?.thinkingSignature) continue;

      const storedItem = JSON.parse(block.thinkingSignature) as ResponseReasoningItem;
      if (storedItem.encrypted_content) continue;
      block.thinkingSignature = JSON.stringify({
        ...storedItem,
        encrypted_content: item.encrypted_content,
      });
    }
  };
  const finalizeResponse = (
    response: Extract<ResponseStreamEvent, { type: "response.completed" | "response.incomplete" }>["response"],
  ): void => {
    sawTerminalResponseEvent = true;
    backfillReasoningSignatures(response.output ?? []);
    if (response?.id) {
      output.responseId = response.id;
    }
    if (response?.usage) {
      const inputDetails = response.usage.input_tokens_details as
        | { cached_tokens?: number; cache_write_tokens?: number }
        | undefined;
      const cachedTokens = inputDetails?.cached_tokens || 0;
      const cacheWriteTokens = inputDetails?.cache_write_tokens || 0;
      output.usage = {
        // OpenAI includes cached and cache-write tokens in input_tokens, so subtract both.
        input: Math.max(0, (response.usage.input_tokens || 0) - cachedTokens - cacheWriteTokens),
        output: response.usage.output_tokens || 0,
        cacheRead: cachedTokens,
        cacheWrite: cacheWriteTokens,
        reasoning: response.usage.output_tokens_details?.reasoning_tokens || 0,
        totalTokens: response.usage.total_tokens || 0,
        cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
      };
    }
    calculateCost(model, output.usage);
    if (options?.applyServiceTierPricing) {
      const serviceTier = options.resolveServiceTier
        ? options.resolveServiceTier(response?.service_tier, options.serviceTier)
        : (response?.service_tier ?? options.serviceTier);
      options.applyServiceTierPricing(output.usage, serviceTier);
    }
    // Map status to stop reason. For incomplete responses, retain the provider's
    // specific reason so max-output truncation and content filtering stay distinct.
    const status = response?.status;
    const incompleteDetails = response?.incomplete_details as { reason?: unknown } | null | undefined;
    const incompleteReason = typeof incompleteDetails?.reason === "string" ? incompleteDetails.reason : undefined;
    output.rawStopReason = incompleteReason ? `${status}.${incompleteReason}` : status;
    const mappedStop = mapStopReason(status, incompleteReason);
    output.stopReason = mappedStop.stopReason;
    if (mappedStop.errorMessage === undefined) delete output.errorMessage;
    else output.errorMessage = mappedStop.errorMessage;
    if (output.content.some((b) => b.type === "toolCall") && output.stopReason === "stop") {
      output.stopReason = "toolUse";
    }
  };

  for await (const event of openaiStream) {
    await options?.onProviderStreamEvent?.(event, model);
    if (event.type === "response.created") {
      output.responseId = event.response.id;
    } else if (event.type === "response.output_item.added") {
      createSlot(event.output_index, event.item);
    } else if (event.type === "response.reasoning_summary_text.delta") {
      const slot = getSlot(event.output_index, "thinking");
      if (!slot) continue;
      slot.block.thinking += event.delta;
      stream.push({
        type: "thinking_delta",
        contentIndex: slot.contentIndex,
        delta: event.delta,
        partial: output,
      });
    } else if (event.type === "response.reasoning_summary_part.done") {
      const slot = getSlot(event.output_index, "thinking");
      if (!slot) continue;
      slot.block.thinking += "\n\n";
      stream.push({
        type: "thinking_delta",
        contentIndex: slot.contentIndex,
        delta: "\n\n",
        partial: output,
      });
    } else if (event.type === "response.reasoning_text.delta") {
      const slot = getSlot(event.output_index, "thinking");
      if (!slot) continue;
      slot.block.thinking += event.delta;
      stream.push({
        type: "thinking_delta",
        contentIndex: slot.contentIndex,
        delta: event.delta,
        partial: output,
      });
    } else if (event.type === "response.output_text.delta") {
      const slot = getSlot(event.output_index, "text");
      if (!slot) continue;
      slot.block.text += event.delta;
      stream.push({
        type: "text_delta",
        contentIndex: slot.contentIndex,
        delta: event.delta,
        partial: output,
      });
    } else if (event.type === "response.refusal.delta") {
      const slot = getSlot(event.output_index, "text");
      if (!slot) continue;
      slot.block.text += event.delta;
      stream.push({
        type: "text_delta",
        contentIndex: slot.contentIndex,
        delta: event.delta,
        partial: output,
      });
    } else if (event.type === "response.function_call_arguments.delta") {
      const slot = getSlot(event.output_index, "toolCall");
      if (!slot || slot.block.partialJson === undefined) continue;
      slot.block.partialJson += event.delta;
      slot.block.arguments = parseStreamingJson(slot.block.partialJson);
      pushToolCallDelta(slot, event.delta);
    } else if (event.type === "response.function_call_arguments.done") {
      const slot = getSlot(event.output_index, "toolCall");
      if (!slot || slot.block.partialJson === undefined) continue;
      const previousPartialJson = slot.block.partialJson;
      slot.block.partialJson = event.arguments;
      slot.block.arguments = parseStreamingJson(slot.block.partialJson);

      if (event.arguments.startsWith(previousPartialJson)) {
        const delta = event.arguments.slice(previousPartialJson.length);
        if (delta.length > 0) pushToolCallDelta(slot, delta);
      }
    } else if (event.type === "response.custom_tool_call_input.delta") {
      const slot = getSlot(event.output_index, "toolCall");
      if (!slot || !slot.block.customInput) continue;
      pushToolCallDelta(
        slot,
        appendCustomToolCallInput(slot.block, getCustomToolCallInput(slot.block) + event.delta, false),
      );
    } else if (event.type === "response.custom_tool_call_input.done") {
      const slot = getSlot(event.output_index, "toolCall");
      if (!slot || !slot.block.customInput) continue;
      pushToolCallDelta(slot, appendCustomToolCallInput(slot.block, event.input, true));
    } else if (event.type === "response.output_item.done") {
      const item = event.item;
      applyMessagePhaseStopReason(item);
      const slot = getOrCreateSlot(event.output_index, item);

      if (item.type === "reasoning" && slot?.type === "thinking") {
        const summaryText = item.summary?.map((s) => s.text).join("\n\n") || "";
        const contentText = item.content?.map((c) => c.text).join("\n\n") || "";
        slot.block.thinking = summaryText || contentText || slot.block.thinking;
        slot.block.thinkingSignature = JSON.stringify(item);
        reasoningBlocksById.set(item.id, slot.block);
        stream.push({
          type: "thinking_end",
          contentIndex: slot.contentIndex,
          content: slot.block.thinking,
          partial: output,
        });
        outputSlots.delete(event.output_index);
      } else if (item.type === "message" && slot?.type === "text") {
        slot.block.text = item.content?.map((c) => (c.type === "output_text" ? c.text : c.refusal)).join("") || "";
        slot.block.textSignature = encodeTextSignatureV1(item.id, item.phase ?? undefined);
        stream.push({
          type: "text_end",
          contentIndex: slot.contentIndex,
          content: slot.block.text,
          partial: output,
        });
        outputSlots.delete(event.output_index);
      } else if (
        item.type === "function_call" &&
        slot?.type === "toolCall" &&
        slot.block.partialJson !== undefined
      ) {
        slot.block.arguments = parseStreamingJson(item.arguments || slot.block.partialJson || "{}");
        if (item.namespace !== undefined) slot.block.namespace = item.namespace;
        // Finalize in-place and strip the scratch buffer so replay only
        // carries parsed arguments.
        delete slot.block.partialJson;
        stream.push({
          type: "toolcall_end",
          contentIndex: slot.contentIndex,
          toolCall: slot.block,
          partial: output,
        });
        outputSlots.delete(event.output_index);
      } else if (item.type === "custom_tool_call" && slot?.type === "toolCall" && slot.block.customInput) {
        pushToolCallDelta(
          slot,
          appendCustomToolCallInput(slot.block, item.input ?? getCustomToolCallInput(slot.block), true),
        );
        if (item.namespace !== undefined) slot.block.namespace = item.namespace;
        delete slot.block.customInput;
        stream.push({
          type: "toolcall_end",
          contentIndex: slot.contentIndex,
          toolCall: slot.block,
          partial: output,
        });
        outputSlots.delete(event.output_index);
      }
    } else if (event.type === "response.completed" || event.type === "response.incomplete") {
      finalizeResponse(event.response);
    } else if (event.type === "error") {
      throw new Error(`Error Code ${event.code}: ${event.message}` || "Unknown error");
    } else if (event.type === "response.failed") {
      sawTerminalResponseEvent = true;
      output.rawStopReason = event.response?.status;
      const error = event.response?.error;
      const details = event.response?.incomplete_details;
      const msg = error
        ? `${error.code || "unknown"}: ${error.message || "no message"}`
        : details?.reason
          ? `incomplete: ${details.reason}`
          : "Unknown error (no error details in response)";
      throw new Error(msg);
    }
  }
  if (!sawTerminalResponseEvent) {
    throw new Error("OpenAI Responses stream ended before a terminal response event");
  }
  // The agent runs every tool call in the final message. Refuse to hand over calls whose
  // output_item.done never arrived: their arguments may be cut off or mixed up, e.g. when a
  // non-compliant server omits output_index. Finished calls have their scratch buffers removed.
  if (output.stopReason === "toolUse") {
    for (const block of output.content) {
      if (block.type !== "toolCall") continue;
      const toolCall = block as StreamingToolCall;
      if (toolCall.partialJson !== undefined || toolCall.customInput !== undefined) {
        throw new Error(
          `OpenAI Responses stream completed with an unfinished tool call: ${toolCall.name} (${toolCall.id})`,
        );
      }
    }
  }
}

async function runLoop(
  initialContext: AgentContext,
  newMessages: AgentMessage[],
  initialConfig: AgentLoopConfig,
  signal: AbortSignal | undefined,
  emit: AgentEventSink,
  streamFunction: StreamFn,
): Promise<void> {
  let currentContext = initialContext;
  let config = initialConfig;
  let lastCompletedTurn: PrepareNextTurnContext | undefined;
  let explicitContinuation = false;
  // Check for steering messages at start (user may have typed while waiting)
  let pendingMessages: AgentMessage[] = (await config.getSteeringMessages?.()) || [];

  // Outer loop: continues when queued follow-up messages arrive after agent would stop
  while (true) {
    let hasMoreToolCalls = true;

    // Inner loop: process tool calls and steering messages
    while (hasMoreToolCalls || pendingMessages.length > 0) {
      let preparedMessages: AgentMessage[] = [];
      if (lastCompletedTurn) {
        const nextTurnSnapshot = await config.prepareNextTurn?.(lastCompletedTurn);
        if (nextTurnSnapshot) {
          currentContext = nextTurnSnapshot.context ?? currentContext;
          preparedMessages = nextTurnSnapshot.messages ?? [];
          config = {
            ...config,
            model: nextTurnSnapshot.model ?? config.model,
            reasoning:
              nextTurnSnapshot.thinkingLevel === undefined
                ? config.reasoning
                : nextTurnSnapshot.thinkingLevel === "off"
                  ? undefined
                  : nextTurnSnapshot.thinkingLevel,
          };
        }
        // Preparation can be long-running (for example, compaction). Pick up steering
        // queued while it ran. Only poll again if the earlier poll returned nothing;
        // otherwise one-at-a-time mode would deliver two messages in this turn.
        if (pendingMessages.length === 0) {
          pendingMessages = (await config.getSteeringMessages?.()) || [];
        }
        await emit({ type: "turn_start" });
      }

      // Process prepared and queued messages before the next assistant response.
      for (const message of declareToolChanges(currentContext, [...preparedMessages, ...pendingMessages])) {
        await emit({ type: "message_start", message });
        await emit({ type: "message_end", message });
        currentContext.messages.push(message);
        newMessages.push(message);
      }
      pendingMessages = [];

      const requestUpdate = await config.prepareRequest?.(
        {
          context: currentContext,
          model: config.model,
          thinkingLevel: config.reasoning ?? "off",
        },
        signal,
      );
      if (requestUpdate) {
        currentContext = requestUpdate.context ?? currentContext;
        config = {
          ...config,
          model: requestUpdate.model ?? config.model,
          reasoning:
            requestUpdate.thinkingLevel === undefined
              ? config.reasoning
              : requestUpdate.thinkingLevel === "off"
                ? undefined
                : requestUpdate.thinkingLevel,
        };
      }

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

      if (message.stopReason === "error" || message.stopReason === "aborted") {
        lastCompletedTurn = {
          message,
          toolResults: [],
          context: currentContext,
          newMessages,
        };
        await config.finishTurn?.(lastCompletedTurn, signal);
        await emit({ type: "turn_end", message, toolResults: [] });
        await emit({ type: "agent_end", messages: newMessages });
        return;
      }

      // Check for tool calls
      const toolCalls = message.content.filter((c) => c.type === "toolCall");

      const toolResults: ToolResultMessage[] = [];
      hasMoreToolCalls = false;
      if (toolCalls.length > 0) {
        // A "length" stop means the output was cut off by the token limit, so
        // every tool call in the message may carry truncated arguments. Fail
        // them all instead of executing potentially borked calls.
        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);
        }
      }

      lastCompletedTurn = {
        message,
        toolResults,
        context: currentContext,
        newMessages,
      };
      const decision = await config.finishTurn?.(lastCompletedTurn, signal);
      await emit({ type: "turn_end", message, toolResults });

      if (decision?.action === "end") {
        await emit({ type: "agent_end", messages: newMessages });
        return;
      }

      explicitContinuation = decision?.action === "continue";
      pendingMessages = (await config.getSteeringMessages?.()) || [];
      if (hasMoreToolCalls || pendingMessages.length > 0) {
        explicitContinuation = false;
      }
    }

    // Agent would stop here. Check for follow-up messages.
    const followUpMessages = (await config.getFollowUpMessages?.()) || [];
    if (followUpMessages.length > 0) {
      // Set as pending so inner loop processes them
      explicitContinuation = false;
      pendingMessages = followUpMessages;
      continue;
    }

    // No natural request was selected, so fulfill the continuation decision with one context-only turn.
    if (explicitContinuation) {
      explicitContinuation = false;
      continue;
    }

    // No more messages, exit
    break;
  }

  await emit({ type: "agent_end", messages: newMessages });
}