From 4b4a7c78baf52e05a3c793f01638e7b88f067c77 Mon Sep 17 00:00:00 2001 From: lyon Date: Sat, 27 Jun 2026 16:11:27 +0800 Subject: [PATCH] fix(web): recover workbench in-place refresh --- .../scripts/workbench-e2e-server.ts | 7 +- web/hwlab-cloud-web/src/stores/workbench.ts | 137 +++++++++++++++--- 2 files changed, 122 insertions(+), 22 deletions(-) diff --git a/web/hwlab-cloud-web/scripts/workbench-e2e-server.ts b/web/hwlab-cloud-web/scripts/workbench-e2e-server.ts index f79ab697..92abcf94 100644 --- a/web/hwlab-cloud-web/scripts/workbench-e2e-server.ts +++ b/web/hwlab-cloud-web/scripts/workbench-e2e-server.ts @@ -109,6 +109,11 @@ const sseClients = new Set(); const server = createServer((request, response) => { void handleRequest(request, response).catch((error) => { + if (response.headersSent) { + console.error("workbench-e2e-server stream error", error); + response.destroy(); + return; + } json(response, 500, { ok: false, status: 500, error: { code: "e2e_server_error", message: error instanceof Error ? error.message : String(error) } }); }); }); @@ -2372,7 +2377,7 @@ function authPayload(): JsonRecord { return { authenticated: false, mode: "server", authMethod: null, identityAuthority: null, sessionKind: null, actor: null, user: null, capabilities: deniedCapabilities, expiresAt: null, valuesRedacted: true, error: { code: "auth_required", message: "Authentication is required", status: 401 } }; } const actor = { id: "usr_e2e_admin", username: "e2e-admin", role: "admin", status: "active" }; - return { authenticated: true, mode: "server", authMethod: "web-session", identityAuthority: "hwlab-local", sessionKind: "browser", actor, user: actor, capabilities: { web: "available", workbench: "available", codeAgent: "available", billing: "unlinked", apiKeys: "available", admin: "available" }, expiresAt: null, valuesRedacted: true }; + return { authenticated: true, mode: "server", authMethod: "web-session", identityAuthority: "hwlab-local", sessionKind: "browser", actor, user: actor, capabilities: { web: "available", workbench: "available", codeAgent: "available", billing: "unlinked", apiKeys: "available", admin: "available" }, access: { nav: { profileId: "e2e-admin", allowedIds: ["*"], valuesRedacted: true } }, expiresAt: null, valuesRedacted: true }; } function authLoginResponse(response: ServerResponse): void { diff --git a/web/hwlab-cloud-web/src/stores/workbench.ts b/web/hwlab-cloud-web/src/stores/workbench.ts index f15321a9..bf09e4ea 100644 --- a/web/hwlab-cloud-web/src/stores/workbench.ts +++ b/web/hwlab-cloud-web/src/stores/workbench.ts @@ -35,6 +35,9 @@ const WORKBENCH_TRACE_EVENTS_TIMEOUT_MS = 5_000; const SESSION_LIST_REALTIME_REFRESH_DELAY_MS = 5_000; const SESSION_LIST_TERMINAL_REFRESH_DELAY_MS = 1_500; const SESSION_LIST_MIN_REFRESH_INTERVAL_MS = 15_000; +const WORKBENCH_REALTIME_ERROR_GAP_FILL_MIN_MS = 2_000; +const WORKBENCH_ACTIVE_TRACE_REST_GAP_FILL_INITIAL_MS = 2_500; +const WORKBENCH_ACTIVE_TRACE_REST_GAP_FILL_REPEAT_MS = 5_000; interface HydrateOptions { sessionId?: string | null; @@ -82,6 +85,8 @@ export const useWorkbenchStore = defineStore("workbench", () => { const realtimeSessionMessagesInFlight = new Set(); const realtimeSessionMessagesLastRefreshAtById = new Map(); const realtimeSessionRefreshInFlight = new Set(); + const realtimeErrorGapFillLastAtByKey = new Map(); + const activeTraceRestGapFillTimers = new Map(); let realtimeStream: WorkbenchEventStream | null = null; let realtimeKey = ""; const realtimeOutboxSeqByKey = new Map(); @@ -420,6 +425,12 @@ export const useWorkbenchStore = defineStore("workbench", () => { updateSessionMessages(ownerSessionId, (source) => source.map((message) => messageMatchesTraceAuthority(message, traceId, authoritySessionId, ownerSessionId) ? project(message) : message)); } + function latestMessageForTrace(traceId: string | null | undefined, source: ChatMessage[] = messages.value): ChatMessage | null { + const id = firstNonEmptyString(traceId); + if (!id) return null; + return [...source].reverse().find((message) => firstNonEmptyString(message.traceId, message.runnerTrace?.traceId) === id) ?? null; + } + function appendActiveMessages(...items: ChatMessage[]): void { replaceActiveMessages([...messages.value, ...items]); } @@ -564,21 +575,21 @@ export const useWorkbenchStore = defineStore("workbench", () => { return false; } - function fetchWorkbenchTurnStatus(traceId: string, useActivityTimeout = shouldUseActivityTimeoutForTrace(traceId)): Promise> { + function fetchWorkbenchTurnStatus(traceId: string, useActivityTimeout = shouldUseActivityTimeoutForTrace(traceId), options: { force?: boolean } = {}): Promise> { const activitySource = useActivityTimeout ? () => activityRef.value : null; return runWorkbenchReadHydration( () => api.workbench.turn(traceId, 8000, activitySource), workbenchReadCooldownKey("turn", traceId), - { minIntervalMs: WORKBENCH_TURN_STATUS_MIN_REFRESH_MS }, + { minIntervalMs: WORKBENCH_TURN_STATUS_MIN_REFRESH_MS, force: options.force }, ); } - function fetchWorkbenchTraceEvents(traceId: string, afterProjectedSeq: number, useActivityTimeout = shouldUseActivityTimeoutForTrace(traceId)): Promise> { + function fetchWorkbenchTraceEvents(traceId: string, afterProjectedSeq: number, useActivityTimeout = shouldUseActivityTimeoutForTrace(traceId), options: { force?: boolean } = {}): Promise> { const activitySource = useActivityTimeout ? () => activityRef.value : null; return runWorkbenchReadHydration( () => api.workbench.traceEvents(traceId, WORKBENCH_TRACE_EVENTS_TIMEOUT_MS, activitySource, { afterProjectedSeq, limit: TRACE_HYDRATION_PAGE_LIMIT }), workbenchReadCooldownKey("trace-events", traceId), - { minIntervalMs: WORKBENCH_TRACE_EVENTS_MIN_REFRESH_MS }, + { minIntervalMs: WORKBENCH_TRACE_EVENTS_MIN_REFRESH_MS, force: options.force }, ); } @@ -621,13 +632,13 @@ export const useWorkbenchStore = defineStore("workbench", () => { } } - async function refreshMessageProjectionForTrace(sessionId: string | null | undefined, traceId: string): Promise { + async function refreshMessageProjectionForTrace(sessionId: string | null | undefined, traceId: string, options: { force?: boolean } = {}): Promise { const id = normalizeWorkbenchSessionId(sessionId); if (!id) return; const response = await runWorkbenchReadHydration( () => api.workbench.sessionMessages(id, { limit: 100 }), workbenchReadCooldownKey("session-messages", id), - { minIntervalMs: WORKBENCH_SESSION_MESSAGES_MIN_REFRESH_MS }, + { minIntervalMs: WORKBENCH_SESSION_MESSAGES_MIN_REFRESH_MS, force: options.force }, ); if (!response.ok || !response.data) { if (shouldSuppressTransientWorkbenchReadFailure(response)) return; @@ -700,13 +711,13 @@ export const useWorkbenchStore = defineStore("workbench", () => { return true; } - async function refreshTurnStatusByTraceId(traceId: string | null | undefined): Promise { + async function refreshTurnStatusByTraceId(traceId: string | null | undefined, options: { force?: boolean } = {}): Promise { const id = firstNonEmptyString(traceId); if (!id) return; - const response = await fetchWorkbenchTurnStatus(id); + const response = await fetchWorkbenchTurnStatus(id, shouldUseActivityTimeoutForTrace(id), { force: options.force }); if (response.ok && response.data) { applyTurnStatusSnapshot(id, response.data); - if (response.data.terminal === true || isTerminalMessageStatus(response.data.status)) completeTrace(id, response.data); + if (response.data.terminal === true || isTerminalMessageStatus(response.data.status)) completeTrace(id, response.data, { forceRead: options.force }); return; } if (shouldSuppressTransientWorkbenchReadFailure(response)) return; @@ -862,6 +873,7 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project completeTrace(canonicalTraceId, response.data as AgentChatResultResponse); return true; } + scheduleActiveTraceRestGapFill(canonicalTraceId, "submit-rest-gap"); restartRealtime("submit"); return true; } @@ -878,6 +890,7 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project recordActivity("cancel:accepted"); if (message.status === "running") chatPending.value = false; currentRequest.value = null; + clearActiveTraceRestGapFill(traceId); void clearActiveTrace(traceId, "cancel-agent-message"); scheduleSessionListRefresh(sessionId, SESSION_LIST_TERMINAL_REFRESH_DELAY_MS); } @@ -900,7 +913,7 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project await submitMessage(retryInput); } - async function hydrateTraceEventsForMessage(message: ChatMessage): Promise { + async function hydrateTraceEventsForMessage(message: ChatMessage, options: { force?: boolean } = {}): Promise { const traceId = message.traceId ?? message.runnerTrace?.traceId; if (!traceId) return; if (traceHydrationInFlight.has(traceId)) return; @@ -908,7 +921,7 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project try { let afterProjectedSeq = traceHydrationProjectedSeq(message.runnerTrace); for (let page = 0; page < TRACE_HYDRATION_MAX_PAGES; page += 1) { - const result = await fetchTraceHydrationPage(traceId, afterProjectedSeq); + const result = await fetchTraceHydrationPage(traceId, afterProjectedSeq, { force: options.force }); if (!result.ok || !result.data) { if (shouldSuppressTransientWorkbenchReadFailure(result)) return; if (messageHasCompletedFinalResponse(message)) return; @@ -925,10 +938,10 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project } } - async function fetchTraceHydrationPage(traceId: string, afterProjectedSeq: number): Promise> { + async function fetchTraceHydrationPage(traceId: string, afterProjectedSeq: number, options: { force?: boolean } = {}): Promise> { let lastResult: ApiResult | null = null; for (let attempt = 0; attempt < TRACE_HYDRATION_MAX_ATTEMPTS; attempt += 1) { - const result = await fetchWorkbenchTraceEvents(traceId, afterProjectedSeq); + const result = await fetchWorkbenchTraceEvents(traceId, afterProjectedSeq, shouldUseActivityTimeoutForTrace(traceId), { force: options.force }); if (result.ok && result.data) return result; lastResult = result; if (attempt < TRACE_HYDRATION_MAX_ATTEMPTS - 1) await delayTraceHydrationRetry(TRACE_HYDRATION_RETRY_DELAY_MS * (attempt + 1)); @@ -1103,6 +1116,7 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project const initial: AgentChatResponse = { status: "running", traceId, turnUrl: `/v1/workbench/turns/${encodeURIComponent(traceId)}` }; if (!messages.value.some((message) => message.traceId === traceId)) appendActiveMessages(makeMessage("agent", "", "running", { traceId, sessionId: selectedSessionId.value ?? undefined, threadId: selectedThreadId.value ?? undefined, title: "Code Agent", traceAutoLifecycle: "running" })); currentRequest.value = { traceId, sessionId: selectedSessionId.value ?? null, threadId: selectedThreadId.value ?? null, status: initial.status }; + scheduleActiveTraceRestGapFill(traceId, "reattach-rest-gap"); restartRealtime("reattach"); } @@ -1134,11 +1148,70 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project traceId, afterSeq, onOpen: () => undefined, - onError: () => undefined, + onError: () => handleRealtimeStreamError(sessionId, traceId), onEvent: (event, eventName) => applyRealtimeEvent(event, eventName) }); } + function handleRealtimeStreamError(sessionId: string | null, traceId: string | null): void { + const activeId = normalizeWorkbenchSessionId(sessionId ?? selectedSessionId.value); + const activeTraceId = firstNonEmptyString(traceId, realtimeTraceId()); + const key = [activeId ?? "", activeTraceId ?? ""].join("|"); + if (!key.trim()) return; + const lastAt = realtimeErrorGapFillLastAtByKey.get(key) ?? 0; + if (lastAt > 0 && Date.now() - lastAt < WORKBENCH_REALTIME_ERROR_GAP_FILL_MIN_MS) return; + realtimeErrorGapFillLastAtByKey.set(key, Date.now()); + if (activeId === activeSessionId.value) void refreshRealtimeSessionMessages(activeId, "realtime-error:messages", { force: true }); + if (activeId) void refreshSessions(activeId, { force: true }); + if (!activeTraceId || !shouldApplyActiveTraceAuthority(activeTraceId, activeId)) return; + void refreshTurnStatusByTraceId(activeTraceId, { force: true }); + const message = latestMessageForTrace(activeTraceId); + if (message) void hydrateTraceEventsForMessage(message, { force: true }); + } + + function scheduleActiveTraceRestGapFill(traceId: string | null | undefined, reason: string, delayMs = WORKBENCH_ACTIVE_TRACE_REST_GAP_FILL_INITIAL_MS): void { + const id = firstNonEmptyString(traceId); + if (!id || typeof window === "undefined") return; + clearActiveTraceRestGapFill(id); + const timer = window.setTimeout(() => { + activeTraceRestGapFillTimers.delete(id); + void refreshActiveTraceFromRest(id, reason); + }, Math.max(0, Math.trunc(delayMs))); + activeTraceRestGapFillTimers.set(id, timer); + } + + function clearActiveTraceRestGapFill(traceId: string | null | undefined): void { + const id = firstNonEmptyString(traceId); + if (!id || typeof window === "undefined") return; + const timer = activeTraceRestGapFillTimers.get(id); + if (timer) window.clearTimeout(timer); + activeTraceRestGapFillTimers.delete(id); + } + + async function refreshActiveTraceFromRest(traceId: string, reason: string): Promise { + const id = firstNonEmptyString(traceId); + if (!id) return; + const ownerBefore = traceOwnerSessionId(id, null); + if (ownerBefore === activeSessionId.value) recordActivity(reason); + await refreshTurnStatusByTraceId(id, { force: true }); + const ownerSessionId = traceOwnerSessionId(id, turnStatusAuthority.value[id]?.sessionId ?? null) ?? ownerBefore; + if (ownerSessionId) { + await refreshSessions(ownerSessionId, { force: true }); + await refreshMessageProjectionForTrace(ownerSessionId, id, { force: true }); + } + const ownerMessages = ownerSessionId ? serverState.value.messagesBySessionId[ownerSessionId] ?? [] : messages.value; + const message = latestMessageForTrace(id, ownerMessages); + if (message) await hydrateTraceEventsForMessage(message, { force: true }); + const turn = turnStatusAuthority.value[id]; + const terminal = turn?.terminal === true || isTerminalMessageStatus(turn?.status) || isTerminalMessageStatus(message?.status); + if (terminal) { + clearActiveTraceRestGapFill(id); + return; + } + const stillActive = currentRequest.value?.traceId === id || isTraceActiveStatus(turn?.status) || isTraceActiveStatus(message?.status); + if (stillActive) scheduleActiveTraceRestGapFill(id, "active-rest-gap:repeat", WORKBENCH_ACTIVE_TRACE_REST_GAP_FILL_REPEAT_MS); + } + function stopRealtime(): void { realtimeStream?.close(); realtimeStream = null; @@ -1300,10 +1373,26 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project } async function refreshTerminalTraceFromRest(traceId: string, reason: string): Promise { - if (terminalRealtimeRefreshInFlight.has(traceId)) return; - terminalRealtimeRefreshInFlight.add(traceId); - terminalRealtimeRefreshInFlight.delete(traceId); - void reason; + const id = firstNonEmptyString(traceId); + if (!id || terminalRealtimeRefreshInFlight.has(id)) return; + terminalRealtimeRefreshInFlight.add(id); + try { + const ownerBefore = traceOwnerSessionId(id, null); + if (ownerBefore === activeSessionId.value) recordActivity(reason); + await refreshTurnStatusByTraceId(id, { force: true }); + const ownerSessionId = traceOwnerSessionId(id, turnStatusAuthority.value[id]?.sessionId ?? null) ?? ownerBefore; + if (ownerSessionId) await refreshSessions(ownerSessionId, { force: true }); + if (ownerSessionId) await refreshMessageProjectionForTrace(ownerSessionId, id, { force: true }); + const ownerMessages = ownerSessionId ? serverState.value.messagesBySessionId[ownerSessionId] ?? [] : messages.value; + const message = latestMessageForTrace(id, ownerMessages); + if (message) { + await hydrateTraceEventsForMessage(message, { force: true }); + } else if (ownerSessionId) { + await refreshRealtimeSessionMessages(ownerSessionId, `${reason}:message-gap`, { force: true }); + } + } finally { + terminalRealtimeRefreshInFlight.delete(id); + } } function installRealtimeVisibilityHandler(): void { @@ -1393,7 +1482,7 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project } } - function completeTrace(traceId: string, result: AgentChatResultResponse): void { + function completeTrace(traceId: string, result: AgentChatResultResponse, options: { forceRead?: boolean } = {}): void { const authoritySessionId = traceResultSessionId(result); const ownerSessionId = traceOwnerSessionId(traceId, authoritySessionId); if (!ownerSessionId) return; @@ -1415,8 +1504,13 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project })); rememberTurnStatus(traceId, result); markWorkbenchTraceProjected(traceId); - void refreshMessageProjectionForTrace(ownerSessionId, traceId); - void hydrateTraceEvents(serverState.value.messagesBySessionId[ownerSessionId] ?? []); + clearActiveTraceRestGapFill(traceId); + void refreshMessageProjectionForTrace(ownerSessionId, traceId, { force: options.forceRead }); + const ownerMessages = serverState.value.messagesBySessionId[ownerSessionId] ?? []; + const terminalMessage = latestMessageForTrace(traceId, ownerMessages); + if (options.forceRead && terminalMessage) void hydrateTraceEventsForMessage(terminalMessage, { force: true }); + else void hydrateTraceEvents(ownerMessages); + if (options.forceRead) void refreshSessions(ownerSessionId, { force: true }); if (ownerSessionId === activeSessionId.value) { chatPending.value = false; currentRequest.value = null; @@ -1596,6 +1690,7 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project } async function clearActiveTrace(traceId: string, reason: string): Promise { + clearActiveTraceRestGapFill(traceId); if (currentRequest.value?.traceId === traceId) currentRequest.value = null; }