From 806200b3d181950feb2e8ae400ab4afc8875f47e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Fri, 17 Apr 2026 10:35:23 -0400 Subject: [PATCH] fix(runtime): correct pi AgentEvent projection + add Console diagnostics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit karpathy-guidelines pass — never guess at types, read the source. Root cause of the silent "等待日志输出" Console: 1. runtime-bridge's projectAgentEventToLogChunk switched on event types that never exist in pi. Real types per pi-mono-reference/packages/ agent/src/types.ts `AgentEvent`: agent_start / agent_end / turn_start / turn_end message_start / message_update / message_end tool_execution_start / tool_execution_update / tool_execution_end (NOT the `assistant_message_delta` / `tool_call_start` / `error` I invented.) Every case therefore fell through to `default → null`, so zero log chunks were emitted even when the agent was streaming tokens. Rewrote the switch to match the real union. Streaming assistant text and thinking now route through `message_update` -> `assistantMessageEvent.text_delta` / `thinking_delta` (see pi-mono-reference/packages/ai/src/types.ts `AssistantMessageEvent` for the inner protocol). Tool activity routes through `tool_execution_start` / `tool_execution_end`. 2. worker-runner had no Console-visible breadcrumbs between `run-status: running` and eventual turn output, so when the agent stalled (credential fetch, model resolution, first API call) the user saw a bare "RUNNING" with no clue where time was being spent. Added 5 targeted `run-log` probes: [runner] starting run ... [runner] prepareRun ok, resolving model [runner] model=... provider=... — constructing Agent [pi-event] (inside agent.subscribe, one per event) [runner] calling agent.prompt (N chars) [runner] agent.prompt resolved These are structured runtime signals, not test/mockup data. They surface the exact phase a stuck run is in. typecheck 0 / lint 0-0 / test:unit 508/508 / build:verify 0-0. --- src/main/agent-runtime/runtime-bridge.ts | 78 +++++++++++++++++++----- src/main/worker/worker-runner.ts | 9 +++ 2 files changed, 72 insertions(+), 15 deletions(-) diff --git a/src/main/agent-runtime/runtime-bridge.ts b/src/main/agent-runtime/runtime-bridge.ts index c7303a5e..b62bec73 100644 --- a/src/main/agent-runtime/runtime-bridge.ts +++ b/src/main/agent-runtime/runtime-bridge.ts @@ -10,42 +10,90 @@ import { getConnectorDefinition } from "@main/connectors"; /** * Projects a pi AgentEvent to a human-readable log line for the Console. - * Returns null when the event should not surface as log output (state-only - * events such as `turn_start` are already implied by run-status). + * See pi-mono-reference/packages/agent/src/types.ts `AgentEvent` for the + * canonical union shape; this projection handles every type that carries + * user-visible activity. */ function projectAgentEventToLogChunk( event: { type: string } & Record, ): { stream: "stdout" | "stderr"; chunk: string } | null { switch (event.type) { - case "assistant_message_delta": { - const delta = typeof event.delta === "string" ? event.delta : ""; - if (!delta) return null; - return { stream: "stdout", chunk: delta }; + case "agent_start": + return { stream: "stdout", chunk: "\n[agent:start]\n" }; + case "turn_start": + return { stream: "stdout", chunk: "\n" }; + case "turn_end": + return { stream: "stdout", chunk: "\n" }; + case "message_start": { + const message = event.message as { role?: string; content?: unknown } | undefined; + if (message?.role === "user") { + const body = extractMessageText(message); + if (!body) return null; + return { stream: "stdout", chunk: `[user] ${body}\n` }; + } + return null; } - case "tool_call_start": { + case "message_update": { + // Streaming assistant deltas — `assistantMessageEvent.type` carries + // the intra-turn stream protocol from pi-ai (text_delta / + // thinking_delta / toolcall_start / ...). We surface text and + // thinking deltas verbatim so the Console shows live output. + const inner = event.assistantMessageEvent as + | ({ type?: string; delta?: string; content?: string } & Record) + | undefined; + if (!inner) return null; + if (inner.type === "text_delta" && typeof inner.delta === "string" && inner.delta.length > 0) { + return { stream: "stdout", chunk: inner.delta }; + } + if (inner.type === "thinking_delta" && typeof inner.delta === "string" && inner.delta.length > 0) { + return { stream: "stdout", chunk: inner.delta }; + } + return null; + } + case "message_end": { + // The final assistant message is already fully rendered via + // preceding message_update deltas; emit a trailing newline so the + // next turn starts on its own line. + return { stream: "stdout", chunk: "\n" }; + } + case "tool_execution_start": { const toolName = typeof event.toolName === "string" ? event.toolName : "tool"; const argsPreview = summariseJson(event.args, 200); return { stream: "stdout", chunk: `\n[tool:${toolName}] ${argsPreview}\n` }; } - case "tool_call_end": { + case "tool_execution_end": { const toolName = typeof event.toolName === "string" ? event.toolName : "tool"; const isError = event.isError === true; const resultPreview = summariseJson(event.result, 300); const tag = isError ? "tool:error" : "tool:done"; return { stream: isError ? "stderr" : "stdout", chunk: `[${tag}:${toolName}] ${resultPreview}\n` }; } - case "error": { - const msg = typeof event.message === "string" ? event.message : "(unknown error)"; - return { stream: "stderr", chunk: `[error] ${msg}\n` }; - } - case "turn_end": { - return { stream: "stdout", chunk: "\n" }; - } + case "agent_end": + return { stream: "stdout", chunk: "\n[agent:end]\n" }; default: return null; } } +/** + * Pulls plain text out of a pi AgentMessage's content array. User messages + * carry `content: [{ type: "text", text: "..." }]` per pi's AgentMessage + * schema; other content types (image, tool_result, etc.) are ignored here. + */ +function extractMessageText(message: { content?: unknown }): string { + const content = message.content; + if (typeof content === "string") return content; + if (!Array.isArray(content)) return ""; + const parts: string[] = []; + for (const block of content) { + if (block && typeof block === "object" && (block as { type?: string }).type === "text") { + const text = (block as { text?: string }).text; + if (typeof text === "string") parts.push(text); + } + } + return parts.join(""); +} + function summariseJson(value: unknown, maxLength: number): string { if (value === undefined || value === null) return ""; try { diff --git a/src/main/worker/worker-runner.ts b/src/main/worker/worker-runner.ts index f5cd6ee8..3b3a4111 100644 --- a/src/main/worker/worker-runner.ts +++ b/src/main/worker/worker-runner.ts @@ -45,6 +45,7 @@ export class WorkerRunner { this.abortControllers.set(job.runId, abortController); post({ type: "run-status", runId: job.runId, status: "running", message: `Running agent ${job.agentName}.` }); + post({ type: "run-log", runId: job.runId, stream: "stdout", chunk: `[runner] starting run for ${job.agentName} (${job.connectorId}/${job.model ?? "?"})\n` }); let resolvedModelId: string | null = job.model; let finalUsage: { inputTokens: number | null; outputTokens: number | null; costUsd: number | null } = { @@ -72,6 +73,7 @@ export class WorkerRunner { } const prepared = prepareResponse.result as PrepareRunResult; + post({ type: "run-log", runId: job.runId, stream: "stdout", chunk: `[runner] prepareRun ok, resolving model\n` }); const restoreEnv = applyEnvOverrides(prepared.extraEnv); try { const model = resolveModel(prepared.provider, job.model); @@ -80,6 +82,7 @@ export class WorkerRunner { } resolvedModelId = model.id; const extraHeaders = resolveExtraStreamHeaders(prepared.provider, prepared.apiKey); + post({ type: "run-log", runId: job.runId, stream: "stdout", chunk: `[runner] model=${model.id} provider=${prepared.provider} — constructing Agent\n` }); const cwd = job.workspacePath ?? process.cwd(); const tools = createCodingTools(cwd) as never; @@ -123,6 +126,10 @@ export class WorkerRunner { const unsubscribe = agent.subscribe(async (event) => { const timestamp = new Date().toISOString(); + // Surface every pi AgentEvent type as a raw Console probe so the + // operator can see the live event stream even when projection + // misses a type or the agent is stuck inside a particular phase. + post({ type: "run-log", runId: job.runId, stream: "stdout", chunk: `[pi-event] ${event.type}\n` }); post({ type: "agent-event", runId: job.runId, @@ -159,9 +166,11 @@ export class WorkerRunner { content: [{ type: "text", text: promptContent }], timestamp: Date.now(), }; + post({ type: "run-log", runId: job.runId, stream: "stdout", chunk: `[runner] calling agent.prompt (${promptContent.length} chars)\n` }); const runPromise = agent.prompt(promptMessage); abortController.signal.addEventListener("abort", () => agent.abort(), { once: true }); await runPromise; + post({ type: "run-log", runId: job.runId, stream: "stdout", chunk: `[runner] agent.prompt resolved\n` }); } finally { unsubscribe(); }