Merge remote-tracking branch 'origin/v0.3' into fix/2721-nc01-ci-resource-governance

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;

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<QueuedRealtimeEvent>({
     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);
     },

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",

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");
 }

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;