fix(web): v0.2 round-5 workbench: poll-based runner trace streaming for Code Agent real-time events (HWLAB #775 round 2) (#781)
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/<id>/stream` SSE endpoint (404 today), so this round uses the existing `/v1/agent/chat/trace/<id>` JSON endpoint as a polling source. ## 修复对照（迁移前 → 迁移后） | 类别 | 迁移前 (`893f46bc`) | 迁移后 (round 2) | 状态 | |---|---|---|---| | **Trace 事件实时滚动** | `app-device-pod.ts:708-756` `subscribeRunnerTrace` EventSource on `/v1/agent/chat/trace/<id>/stream`，`TRACE_STREAM_FALLBACK_MS` 后降级到 `pollRunnerTrace` (3s 间隔) | `state/runner-trace.ts:pollRunnerTrace` 直接走 `/v1/agent/chat/trace/<id>` 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/<id>` | `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 <ci@hwlab.local>
This commit is contained in:
Lyon
@@ -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<ChatMessage["runnerTrace"]>): NonNullable<ChatMessage["runnerTrace"]> {
|
||||
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<ChatMessage["runnerTrace"]> {
|
||||
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<AgentChatResultResponse | null> {
|
||||
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<void>((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<TraceSnapshot | null> {
|
||||
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<void>((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<TraceSnapshot | null> {
|
||||
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())
|
||||
};
|
||||
}
|
||||
@@ -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<ChatMessage["runnerTrace"]> }
|
||||
| { 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<TraceSnapshot | null>(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<AgentChatResultResponse | null> {
|
||||
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<void>((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; }
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user