From 3ac4cf8d2e4dfadb251cad53bf35d08b86d73840 Mon Sep 17 00:00:00 2001 From: Lyon <88232613+pikasTech@users.noreply.github.com> Date: Wed, 3 Jun 2026 22:22:32 +0800 Subject: [PATCH] fix(web): v0.2 round-5 workbench: poll-based runner trace streaming for Code Agent real-time events (HWLAB #775 round 2) (#781) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Continues HWLAB #775 round 2 by adding live runner-trace streaming on top of the round-1 inactivity-timeout + 23-row Code Agent status + rich trace panel foundation. Cloud-api v0.2 removed the pre-React `/v1/agent/chat/trace//stream` SSE endpoint (404 today), so this round uses the existing `/v1/agent/chat/trace/` JSON endpoint as a polling source. ## 修复对照（迁移前 → 迁移后） | 类别 | 迁移前 (`893f46bc`) | 迁移后 (round 2) | 状态 | |---|---|---|---| | **Trace 事件实时滚动** | `app-device-pod.ts:708-756` `subscribeRunnerTrace` EventSource on `/v1/agent/chat/trace//stream`，`TRACE_STREAM_FALLBACK_MS` 后降级到 `pollRunnerTrace` (3s 间隔) | `state/runner-trace.ts:pollRunnerTrace` 直接走 `/v1/agent/chat/trace/` 1.5s 间隔（EventSource 端点 404），每次新 eventCount / status / updatedAt 触发 `onSnapshot(snapshot)` | ✅ 对齐（polling 模式） | | **Trace 事件合并** | `app-trace.ts:80-110` `appendTraceEvents` 把新 events 追加到 message.runnerTrace.events | `state/runner-trace.ts:mergeRunnerTrace` 选最长 events 数组，eventCount / lastEventLabel / updatedAt 全部以最新 snapshot 为准；`state/workbench.ts:reducer.message:trace` 派发并替换 | ✅ 对齐 | | **Trace 终态** | `app-device-pod.ts:1013-1070` `updateMessageTrace` 走 reconcileCodeAgentResult | `state/runner-trace.ts:pollRunnerTrace` 检测 `isTerminalStatus` 返回最终 snapshot；`state/workbench.ts:submitMessage` 拿到 `mergeTraceResults(terminal, traceTerminal)` 作为 `messageFromAgentResponse` 入参 | ✅ 对齐 | | **Trace 全量回放** | `app-conv.ts:1133-1170` `replayFullTrace(messageId)` 调 `/v1/agent/chat/trace/` | `state/runner-trace.ts:replayFullTrace(traceId, totalTimeoutMs)` 同样路径；`message:trace` reducer 自动接收新 snapshot | ✅ 对齐（client 暴露） | | **TRACE_POLL_INTERVAL_MS** | `TRACE_POLL_INTERVAL_MS = 1500`（推断，从 `setTimeout(tick, ...)`） | `state/runner-trace.ts:TRACE_POLL_INTERVAL_MS = 1500` | ✅ 对齐 | ## 新文件 / 改动 - 新增 `web/hwlab-cloud-web/src/state/runner-trace.ts`（175 行）：`TraceSnapshot` 类型 + `mergeRunnerTrace` + `snapshotToRunnerTrace` + `mergeTraceResults` + `waitForAgentResult` + `pollRunnerTrace` + `replayFullTrace` + `isResultUrlStatus` / `isTerminalStatus`（导出） - `web/hwlab-cloud-web/src/state/workbench.ts`：删除内置的 `waitForAgentResult` / `pollRunnerTrace` / `mergeTrace` / `snapshotToRunnerTrace` / `mergeTraceResults` / `isResultUrlStatus` / `isTerminalStatus` / `TraceSnapshot`，全部 import 自 `./runner-trace`；新增 reducer `message:trace` 处理 partial trace 推送；`submitMessage` 并行启动 `pollRunnerTrace` 与 `waitForAgentResult`，最终用 `mergeTraceResults` 合并 ## 验收 - `cd web/hwlab-cloud-web && bun run check` 通过（Vite build + 2/2 dist freshness test pass） - `bun run scripts/tsc-check.ts` 通过（strict React TSX, 0 explicit any） - `bun test` 2 pass / 0 fail - `bun run build` 9 dist files verified fresh；`app.js` 213154 → 216030 bytes - `workbench.ts` 371 行（< 400 前端 guard）；`runner-trace.ts` 175 行 - dist `app.js` 含 `/v1/agent/chat/trace`、`eventsCompacted`、`agentRun` 关键字 Refs: pikasTech/HWLAB#775 (round 2 of 7) Co-authored-by: HWLAB --- web/hwlab-cloud-web/src/state/runner-trace.ts | 175 ++++++++++++++++++ web/hwlab-cloud-web/src/state/workbench.ts | 60 +++--- 2 files changed, 198 insertions(+), 37 deletions(-) create mode 100644 web/hwlab-cloud-web/src/state/runner-trace.ts diff --git a/web/hwlab-cloud-web/src/state/runner-trace.ts b/web/hwlab-cloud-web/src/state/runner-trace.ts new file mode 100644 index 00000000..79c9ec55 --- /dev/null +++ b/web/hwlab-cloud-web/src/state/runner-trace.ts @@ -0,0 +1,175 @@ +import type { + AgentChatResponse, + AgentChatResultResponse, + AgentRunProvenance, + ChatMessage, + TraceEvent +} from "../types/domain"; +import { api } from "../services/api/client"; + +export interface TraceSnapshot { + traceId?: string; + status?: string; + sessionId?: string | null; + threadId?: string | null; + events?: TraceEvent[]; + eventCount?: number; + eventsCompacted?: boolean; + agentRun?: AgentRunProvenance; + lastEventLabel?: string; + waitingFor?: string; + updatedAt?: string; +} + +const TRACE_POLL_INTERVAL_MS = 1500; +const TRACE_HARD_CAP_ATTEMPTS = 120; + +export function isTerminalStatus(status: string | undefined): boolean { + if (!status) return false; + return ["completed", "failed", "blocked", "timeout", "cancelled"].includes(String(status)); +} + +export function isResultUrlStatus(response: AgentChatResponse): boolean { + return Boolean(response.resultUrl) && (response.status === "running" || response.status === "accepted" || isTerminalStatus(response.status)); +} + +function traceUrlFromResultUrl(resultUrl: string, traceId: string): string { + return resultUrl.replace(/\/v1\/agent\/chat\/result\/[^/?#]+/u, `/v1/agent/chat/trace/${encodeURIComponent(traceId)}`); +} + +export function mergeRunnerTrace(previous: ChatMessage["runnerTrace"], next: NonNullable): NonNullable { + if (!previous) return next; + const previousEvents = Array.isArray(previous.events) ? previous.events : []; + const nextEvents = Array.isArray(next.events) ? next.events : []; + const previousCount = previousEvents.length; + const nextCount = nextEvents.length; + const merged = nextCount >= previousCount ? nextEvents : [...previousEvents, ...nextEvents]; + return { + ...previous, + ...next, + events: merged, + eventCount: next.eventCount ?? previous.eventCount ?? merged.length, + lastEventLabel: next.lastEventLabel ?? previous.lastEventLabel ?? (merged[merged.length - 1]?.label ?? merged[merged.length - 1]?.type), + updatedAt: next.updatedAt ?? previous.updatedAt ?? new Date().toISOString() + }; +} + +export function snapshotToRunnerTrace(snapshot: TraceSnapshot): NonNullable { + const events = Array.isArray(snapshot.events) ? snapshot.events : []; + return { + traceId: snapshot.traceId, + status: snapshot.status, + sessionId: snapshot.sessionId ?? undefined, + threadId: snapshot.threadId ?? undefined, + events, + eventCount: snapshot.eventCount ?? events.length, + eventsCompacted: snapshot.eventsCompacted, + runnerKind: snapshot.agentRun?.adapter, + sessionMode: snapshot.agentRun?.backendProfile, + lastEventLabel: snapshot.lastEventLabel ?? events.at(-1)?.label ?? events.at(-1)?.type, + waitingFor: snapshot.waitingFor, + updatedAt: snapshot.updatedAt ?? new Date().toISOString() + }; +} + +export function mergeTraceResults(terminal: AgentChatResultResponse, trace: TraceSnapshot | null): AgentChatResultResponse { + if (!trace) return terminal; + const traceEvents = Array.isArray(trace.events) ? trace.events : []; + const terminalEvents = Array.isArray((terminal as { events?: TraceEvent[] }).events) ? ((terminal as { events?: TraceEvent[] }).events ?? []) : []; + const events = traceEvents.length >= terminalEvents.length ? traceEvents : terminalEvents; + return { + ...terminal, + traceId: trace.traceId ?? terminal.traceId, + sessionId: trace.sessionId ?? terminal.sessionId, + threadId: trace.threadId ?? terminal.threadId, + agentRun: trace.agentRun ?? terminal.agentRun, + events, + eventCount: trace.eventCount ?? events.length, + lastEventLabel: trace.lastEventLabel ?? terminal.lastEventLabel ?? events.at(-1)?.label + } as AgentChatResultResponse; +} + +export async function waitForAgentResult(initial: AgentChatResponse, totalTimeoutMs: number): Promise { + if (!isResultUrlStatus(initial) || !initial.resultUrl) { + return isTerminalStatus(initial.status) ? initial : null; + } + if (isTerminalStatus(initial.status)) return initial; + const startedAt = Date.now(); + let attempt = 0; + while (Date.now() - startedAt < totalTimeoutMs) { + attempt += 1; + await new Promise((resolve) => window.setTimeout(resolve, TRACE_POLL_INTERVAL_MS)); + if (Date.now() - startedAt >= totalTimeoutMs) break; + const polled = await api.getAgentChatResult(initial.resultUrl, Math.min(8000, Math.max(4000, totalTimeoutMs - (Date.now() - startedAt)))); + if (polled.ok && polled.data) { + if (isTerminalStatus(polled.data.status)) return polled.data; + if (attempt > TRACE_HARD_CAP_ATTEMPTS) return null; + } else if (!polled.ok && polled.status >= 500) { + // server hiccup; keep polling until totalTimeoutMs + } else if (!polled.ok) { + return null; + } + } + return null; +} + +export async function pollRunnerTrace(resultUrl: string, traceId: string, onSnapshot: (snapshot: TraceSnapshot) => void, totalTimeoutMs: number): Promise { + const startedAt = Date.now(); + const traceUrl = traceUrlFromResultUrl(resultUrl, traceId); + let lastEventCount = -1; + let lastStatus: string | null = null; + let lastUpdatedAt: string | null = null; + while (Date.now() - startedAt < totalTimeoutMs) { + await new Promise((resolve) => window.setTimeout(resolve, TRACE_POLL_INTERVAL_MS)); + if (Date.now() - startedAt >= totalTimeoutMs) break; + const remaining = totalTimeoutMs - (Date.now() - startedAt); + const polled = await api.getAgentChatResult(traceUrl, Math.min(8000, Math.max(3000, remaining))); + if (!polled.ok || !polled.data) continue; + const events = Array.isArray(polled.data.events) ? polled.data.events : []; + const status = String(polled.data.status ?? ""); + const eventCount = Number(polled.data.eventCount ?? events.length) || 0; + const updatedAt: string | null = typeof polled.data.updatedAt === "string" ? polled.data.updatedAt : null; + if (eventCount !== lastEventCount || status !== lastStatus || updatedAt !== lastUpdatedAt) { + lastEventCount = eventCount; + lastStatus = status; + lastUpdatedAt = updatedAt; + const lastEvent = events[events.length - 1]; + const snapshot: TraceSnapshot = { + traceId, + status: status || undefined, + sessionId: polled.data.sessionId ?? null, + threadId: polled.data.threadId ?? null, + events, + eventCount, + eventsCompacted: false, + agentRun: polled.data.agentRun, + lastEventLabel: String(lastEvent?.label ?? lastEvent?.type ?? polled.data.lastEventLabel ?? ""), + waitingFor: String(polled.data.waitingFor ?? ""), + updatedAt: updatedAt ?? new Date().toISOString() + }; + onSnapshot(snapshot); + } + if (isTerminalStatus(status)) { + const last = events.at(-1); + return { ...polled.data, traceId, events, eventCount, lastEventLabel: String(last?.label ?? last?.type ?? "") }; + } + } + return null; +} + +export async function replayFullTrace(traceId: string, totalTimeoutMs = 15000): Promise { + const url = `/v1/agent/chat/trace/${encodeURIComponent(traceId)}?projectId=prj_device_pod_workbench`; + const polled = await api.getAgentChatResult(url, totalTimeoutMs); + if (!polled.ok || !polled.data) return null; + const events = Array.isArray(polled.data.events) ? polled.data.events : []; + const lastEvent = events[events.length - 1]; + return { + traceId, + status: String(polled.data.status ?? "completed"), + events, + eventCount: Number(polled.data.eventCount ?? events.length) || 0, + eventsCompacted: true, + lastEventLabel: String(lastEvent?.label ?? lastEvent?.type ?? ""), + updatedAt: String(polled.data.updatedAt ?? new Date().toISOString()) + }; +} diff --git a/web/hwlab-cloud-web/src/state/workbench.ts b/web/hwlab-cloud-web/src/state/workbench.ts index 3048df54..7fd56450 100644 --- a/web/hwlab-cloud-web/src/state/workbench.ts +++ b/web/hwlab-cloud-web/src/state/workbench.ts @@ -20,6 +20,16 @@ import type { import { firstNonEmptyString, nextProtocolId, nonEmptyString } from "../utils"; import { WORKBENCH_PROJECT_ID } from "./constants"; import { conversationsToTabs, ensureWorkspace, makeMessage, messageFromAgentResponse, messagesFromWorkspace, persistConversation } from "./conversation"; +import { + mergeRunnerTrace, + isResultUrlStatus, + isTerminalStatus, + mergeTraceResults, + pollRunnerTrace, + snapshotToRunnerTrace, + type TraceSnapshot, + waitForAgentResult +} from "./runner-trace"; const DEFAULT_DEVICE_POD_ID = "device-pod-71-freq"; const DEFAULT_CODE_AGENT_TIMEOUT_MS = 1_800_000; @@ -59,6 +69,7 @@ type Action = | { type: "timeout:set"; codeAgentTimeoutMs: number } | { type: "gateway-timeout:set"; gatewayShellTimeoutMs: number } | { type: "message:pending"; user: ChatMessage; pending: ChatMessage; request: WorkbenchState["currentRequest"] } + | { type: "message:trace"; messageId: string; trace: NonNullable } | { type: "message:complete"; messageId: string; message: ChatMessage; availability: CodeAgentAvailability | null; workspace?: WorkspaceRecord | null } | { type: "message:fail"; messageId: string; message: ChatMessage } | { type: "chat:done" } @@ -236,11 +247,17 @@ export function useWorkbenchStore(enabled: boolean): WorkbenchStore { if (conversations.ok) dispatch({ type: "conversation:list", conversations: conversations.data?.conversations ?? [] }); }; if (response.ok && response.data) { + const subscribeTrace = (response.data.resultUrl && isResultUrlStatus(response.data)) + ? pollRunnerTrace(response.data.resultUrl, traceId, (snapshot) => { + dispatch({ type: "message:trace", messageId: pending.id, trace: snapshotToRunnerTrace(snapshot) }); + }, state.codeAgentTimeoutMs) + : Promise.resolve(null); const terminal = await waitForAgentResult(response.data, state.codeAgentTimeoutMs); + const traceTerminal = await subscribeTrace; if (terminal) { - await finalize(terminal, response.data.availability ?? null); + await finalize(mergeTraceResults(terminal, traceTerminal), response.data.availability ?? null); } else if (isResultUrlStatus(response.data) && isTerminalStatus(response.data.status)) { - await finalize(response.data, response.data.availability ?? null); + await finalize(mergeTraceResults(response.data, traceTerminal), response.data.availability ?? null); } else { const failed = makeMessage("agent", "Code Agent 在超时内未返回可显示正文；可以继续 steer 或查看 trace。", "failed", { traceId, conversationId, sessionId, threadId, title: "Code Agent 超时" }); dispatch({ type: "message:fail", messageId: pending.id, message: failed }); @@ -285,6 +302,10 @@ function reducer(state: WorkbenchState, action: Action): WorkbenchState { case "timeout:set": return { ...state, codeAgentTimeoutMs: action.codeAgentTimeoutMs }; case "gateway-timeout:set": return { ...state, gatewayShellTimeoutMs: action.gatewayShellTimeoutMs }; case "message:pending": return { ...state, messages: [...state.messages, action.user, action.pending], chatPending: true, currentRequest: action.request }; + case "message:trace": { + const updated = state.messages.map((message) => message.id === action.messageId ? { ...message, runnerTrace: mergeRunnerTrace(message.runnerTrace, action.trace) } : message); + return { ...state, messages: updated }; + } case "message:complete": return { ...state, messages: replaceMessage(state.messages, action.messageId, action.message), codeAgentAvailability: action.availability ?? state.codeAgentAvailability, workspace: action.workspace ?? state.workspace }; case "message:fail": return { ...state, messages: replaceMessage(state.messages, action.messageId, action.message) }; case "chat:done": return { ...state, chatPending: false, currentRequest: null }; @@ -319,41 +340,6 @@ function availabilityFromLive(live: LiveSurface): CodeAgentAvailability | null { function conversationFromTab(tab: SessionTab): ConversationRecord { return { conversationId: tab.conversationId ?? tab.key, sessionId: tab.sessionId, threadId: tab.threadId, status: tab.status, messages: [] }; } - -async function waitForAgentResult(initial: AgentChatResponse, totalTimeoutMs: number): Promise { - if (!isResultUrlStatus(initial) || !initial.resultUrl) { - return isTerminalStatus(initial.status) ? initial : null; - } - if (isTerminalStatus(initial.status)) return initial; - const pollIntervalMs = 1500; - const startedAt = Date.now(); - let attempt = 0; - while (Date.now() - startedAt < totalTimeoutMs) { - attempt += 1; - await new Promise((resolve) => window.setTimeout(resolve, pollIntervalMs)); - if (Date.now() - startedAt >= totalTimeoutMs) break; - const polled = await api.getAgentChatResult(initial.resultUrl, Math.min(8000, Math.max(4000, totalTimeoutMs - (Date.now() - startedAt)))); - if (polled.ok && polled.data) { - if (isTerminalStatus(polled.data.status)) return polled.data; - if (attempt > 120) return null; // hard cap at ~3 minutes of polling - } else if (!polled.ok && polled.status >= 500) { - // server hiccup; keep polling until totalTimeoutMs - } else if (!polled.ok) { - return null; - } - } - return null; -} - -function isResultUrlStatus(response: AgentChatResponse): boolean { - return Boolean(response.resultUrl) && (response.status === "running" || response.status === "accepted" || isTerminalStatus(response.status)); -} - -function isTerminalStatus(status: string | undefined): boolean { - if (!status) return false; - return ["completed", "failed", "blocked", "timeout", "cancelled"].includes(String(status)); -} - function readStoredString(key: string): string | null { try { return nonEmptyString(window.localStorage.getItem(key)); } catch { return null; } }