mirrors/pikasTech-HWLAB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: refresh workbench messages on realtime trace gaps

Browse Source
This commit is contained in:
lyon
2026-06-26 16:26:16 +08:00
parent 496895cae5
commit 54f8b468aa
1 changed files with 28 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
web/hwlab-cloud-web/src/stores/workbench.ts
+28 -5
View File
@@ -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<string>();
const realtimeSessionMessagesInFlight = new Set<string>();
const realtimeSessionMessagesLastRefreshAtById = new Map<string, number>();
const realtimeSessionRefreshInFlight = new Set<string>();
let realtimeStream: WorkbenchEventStream | null = null;
let realtimeKey = "";
@@ -437,7 +440,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
void source;
}
function runWorkbenchReadHydration<T>(task: () => Promise<T>, cooldownKey?: string, options: { minIntervalMs?: number } = {}): Promise<T> {
function runWorkbenchReadHydration<T>(task: () => Promise<T>, cooldownKey?: string, options: { minIntervalMs?: number; force?: boolean } = {}): Promise<T> {
return new Promise<T>((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<void> {
async function refreshSessionMessageProjectionPage(sessionId: string | null | undefined, options: { force?: boolean } = {}): Promise<void> {
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<void> {
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<void> {
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);
}