Turn Stream and Tools 深度解析

20import type { ResponseInputItem, ResponseStreamEvent } from 'openai/resources/responses/responses'\n21import { env } from '../env.js'\n22import { redis } from '../redis.js'\n23import { classifyInboxTriage, gateSyntheticWake } from './inbox-triage.js'\n24import { GLANCE_YIELD_RULES } from './glance-protocol.js'\n25import { TOOL_DEFS_RESPONSES, executePodTool } from './runtime/pod-tools.js'\n26import {\n27  TURN_STATUS_VALUES,\n28  bashOutputHasReplySideEffect,\n29  bashOutputSideEffects,\n30  bashOutputSideEffectChannelUnreliable,\n31  type CliSideEffect,\n32  type ToolResult,\n33  type TurnStatusOutput,\n34  type TurnStatusValue,\n35} from './tools-shared.js'\n36import { hydrate as hydrateFs, commit as commitFs, teardown as teardownFs, type FsNamespace } from './runtime/fs-namespace.js'\n37import type { PollWakeBrief } from './runtime/wake-bus.js'\n38import { runtime } from './runtime/select.js'\n39import type { AgentRuntimeClient } from './runtime/client.js'\n40import { BUSY_STATUS_HEARTBEAT_MS } from '../status.js'\n41import { errorText, type AgentRunStatus } from './observability.js'\n42import { getLlmClient } from '../llm.js'\n43import { enforceModelPolicy, realTaskModel, supportModel } from './model-policy.js'\n44import { materializeImage } from './image-fetcher.js'\n45import {\n46  applyResponseStreamEvent,\n47  consumeResponseStream,\n48  newResponseStreamState,\n49  type ResponseStreamState,\n50} from './turn-stream.js'\n51import { compactHistoryWithSummary, estimateHistoryTokens, estimateTokens } from './turn-compaction.js'\n52import { usageFromOpenAI, addUsage, EMPTY_USAGE, type TokenUsage } from './cost.js'\n53import { recordLlmCall, classifyLlmCallError, readStreamUsage, readStreamReasoningTokens } from './llm-ledger.js'\n54import { resolveDeclaredAutoRelayTarget } from './auto-relay.js'\n55import {\n56  canDrainSteer,\n57  drainSteer,\n58  resetSteerForAgent,\n59  isSteerSaturated,\n60  isSteerByteBudgetExhausted,\n61  recordSteerBytes,\n62  registerActiveToolBatch,\n63  clearActiveToolBatch,\n64  SUMMARIZE_THRESHOLD as STEER_SUMMARIZE_THRESHOLD,\n65  type SteerItem,\n66} from './steer.js'

29export interface PendingToolCall {\n30  call_id: string\n31  name: string\n32  arguments: string\n33}\n34\n35export interface ResponseStreamState {\n36  pendingTools: Record<string, PendingToolCall>\n37  responseTextByPart: Map<string, string>\n38  responseId?: string\n39  responseStatus?: string\n40  /** Traced (truncation-safe) representation of `response.completed.output`,\n41   *  rendered eagerly via the optional `traceResponseOutputItem` callback\n42   *  so the observability event can stay small even when the underlying\n43   *  response carries giant tool outputs. */\n44  responseOutput: unknown[]\n45  responseUsage: unknown\n46  /** Sum of `usage.input_tokens + usage.output_tokens` from the\n47   *  `response.completed` event. The caller surfaces this to the per-turn\n48   *  75%-budget guard; reset every hop. */\n49  totalTokens: number\n50}\n51\n52export function newResponseStreamState(): ResponseStreamState {\n53  return {\n54    pendingTools: {},\n55    responseTextByPart: new Map<string, string>(),\n56    responseOutput: [],\n57    responseUsage: null,\n58    totalTokens: 0,\n59  }\n60}\n61\n62export interface ApplyStreamEventOptions {\n63  /** Optional mapper for `response.completed.output` items. The production\n64   *  caller passes `traceResponseOutputItem` to render each item into a\n65   *  truncation-safe shape for observability events. Tests can omit this\n66   *  to keep raw items, which makes assertions easier. */\n67  traceItem?: (item: unknown) => unknown\n68}\n69\n70export interface ConsumeStreamOptions {\n71  /** Maximum silence between stream events. Once the HTTP stream has been\n72   *  established, the OpenAI SDK request timeout no longer protects every\n73   *  `for await` read; this bounds the no-progress case explicitly. */\n74  idleTimeoutMs?: number\n75  /** Maximum wall-clock time for one model hop stream, even if it dribbles\n76   *  occasional events. Agent tasks should continue in another hop, not pin\n77   *  a worker forever inside one response body. */\n78  wallTimeoutMs?: number\n79}\n80\n81function abortStream(stream: AsyncIterable<ResponseStreamEvent>): void {\n82  const maybeAbort = (stream as unknown as { controller?: { abort?: () => void } }).controller?.abort\n83  if (typeof maybeAbort === 'function') {\n84    try {\n85      maybeAbort.call((stream as unknown as { controller?: unknown }).controller)\n86    } catch {\n87      // Best effort only. The caller is already failing the turn.\n88    }\n89  }\n90}

110export async function consumeResponseStream(\n111  stream: AsyncIterable<ResponseStreamEvent>,\n112  onEvent: (event: ResponseStreamEvent) => void,\n113  opts: ConsumeStreamOptions = {},\n114): Promise<void> {\n115  const idleTimeoutMs = opts.idleTimeoutMs ?? RESPONSE_STREAM_IDLE_TIMEOUT_MS\n116  const wallTimeoutMs = opts.wallTimeoutMs ?? RESPONSE_STREAM_WALL_TIMEOUT_MS\n117  const deadline = Date.now() + wallTimeoutMs\n118  const iterator = stream[Symbol.asyncIterator]()\n119\n120  while (true) {\n121    const remainingWallMs = deadline - Date.now()\n122    if (remainingWallMs <= 0) {\n123      abortStream(stream)\n124      throw new ResponseStreamTimeoutError('wall', wallTimeoutMs)\n125    }\n126    const waitMs = Math.min(idleTimeoutMs, remainingWallMs)\n127    const timeoutKind = waitMs === remainingWallMs ? 'wall' : 'idle'\n128    let result: IteratorResult<ResponseStreamEvent>\n129    try {\n130      result = await nextWithTimeout(iterator.next(), waitMs, timeoutKind)\n131    } catch (err) {\n132      if (err instanceof ResponseStreamTimeoutError) abortStream(stream)\n133      throw err\n134    }\n135    if (result.done) return\n136    onEvent(result.value)\n137  }\n138}

43/** Dispatch a single tool call. Logs to tool_calls table. */\n44export async function executeTool(args: {\n45  agentId: string\n46  name: string\n47  argsJson: string\n48  messageId?: string | null\n49  runId?: string | null\n50  companyId?: string | null\n51  /** Per-turn FS namespace. When provided, bash runs with cwd\n52   *  = ns.rootDir and the three native FS tools operate on it.\n53   *  Optional so other call sites (rehire, manual replay) can keep\n54   *  running without a namespace. */\n55  ns?: FsNamespace | null\n56}): Promise<ToolResult> {\n57  const t0 = Date.now()\n58  const id = `t-${randomUUID()}`\n59  let parsed: Record<string, unknown> = {}\n60  try {\n61    parsed = JSON.parse(args.argsJson || '{}')\n62  } catch {\n63    parsed = {}\n64  }\n65\n66  await pool.query(\n67    `INSERT INTO tool_calls (id, message_id, agent_id, name, args, status, run_id, company_id)\n68     VALUES ($1,$2,$3,$4,$5::jsonb,'pending',$6,$7)`,\n69    [id, args.messageId ?? null, args.agentId, args.name, JSON.stringify(parsed), args.runId ?? null, args.companyId ?? null],\n70  )\n71\n72  let result: ToolResult\n73  try {\n74    switch (args.name) {\n75      case 'palette':            result = await tPalette(parsed, args.companyId ?? null, args.agentId); break\n76      case 'dm_with':            result = await tDmWith(parsed, args.agentId); break\n77      case 'pull_group':         result = await tPullGroup(parsed, args.agentId); break\n78      case 'set_turn_status':    result = tSetTurnStatus(parsed); break\n79      case 'shell':\n80      case 'bash':               result = await tBash(parsed, args.agentId, args.ns ?? null); break\n81      case 'react':              result = await tReact(parsed, args.agentId); break\n82      // Native FS tools — require an active per-turn namespace. Reject\n83      // gracefully if called outside a turn (shouldn't happen via the\n84      // normal LLM path; defensive).\n85      case 'read_file':\n86        result = args.ns ? await tReadFile(parsed, args.ns) : noNamespace('read_file', args.name)\n87        break\n88      case 'write_file':\n89        result = args.ns ? await tWriteFile(parsed, args.ns) : noNamespace('write_file', args.name)\n90        break\n91      case 'edit_file':\n92        result = args.ns ? await tEditFile(parsed, args.ns) : noNamespace('edit_file', args.name)\n93        break\n94      default:\n95        result = {\n96          ok: false, output: null, error: 'unknown tool',\n97          durationMs: Date.now() - t0,\n98          display: { name: args.name, arg: '', status: 'unknown tool', detail: `tool not implemented: ${args.name}` },\n99        }\n100    }\n101  } catch (err) {\n102    result = {\n103      ok: false, output: null, error: String(err), durationMs: Date.now() - t0,\n104      display: { name: args.name, arg: JSON.stringify(parsed).slice(0, 80), status: 'error', detail: String(err) },\n105    }\n106  }\n107\n108  await pool.query(\n109    `UPDATE tool_calls\n110        SET result = $2::jsonb, status = $3, error = $4, duration_ms = $5\n111      WHERE id = $1`,\n112    [id, JSON.stringify(result.output ?? null), result.ok ? 'ok' : 'error', result.error ?? null, result.durationMs],\n113  )\n114\n115  return result\n116}