From 54f8b468aa04b2bb532b6cb9d5ee45b30c20475d Mon Sep 17 00:00:00 2001 From: lyon Date: Fri, 26 Jun 2026 16:26:16 +0800 Subject: [PATCH] fix: refresh workbench messages on realtime trace gaps --- web/hwlab-cloud-web/src/stores/workbench.ts | 33 +++++++++++++++++---- 1 file changed, 28 insertions(+), 5 deletions(-) diff --git a/web/hwlab-cloud-web/src/stores/workbench.ts b/web/hwlab-cloud-web/src/stores/workbench.ts index 2390e2d9..ae08f1db 100644 --- a/web/hwlab-cloud-web/src/stores/workbench.ts +++ b/web/hwlab-cloud-web/src/stores/workbench.ts @@ -28,6 +28,7 @@ const WORKBENCH_READ_FAILURE_COOLDOWN_MS = 5_000; const WORKBENCH_TURN_STATUS_MIN_REFRESH_MS = 2_000; const WORKBENCH_TRACE_EVENTS_MIN_REFRESH_MS = 4_000; const WORKBENCH_SESSION_MESSAGES_MIN_REFRESH_MS = 5_000; +const WORKBENCH_REALTIME_SESSION_MESSAGES_MIN_REFRESH_MS = 1_000; 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; @@ -76,6 +77,8 @@ export const useWorkbenchStore = defineStore("workbench", () => { const traceHydrationQueue: ChatMessage[] = []; let traceHydrationPumpActive = false; const terminalRealtimeRefreshInFlight = new Set(); + const realtimeSessionMessagesInFlight = new Set(); + const realtimeSessionMessagesLastRefreshAtById = new Map(); const realtimeSessionRefreshInFlight = new Set(); let realtimeStream: WorkbenchEventStream | null = null; let realtimeKey = ""; @@ -437,7 +440,7 @@ export const useWorkbenchStore = defineStore("workbench", () => { void source; } - function runWorkbenchReadHydration(task: () => Promise, cooldownKey?: string, options: { minIntervalMs?: number } = {}): Promise { + function runWorkbenchReadHydration(task: () => Promise, cooldownKey?: string, options: { minIntervalMs?: number; force?: boolean } = {}): Promise { return new Promise((resolve, reject) => { const run = () => { if (cooldownKey) { @@ -451,7 +454,7 @@ export const useWorkbenchStore = defineStore("workbench", () => { if (cooldownUntilMs > 0) workbenchReadCooldownUntilByKey.delete(cooldownKey); const minIntervalMs = Math.max(0, Math.trunc(options.minIntervalMs ?? 0)); const lastStartedAtMs = workbenchReadLastStartedAtByKey.get(cooldownKey) ?? 0; - if (minIntervalMs > 0 && lastStartedAtMs > 0 && Date.now() - lastStartedAtMs < minIntervalMs) { + if (!options.force && minIntervalMs > 0 && lastStartedAtMs > 0 && Date.now() - lastStartedAtMs < minIntervalMs) { resolve(createWorkbenchReadThrottledResult(cooldownKey, lastStartedAtMs + minIntervalMs) as T); const next = workbenchReadHydrationQueue.shift(); if (next) next(); @@ -586,19 +589,34 @@ export const useWorkbenchStore = defineStore("workbench", () => { return isTraceActiveStatus(turn?.status) || isTraceActiveStatus(message?.status); } - async function refreshSessionMessageProjectionPage(sessionId: string | null | undefined): Promise { + async function refreshSessionMessageProjectionPage(sessionId: string | null | undefined, 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) return; const pageMessages = Array.isArray(response.data.messages) ? response.data.messages.map((message) => normalizeChatMessage(message as ChatMessage)) : []; rememberSessionMessages(id, mergeMessageProjectionPage(id, pageMessages)); } + async function refreshRealtimeSessionMessages(sessionId: string | null | undefined, reason: string): Promise { + const id = normalizeWorkbenchSessionId(sessionId); + if (!id || id !== activeSessionId.value || realtimeSessionMessagesInFlight.has(id)) return; + const lastRefreshAt = realtimeSessionMessagesLastRefreshAtById.get(id) ?? 0; + if (lastRefreshAt > 0 && Date.now() - lastRefreshAt < WORKBENCH_REALTIME_SESSION_MESSAGES_MIN_REFRESH_MS) return; + realtimeSessionMessagesLastRefreshAtById.set(id, Date.now()); + realtimeSessionMessagesInFlight.add(id); + try { + recordActivity(reason); + await refreshSessionMessageProjectionPage(id, { force: true }); + } finally { + realtimeSessionMessagesInFlight.delete(id); + } + } + async function refreshMessageProjectionForTrace(sessionId: string | null | undefined, traceId: string): Promise { const id = normalizeWorkbenchSessionId(sessionId); if (!id) return; @@ -1269,6 +1287,8 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project const traceId = firstNonEmptyString(turn.traceId); if (!traceId) return; if (!shouldApplyActiveTraceAuthority(traceId, traceResultSessionId(turn))) return; + const activeId = activeSessionId.value; + if (activeId && !messages.value.some((message) => firstNonEmptyString(message.traceId, message.runnerTrace?.traceId) === traceId)) void refreshRealtimeSessionMessages(activeId, `realtime-turn-gap:${traceId}`); const status = firstNonEmptyString(turn.status) ?? undefined; applyTurnStatusSnapshot(traceId, { ...turn, traceId, status, running: turn.running === true, terminal: turn.terminal === true, sessionId: firstNonEmptyString(turn.sessionId) ?? undefined, threadId: firstNonEmptyString(turn.threadId) ?? undefined, agentRun: turn.agentRun as AgentRunProvenance | undefined } as AgentChatResultResponse); if (turn.terminal === true || isTerminalMessageStatus(status)) void refreshTerminalTraceFromRest(traceId, "realtime-turn-snapshot"); @@ -1310,7 +1330,10 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project const projection = clearCompletedDiagnostics ? nonBlockingProjection(trace.projection ?? null) : trace.projection ?? runnerTrace.projection ?? message.projection ?? null; return { ...message, ...messageTimingPatchForMerge(message, trace), runnerTrace, error: clearCompletedDiagnostics ? null : error ?? message.error ?? null, projection, projectionStatus: projection?.projectionStatus ?? null, projectionHealth: projection?.projectionHealth ?? null, blocker: projection?.blocker ?? null, updatedAt: new Date().toISOString() }; })); - if (!matchedMessage && ownerSessionId === activeSessionId.value) void refreshRealtimeSessionFromRest(ownerSessionId, `realtime-trace-snapshot:${traceId}`); + if (!matchedMessage && ownerSessionId === activeSessionId.value) { + void refreshRealtimeSessionMessages(ownerSessionId, `realtime-trace-gap:${traceId}`); + void refreshRealtimeSessionFromRest(ownerSessionId, `realtime-trace-snapshot:${traceId}`); + } markWorkbenchTraceProjected(traceId); scheduleSessionListRefresh(ownerSessionId, SESSION_LIST_REALTIME_REFRESH_DELAY_MS); }