From 7fbb55802bbd826aa05653dddccc22b221c0d968 Mon Sep 17 00:00:00 2001 From: root Date: Tue, 21 Jul 2026 04:52:06 +0200 Subject: [PATCH] =?UTF-8?q?fix:=20=E9=99=8D=E4=BD=8E=20Workbench=20?= =?UTF-8?q?=E9=95=BF=E7=A8=8B=E4=BA=8B=E4=BB=B6=E6=8A=95=E5=BD=B1=E5=BC=80?= =?UTF-8?q?=E9=94=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../src/api/workbench-events.test.ts | 28 +++++++++++++++++++ .../src/api/workbench-events.ts | 4 ++- .../stores/workbench-live-kafka-event.test.ts | 15 +++++++++- .../src/stores/workbench-live-kafka-event.ts | 7 +++++ web/hwlab-cloud-web/src/stores/workbench.ts | 9 ++++-- 5 files changed, 58 insertions(+), 5 deletions(-) diff --git a/web/hwlab-cloud-web/src/api/workbench-events.test.ts b/web/hwlab-cloud-web/src/api/workbench-events.test.ts index 5e4f9a3e..df420526 100644 --- a/web/hwlab-cloud-web/src/api/workbench-events.test.ts +++ b/web/hwlab-cloud-web/src/api/workbench-events.test.ts @@ -63,6 +63,34 @@ test("product EventSource tees projection frames once before reducer delivery", } }); +test("product EventSource batches a burst until the configured scheduler runs", async () => { + const originalEventSource = globalThis.EventSource; + const delivered: WorkbenchRealtimeEvent[] = []; + FakeProductEventSource.instances = []; + globalThis.EventSource = FakeProductEventSource as unknown as typeof EventSource; + try { + const stream = connectWorkbenchEvents({ + realtimeCapabilities: { liveKafkaSse: true, kafkaRefreshReplay: true, projectionRealtime: false }, + sessionId: "ses_frame_batch", + flushYieldMs: 1, + onEvent: (event) => delivered.push(event) + }); + assert.ok(stream); + const source = FakeProductEventSource.instances[0]; + source.emitRaw("hwlab.event.v1", '{"schema":"hwlab.event.v1","eventId":"evt_1"}'); + source.emitRaw("hwlab.event.v1", '{"schema":"hwlab.event.v1","eventId":"evt_2"}'); + + assert.equal(delivered.length, 0); + await Promise.resolve(); + assert.equal(delivered.length, 0); + await new Promise((resolve) => setTimeout(resolve, 10)); + assert.equal(delivered.length, 2); + stream.close(); + } finally { + globalThis.EventSource = originalEventSource; + } +}); + class FakeProductEventSource { static instances: FakeProductEventSource[] = []; onopen: ((event: Event) => void) | null = null; diff --git a/web/hwlab-cloud-web/src/api/workbench-events.ts b/web/hwlab-cloud-web/src/api/workbench-events.ts index 1653257d..1c58902f 100644 --- a/web/hwlab-cloud-web/src/api/workbench-events.ts +++ b/web/hwlab-cloud-web/src/api/workbench-events.ts @@ -122,11 +122,13 @@ export function connectWorkbenchEvents(options: WorkbenchEventStreamOptions): Wo const connectStartedAt = typeof performance === "undefined" ? Date.now() : performance.now(); recordWorkbenchSseLifecycle({ state: "connect", route: eventRoute, sessionId: options.sessionId, traceId: options.traceId }); const source = new EventSource(eventRoute, { withCredentials: true }); + const scheduleFlush = (flush: () => void) => scheduleRealtimeFlushYield(flush, options.flushYieldMs); const queue = createCoalescedEventQueue({ keyOf: (item) => realtimeCoalesceKey(item.payload, item.eventName), maxItemsPerChunk: options.flushMaxItemsPerChunk, maxChunkMs: options.flushMaxChunkMs, - yieldSchedule: (flush) => scheduleRealtimeFlushYield(flush, options.flushYieldMs), + schedule: scheduleFlush, + yieldSchedule: scheduleFlush, onFlush: (items) => { for (const item of items) options.onEvent(item.payload, item.eventName); }, diff --git a/web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.test.ts b/web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.test.ts index 97b30b58..3015f25d 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.test.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.test.ts @@ -2,7 +2,7 @@ import assert from "node:assert/strict"; import { test } from "bun:test"; import type { ChatMessage, TraceEvent } from "../types"; -import { projectWorkbenchLiveKafkaMessage, projectWorkbenchLiveKafkaUserMessage, projectWorkbenchLiveKafkaUserSession, workbenchAgentMessageIdForTrace, workbenchLiveKafkaEnvelope, workbenchLiveKafkaEventIsUserMessage, workbenchLiveKafkaProjectionTarget, workbenchUserMessageIdForTrace } from "./workbench-live-kafka-event"; +import { projectWorkbenchLiveKafkaMessage, projectWorkbenchLiveKafkaUserMessage, projectWorkbenchLiveKafkaUserSession, workbenchAgentMessageIdForTrace, workbenchLiveKafkaEnvelope, workbenchLiveKafkaEventIsUserMessage, workbenchLiveKafkaProjectionTarget, workbenchLiveKafkaSessionSummaryChanged, workbenchUserMessageIdForTrace } from "./workbench-live-kafka-event"; import { createWorkbenchServerState, reduceWorkbenchServerState, selectActiveMessages, type WorkbenchServerState } from "./workbench-server-state"; import { sessionToSessionTab } from "./workbench-session"; @@ -12,6 +12,19 @@ test("only canonical HWLAB Kafka envelopes bypass terminal priority", () => { assert.equal(workbenchLiveKafkaEnvelope(null), false); }); +test("running Kafka events do not rebuild an unchanged session summary", () => { + const previous = { + sessionId: "ses_longrun", + threadId: "thread_longrun", + status: "running", + lastTraceId: "trc_longrun", + lastUserMessageAt: "2026-07-21T02:18:22.034Z", + updatedAt: "2026-07-21T02:18:22.034Z" + }; + assert.equal(workbenchLiveKafkaSessionSummaryChanged(previous, { ...previous }), false); + assert.equal(workbenchLiveKafkaSessionSummaryChanged(previous, { ...previous, status: "completed" }), true); +}); + test("live Kafka user event restores and idempotently updates the user bubble", () => { const event = traceEvent(1, "user", { userMessageId: "msg_live_user_user", diff --git a/web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.ts b/web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.ts index 5d8e1e04..181c33a7 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.ts @@ -121,6 +121,13 @@ export function projectWorkbenchLiveKafkaUserSession(input: WorkbenchLiveKafkaUs }; } +export function workbenchLiveKafkaSessionSummaryChanged(previous: WorkbenchSessionRecord, next: WorkbenchSessionRecord): boolean { + return firstNonEmptyString(previous.threadId) !== firstNonEmptyString(next.threadId) + || firstNonEmptyString(previous.status) !== firstNonEmptyString(next.status) + || firstNonEmptyString(previous.lastTraceId) !== firstNonEmptyString(next.lastTraceId) + || firstNonEmptyString(previous.lastUserMessageAt) !== firstNonEmptyString(next.lastUserMessageAt); +} + export function workbenchLiveKafkaEventIsTerminal(event: TraceEvent | null | undefined): boolean { return Boolean(event?.terminal === true || firstNonEmptyString(event?.eventType) === "terminal" || firstNonEmptyString(event?.type) === "result"); } diff --git a/web/hwlab-cloud-web/src/stores/workbench.ts b/web/hwlab-cloud-web/src/stores/workbench.ts index 8b3c0b35..ced24e2e 100644 --- a/web/hwlab-cloud-web/src/stores/workbench.ts +++ b/web/hwlab-cloud-web/src/stores/workbench.ts @@ -21,7 +21,7 @@ import { initialWorkbenchSessionIdFromLocation } from "./workbench-projection"; import { cleanupWorkbenchServerStateSessions, selectActiveMessages, selectActiveSession, selectSessionList, selectSessionStatusAuthority, selectTraceAuthorityById, selectTurnStatusAuthority, type WorkbenchServerAction } from "./workbench-server-state"; import { cleanupDroppedWorkbenchSessionCaches, trimWorkbenchSessionCache } from "./workbench-session-cache"; import { reduceWorkbenchRealtimeEvent, workbenchRealtimeEventIsBusinessActivity, type WorkbenchRealtimeAction } from "./workbench-event-reducer"; -import { projectWorkbenchLiveKafkaMessage, projectWorkbenchLiveKafkaUserMessage, projectWorkbenchLiveKafkaUserSession, workbenchLiveKafkaEnvelope, workbenchLiveKafkaProjectionTarget } from "./workbench-live-kafka-event"; +import { projectWorkbenchLiveKafkaMessage, projectWorkbenchLiveKafkaUserMessage, projectWorkbenchLiveKafkaUserSession, workbenchLiveKafkaEnvelope, workbenchLiveKafkaProjectionTarget, workbenchLiveKafkaSessionSummaryChanged } from "./workbench-live-kafka-event"; import { projectRejectedWorkbenchAdmission } from "./workbench-admission-failure"; import { messageHasSealedTerminalResult, messageIsSealedTerminal, traceAuthorityIsSealed } from "./workbench-terminal-authority"; import { @@ -1109,14 +1109,17 @@ export const useWorkbenchStore = defineStore("workbench", () => { const existing = sessions.value.find((session) => session.sessionId === ownerSessionId) ?? null; if (existing) { const lastUserMessageAt = firstNonEmptyString(existing.lastUserMessageAt, turnStartedAt); - rememberSessionList(mergeSessionIntoList(sessions.value, { + const nextSession = { ...existing, threadId: message.threadId ?? existing.threadId, status, lastTraceId: traceId, lastUserMessageAt, updatedAt: lastUserMessageAt ?? existing.updatedAt - })); + }; + if (workbenchLiveKafkaSessionSummaryChanged(existing, nextSession)) { + rememberSessionList(mergeSessionIntoList(sessions.value, nextSession)); + } } if (terminal && currentRequest.value?.traceId === traceId) { chatPending.value = false;