diff --git a/internal/cloud/kafka-event-bridge.test.ts b/internal/cloud/kafka-event-bridge.test.ts index faf6dd57..6db47a25 100644 --- a/internal/cloud/kafka-event-bridge.test.ts +++ b/internal/cloud/kafka-event-bridge.test.ts @@ -379,6 +379,19 @@ test("projects AgentRun fileChange and diff events without dropping paths or uni assert.deepEqual(fileChange.event.changes, changes); assert.equal(fileChange.event.fileCount, 2); + const truncatedFileChange = projectAgentRunKafkaEventToHwlabEvent({ + schema: "agentrun.event.v1", + eventType: "agentrun.event.committed", + eventId: "evt_file_change_truncated", + traceId: "trc_file_change", + hwlabSessionId: "ses_file_change", + runId: "run_file_change", + event: { id: "evt_file_change_truncated", seq: 22, type: "tool_call", payload: { method: "item/completed", itemId: "item_large", type: "fileChange", toolName: "fileChange", status: "completed", changes: [{ path: "src/large.ts", kind: "add", additions: 400, deletions: 0, diffTruncated: true }], fileCount: 1, summary: { text: "bounded file change metadata", outputTruncated: true }, outputTruncated: true } } + }, { source: "hwlab-test", sourceTopic: "agentrun.event.debug.v1", sourceOffset: "22" }); + assert.equal(truncatedFileChange.event.message, "AgentRun fileChange (1 file)"); + assert.deepEqual(truncatedFileChange.event.changes, [{ path: "src/large.ts", kind: "add", additions: 400, deletions: 0, diffTruncated: true }]); + assert.doesNotMatch(truncatedFileChange.event.message, /\[object Object\]/u); + const diff = projectAgentRunKafkaEventToHwlabEvent({ schema: "agentrun.event.v1", eventType: "agentrun.event.committed", diff --git a/internal/cloud/kafka-event-bridge.ts b/internal/cloud/kafka-event-bridge.ts index 737b5c3b..ac7896a4 100644 --- a/internal/cloud/kafka-event-bridge.ts +++ b/internal/cloud/kafka-event-bridge.ts @@ -1632,13 +1632,14 @@ function mapAgentRunSourceEventToHwlabEvent({ input, sourceEvent, payload, run, } if (type === "tool_call") { const toolName = firstText(payload.toolName, payload.name, payload.method) || "tool"; + const changes = Array.isArray(payload.changes) ? payload.changes : []; return { ...base, type: "tool", eventType: "tool", status: firstText(payload.status) || "running", label: `agentrun:tool:${toolName}`, - message: textPayload(payload, `AgentRun tool ${toolName}`), + message: toolName === "fileChange" ? `AgentRun fileChange (${changes.length} file${changes.length === 1 ? "" : "s"})` : textPayload(payload, `AgentRun tool ${toolName}`), toolName, toolType: firstText(payload.type), method: firstText(payload.method), @@ -1654,7 +1655,7 @@ function mapAgentRunSourceEventToHwlabEvent({ input, sourceEvent, payload, run, commandTruncated: payload.commandTruncated === true, outputBytes: integerValue(payload.outputBytes ?? payload.summary?.outputBytes), outputTruncated: payload.outputTruncated === true || payload.summary?.outputTruncated === true, - ...(Array.isArray(payload.changes) ? { changes: payload.changes, fileCount: integerValue(payload.fileCount) } : {}), + ...(changes.length > 0 ? { changes, fileCount: integerValue(payload.fileCount) } : {}), terminal: false }; } @@ -1936,7 +1937,7 @@ function kafkaHeaders(projected) { } function textPayload(payload, fallback) { - return firstText(payload.text, payload.message, payload.summary, payload.output, payload.errorMessage, payload.reason) || fallback; + return firstText(payload.text, payload.message, payload.summary?.text, payload.output, payload.errorMessage, payload.reason) || fallback; } function truthy(value) { diff --git a/tools/hwlab-cli/kafka-regenerate.test.ts b/tools/hwlab-cli/kafka-regenerate.test.ts index efb720ee..64d25bd0 100644 --- a/tools/hwlab-cli/kafka-regenerate.test.ts +++ b/tools/hwlab-cli/kafka-regenerate.test.ts @@ -551,15 +551,48 @@ test("fileChange and turn diff render real paths and unified diff as valid Markd ]); const markdown = renderTraceRowsMarkdown(rows); assert.equal(rows.length, 2); - assert.match(markdown, /fileChange src\/example\.ts, docs\/说明\.md/u); - assert.match(markdown, /path: src\/example\.ts/u); - assert.match(markdown, /path: docs\/说明\.md/u); + assert.match(markdown, /2 个文件 · \+2 -1 · src\/example\.ts, docs\/说明\.md/u); + assert.match(markdown, /#### `src\/example\.ts`/u); + assert.match(markdown, /#### `docs\/说明\.md`/u); assert.match(markdown, /\+export const value = 2;/u); - assert.match(markdown, /turn\/diff\/updated/u); + assert.match(markdown, /文件 diff 已更新/u); assert.match(markdown, /\+# 文件编辑可见/u); assert.equal((markdown.match(/
/gu) ?? []).length, 2); assert.equal((markdown.match(/<\/details>/gu) ?? []).length, 2); - assert.equal((markdown.match(/```text/gu) ?? []).length, 2); + assert.ok(rows.every((row) => row.bodyFormat === "markdown")); +}); + +test("fileChange rendering uses preserved stats and ignores historical object coercion", () => { + const rows = traceDisplayRows({ eventSource: "hwlab-kafka-sse" }, [ + { + sourceSeq: 53, + type: "tool", + eventType: "tool", + status: "completed", + label: "agentrun:tool:fileChange", + toolName: "fileChange", + itemId: "item_large_file_change", + changes: [{ path: "/agentrun-workspace/src/large.ts", kind: "add", additions: 400, deletions: 0, diffTruncated: true }], + message: "bounded file change metadata", + createdAt: "2026-07-21T08:04:30.452Z" + }, + { + sourceSeq: 54, + type: "tool", + eventType: "tool", + status: "completed", + label: "agentrun:tool:fileChange", + toolName: "fileChange", + itemId: "item_historical_file_change", + message: "[object Object]", + createdAt: "2026-07-21T08:04:31.452Z" + } + ]); + assert.equal(rows[0]?.preview, "src/large.ts · +400 -0"); + assert.match(rows[0]?.body ?? "", /新增 · \+400 -0/u); + assert.doesNotMatch(JSON.stringify(rows), /\[object Object\]/u); + assert.equal(rows[1]?.body, null); + assert.equal(rows[1]?.preview, "fileChange"); }); test("outer final response suppression follows formal assistant identity instead of equal text", () => { diff --git a/tools/src/hwlab-cli/trace-renderer.ts b/tools/src/hwlab-cli/trace-renderer.ts index c18d7e7d..008f37c7 100644 --- a/tools/src/hwlab-cli/trace-renderer.ts +++ b/tools/src/hwlab-cli/trace-renderer.ts @@ -403,13 +403,21 @@ function traceDiffFiles(event: TraceEvent): TraceDiffFile[] { return changes.map((change) => traceDiffFile( traceDiffPath(nonEmptyString(change.path)), traceDiffKind(change.kind), - nonEmptyString(change.diff) ?? "" + traceDiffText(change.diff) ?? "", + numberOrNull(change.additions), + numberOrNull(change.deletions) )); } - const diff = nonEmptyString(event.diff ?? event.message); + const diff = traceDiffText(event.diff ?? event.message); return diff ? traceUnifiedDiffFiles(diff) : []; } +function traceDiffText(value: unknown): string | null { + if (typeof value !== "string") return null; + const text = value.trim(); + return text && text !== "[object Object]" ? value : null; +} + function traceUnifiedDiffFiles(diff: string): TraceDiffFile[] { const sections: Array<{ path: string | null; lines: string[] }> = []; let current: { path: string | null; lines: string[] } = { path: null, lines: [] }; @@ -436,7 +444,7 @@ function traceDiffHeaderPath(value: string): string | null { return traceDiffPath(path.startsWith("b/") ? path.slice(2) : path); } -function traceDiffFile(path: string | null, kind: string | null, diff: string): TraceDiffFile { +function traceDiffFile(path: string | null, kind: string | null, diff: string, additionsHint: number | null = null, deletionsHint: number | null = null): TraceDiffFile { const normalizedDiff = traceNormalizedFileDiff(path, kind, diff); let additions = 0; let deletions = 0; @@ -444,6 +452,10 @@ function traceDiffFile(path: string | null, kind: string | null, diff: string): if (line.startsWith("+") && !line.startsWith("+++")) additions += 1; if (line.startsWith("-") && !line.startsWith("---")) deletions += 1; } + if (!normalizedDiff) { + additions = additionsHint ?? additions; + deletions = deletionsHint ?? deletions; + } return { path, kind, diff: normalizedDiff, additions, 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 df420526..c6c48a90 100644 --- a/web/hwlab-cloud-web/src/api/workbench-events.test.ts +++ b/web/hwlab-cloud-web/src/api/workbench-events.test.ts @@ -91,6 +91,57 @@ test("product EventSource batches a burst until the configured scheduler runs", } }); +test("retention replay reaches connected without animation frames", async () => { + const originalEventSource = globalThis.EventSource; + const originalWindow = globalThis.window; + const delivered: Array<[string, string | null]> = []; + let animationFrameCalls = 0; + FakeProductEventSource.instances = []; + globalThis.EventSource = FakeProductEventSource as unknown as typeof EventSource; + Object.defineProperty(globalThis, "window", { + configurable: true, + value: { + requestAnimationFrame: () => { + animationFrameCalls += 1; + return 1; + }, + cancelAnimationFrame: () => undefined + } + }); + try { + const stream = connectWorkbenchEvents({ + realtimeCapabilities: { liveKafkaSse: true, kafkaRefreshReplay: true, projectionRealtime: false }, + sessionId: "ses_background_replay", + flushMaxItemsPerChunk: 4, + flushMaxChunkMs: 8, + flushYieldMs: 0, + onEvent: (event, eventName) => delivered.push([eventName, event.eventId ?? null]) + }); + assert.ok(stream); + const source = FakeProductEventSource.instances[0]; + for (let index = 0; index < 395; index += 1) { + source.emitRaw("hwlab.event.v1", JSON.stringify({ schema: "hwlab.event.v1", eventId: `evt_${index}` })); + } + source.emitRaw("workbench.connected", JSON.stringify({ type: "connected", filters: { sessionId: "ses_background_replay" } })); + + await waitFor(() => delivered.length === 396, 1_000); + assert.equal(animationFrameCalls, 0); + assert.deepEqual(delivered.at(-1), ["workbench.connected", null]); + stream.close(); + } finally { + globalThis.EventSource = originalEventSource; + Object.defineProperty(globalThis, "window", { configurable: true, value: originalWindow }); + } +}); + +async function waitFor(predicate: () => boolean, timeoutMs: number): Promise { + const deadline = Date.now() + timeoutMs; + while (!predicate()) { + if (Date.now() >= deadline) throw new Error(`condition did not pass within ${timeoutMs}ms`); + await new Promise((resolve) => setTimeout(resolve, 5)); + } +} + 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 1c58902f..aa7dd237 100644 --- a/web/hwlab-cloud-web/src/api/workbench-events.ts +++ b/web/hwlab-cloud-web/src/api/workbench-events.ts @@ -179,6 +179,9 @@ export function connectWorkbenchEvents(options: WorkbenchEventStreamOptions): Wo close() { queue.clear("stream-close"); for (const { name, listener } of listeners) source.removeEventListener(name, listener); + source.onopen = null; + source.onerror = null; + source.onmessage = null; recordWorkbenchSseLifecycle({ state: "close", route: eventRoute, sessionId: options.sessionId, traceId: options.traceId }); source.close(); } @@ -211,12 +214,23 @@ function enqueueRealtimeFrame(raw: unknown, eventName: string, options: Workbenc if (decoded.payload) queue.push({ payload: decoded.payload as WorkbenchRealtimeEvent, eventName }); } -function scheduleRealtimeFlushYield(flush: () => void, yieldMs: number | null | undefined): () => void { - if (typeof window !== "undefined" && typeof window.requestAnimationFrame === "function" && (!Number.isFinite(Number(yieldMs)) || Number(yieldMs) <= 0)) { - const id = window.requestAnimationFrame(() => flush()); - return () => window.cancelAnimationFrame(id); - } +export function scheduleRealtimeFlushYield(flush: () => void, yieldMs: number | null | undefined): () => void { const delay = Math.max(0, Math.trunc(Number(yieldMs ?? 0))); + if (delay === 0 && typeof MessageChannel !== "undefined") { + const channel = new MessageChannel(); + let canceled = false; + channel.port1.onmessage = () => { + channel.port1.close(); + channel.port2.close(); + if (!canceled) flush(); + }; + channel.port2.postMessage(undefined); + return () => { + canceled = true; + channel.port1.close(); + channel.port2.close(); + }; + } const id = setTimeout(() => flush(), delay); return () => clearTimeout(id); } diff --git a/web/hwlab-cloud-web/src/composables/useBottomFollowScroll.test.ts b/web/hwlab-cloud-web/src/composables/useBottomFollowScroll.test.ts new file mode 100644 index 00000000..0461eafb --- /dev/null +++ b/web/hwlab-cloud-web/src/composables/useBottomFollowScroll.test.ts @@ -0,0 +1,32 @@ +import assert from "node:assert/strict"; +import { test } from "bun:test"; + +import { resolveBottomFollowingAfterScroll } from "./useBottomFollowScroll"; + +test("content growth before alignment does not cancel bottom following", () => { + assert.equal(resolveBottomFollowingAfterScroll({ + following: true, + previousScrollTop: 400, + currentScrollTop: 400, + distanceToBottom: 180, + threshold: 24 + }), true); +}); + +test("an actual upward scroll pauses following until the user returns to bottom", () => { + const paused = resolveBottomFollowingAfterScroll({ + following: true, + previousScrollTop: 580, + currentScrollTop: 300, + distanceToBottom: 280, + threshold: 24 + }); + assert.equal(paused, false); + assert.equal(resolveBottomFollowingAfterScroll({ + following: paused, + previousScrollTop: 300, + currentScrollTop: 556, + distanceToBottom: 24, + threshold: 24 + }), true); +}); diff --git a/web/hwlab-cloud-web/src/composables/useBottomFollowScroll.ts b/web/hwlab-cloud-web/src/composables/useBottomFollowScroll.ts index a847c062..b9dfcf95 100644 --- a/web/hwlab-cloud-web/src/composables/useBottomFollowScroll.ts +++ b/web/hwlab-cloud-web/src/composables/useBottomFollowScroll.ts @@ -5,12 +5,27 @@ export interface BottomFollowScrollOptions { initialFollowing?: boolean; } +export interface BottomFollowPositionState { + following: boolean; + previousScrollTop: number; + currentScrollTop: number; + distanceToBottom: number; + threshold: number; +} + +export function resolveBottomFollowingAfterScroll(state: BottomFollowPositionState): boolean { + if (state.distanceToBottom <= state.threshold) return true; + if (!state.following) return false; + return state.currentScrollTop + 1 >= state.previousScrollTop; +} + export function useBottomFollowScroll(containerRef: Ref, options: BottomFollowScrollOptions = {}) { const threshold = options.threshold ?? 24; const following = ref(options.initialFollowing ?? true); let animationFrame: number | null = null; let releaseProgrammaticFrame: number | null = null; let programmaticScroll = false; + let lastScrollTop = 0; let observedElement: T | null = null; let mutationObserver: MutationObserver | null = null; let resizeObserver: ResizeObserver | null = null; @@ -32,8 +47,19 @@ export function useBottomFollowScroll(containerRef: Ref { @@ -64,6 +90,7 @@ export function useBottomFollowScroll(containerRef: Ref(containerRef: Ref(containerRef: Ref { releaseProgrammaticFrame = null; programmaticScroll = false; - syncFollowingFromPosition(); + const element = containerRef.value; + if (element && following.value && distanceToBottom(element) > threshold) scheduleBottomAlignment(); }); } @@ -97,9 +127,12 @@ export function useBottomFollowScroll(containerRef: Ref { - refreshResizeTargets(element); + const onContentChanged = (records: MutationRecord[]) => { + if (records.some((record) => record.type === "childList" && record.target === element)) { + refreshResizeTargets(element); + } if (following.value) scheduleBottomAlignment(); }; mutationObserver = new MutationObserver(onContentChanged); @@ -140,5 +173,5 @@ export function useBottomFollowScroll(containerRef: Ref(containerR const pos = scrollPersistence.scroll(sessionKey.value, tabKey); if (!pos) return; element.scrollTo({ left: pos.x, top: pos.y, behavior: "auto" }); + bottom.syncFollowingFromPosition(); } function persistScroll(): void { diff --git a/web/hwlab-cloud-web/src/stores/workbench.ts b/web/hwlab-cloud-web/src/stores/workbench.ts index 1e54dc61..3f3c04d1 100644 --- a/web/hwlab-cloud-web/src/stores/workbench.ts +++ b/web/hwlab-cloud-web/src/stores/workbench.ts @@ -11,6 +11,7 @@ import { createWorkbenchHealthProbeCache } from "@/utils/workbench-health"; import { agentErrorFromProjection, normalizeApiErrorRecord, normalizeErrorDiagnostic, normalizeProjectionDiagnostic, projectionDiagnosticFromApiFailure, projectionDiagnosticFromFailure } from "@/utils/workbench-error-runtime"; import { readWorkbenchJson, readWorkbenchNumber, readWorkbenchString, removeWorkbenchStorageKey, writeWorkbenchJson, writeWorkbenchString } from "@/utils/workbench-storage-runtime"; import { createWorkbenchStreamTransportRuntime, workbenchRealtimeTraceIdForCapabilities, type WorkbenchRealtimeEvent, type WorkbenchStreamTransportRecovery } from "@/utils/workbench-realtime-runtime"; +import { createWorkbenchReplayVisibilityRuntime } from "@/utils/workbench-replay-visibility"; import { mergeRunnerTrace, type TraceSnapshot } from "@/composables/workbench-trace-snapshot"; import type { AgentChatResponse, AgentChatResultResponse, ApiError, ApiResult, ChatMessage, ErrorDiagnostic, LiveSurface, ProjectionBlocker, ProjectionDiagnostic, ProviderProfile, TraceEvent, WorkbenchSessionRecord, WorkbenchTurnTimingProjection } from "@/types"; import { firstNonEmptyString, nextProtocolId, normalizeWorkbenchSessionId, normalizeWorkbenchSessionRouteId } from "@/utils"; @@ -124,6 +125,7 @@ export const useWorkbenchStore = defineStore("workbench", () => { const turnStatusAuthority = computed(() => selectTurnStatusAuthority(serverState.value)); const traceAuthorityById = computed(() => selectTraceAuthorityById(serverState.value)); const realtimeTransport = createWorkbenchStreamTransportRuntime(); + const replayVisibility = createWorkbenchReplayVisibilityRuntime(); const workbenchProjectionSignalSourceId = nextProtocolId("wbtab"); let workbenchProjectionSignalChannel: BroadcastChannel | null = null; const workbenchHealthProbeCache = createWorkbenchHealthProbeCache({ cacheMs: runtimePolicy.workbenchReadFailureCooldownMs }); @@ -862,6 +864,9 @@ export const useWorkbenchStore = defineStore("workbench", () => { realtimeReplayLoadingSessionId.value = realtimeCapabilities.kafkaRefreshReplay ? sessionId : null; realtimeTransportPhase.value = "connecting"; } + if (realtimeCapabilities.liveKafkaSse && realtimeCapabilities.kafkaRefreshReplay && (scopeChanged || forceReconnect)) { + replayVisibility.begin(realtimeScopeKey); + } if (debugCapabilities.rawHwlabEventWindow.enabled && scopeChanged) rawHwlabIngress.value = createRawHwlabIngressState(realtimeScopeKey); realtimeTransport.restart({ realtimeCapabilities, @@ -876,16 +881,23 @@ export const useWorkbenchStore = defineStore("workbench", () => { ? (frame) => { rawHwlabIngress.value = appendRawHwlabIngressFrame(rawHwlabIngress.value, frame, debugCapabilities.rawHwlabEventWindow); } : undefined, onError: () => { + replayVisibility.discard(realtimeScopeKey); if (realtimeCapabilities.liveKafkaSse) liveRealtimeReadySessionId.value = null; }, onState: (state) => { realtimeTransportPhase.value = state.phase; if (realtimeCapabilities.liveKafkaSse && ["connecting", "error", "closed", "blocked"].includes(state.phase)) liveRealtimeReadySessionId.value = null; + if (["error", "closed", "blocked"].includes(state.phase)) replayVisibility.discard(realtimeScopeKey); }, onRecovery: (recovery) => handleRealtimeRecovery(recovery), onEvent: (event, eventName) => { if (realtimeCapabilities.liveKafkaSse && eventName === "workbench.connected") { const assessment = assessWorkbenchRealtimeConnected(event, sessionId, realtimeCapabilities.kafkaRefreshReplay); + if (assessment.ready) { + for (const frame of replayVisibility.commit(realtimeScopeKey)) applyRealtimeEvent(frame.event, frame.eventName); + } else { + replayVisibility.discard(realtimeScopeKey); + } liveRealtimeReadySessionId.value = assessment.ready ? assessment.connectedSessionId : null; if (assessment.ready && realtimeReplayLoadingSessionId.value === assessment.connectedSessionId) realtimeReplayLoadingSessionId.value = null; if (assessment.warning) { @@ -911,6 +923,7 @@ export const useWorkbenchStore = defineStore("workbench", () => { error.value = null; } } + if (replayVisibility.stage(realtimeScopeKey, { event, eventName })) return; applyRealtimeEvent(event, eventName); } }); diff --git a/web/hwlab-cloud-web/src/utils/workbench-replay-visibility.test.ts b/web/hwlab-cloud-web/src/utils/workbench-replay-visibility.test.ts new file mode 100644 index 00000000..3e70225a --- /dev/null +++ b/web/hwlab-cloud-web/src/utils/workbench-replay-visibility.test.ts @@ -0,0 +1,59 @@ +import { describe, expect, it } from "vitest"; + +import type { WorkbenchRealtimeEvent } from "@/api/workbench-events"; +import { createWorkbenchReplayVisibilityRuntime } from "./workbench-replay-visibility"; + +describe("Workbench replay visibility", () => { + it("keeps the visible projection stable until retained replay reaches live handoff", () => { + const runtime = createWorkbenchReplayVisibilityRuntime(); + let visibleLastEventAt = "2026-07-21T12:00:00.000Z"; + const apply = (event: WorkbenchRealtimeEvent) => { + visibleLastEventAt = String(event.event?.createdAt); + }; + + runtime.begin("session:ses_running"); + expect(runtime.stage("session:ses_running", kafkaFrame("2026-07-21T11:55:00.000Z"))).toBe(true); + expect(runtime.stage("session:ses_running", kafkaFrame("2026-07-21T11:58:00.000Z"))).toBe(true); + expect(runtime.stage("session:ses_running", kafkaFrame("2026-07-21T12:00:01.000Z"))).toBe(true); + + expect(visibleLastEventAt).toBe("2026-07-21T12:00:00.000Z"); + + for (const frame of runtime.commit("session:ses_running")) apply(frame.event); + expect(visibleLastEventAt).toBe("2026-07-21T12:00:01.000Z"); + + expect(runtime.stage("session:ses_running", kafkaFrame("2026-07-21T12:00:03.000Z"))).toBe(false); + apply(kafkaFrame("2026-07-21T12:00:03.000Z").event); + expect(visibleLastEventAt).toBe("2026-07-21T12:00:03.000Z"); + }); + + it("discards a partial replay when its stream is replaced", () => { + const runtime = createWorkbenchReplayVisibilityRuntime(); + runtime.begin("session:ses_old"); + runtime.stage("session:ses_old", kafkaFrame("2026-07-21T11:55:00.000Z")); + + runtime.begin("session:ses_current"); + + expect(runtime.commit("session:ses_old")).toEqual([]); + expect(runtime.commit("session:ses_current")).toEqual([]); + }); + + it("does not delay non-business errors during replay", () => { + const runtime = createWorkbenchReplayVisibilityRuntime(); + runtime.begin("session:ses_running"); + + expect(runtime.stage("session:ses_running", { + eventName: "workbench.error", + event: { type: "error", error: "upstream unavailable" } + })).toBe(false); + }); +}); + +function kafkaFrame(createdAt: string) { + return { + eventName: "hwlab.event.v1", + event: { + schema: "hwlab.event.v1", + event: { traceId: "trc_replay", sessionId: "ses_running", createdAt } + } as WorkbenchRealtimeEvent + }; +} diff --git a/web/hwlab-cloud-web/src/utils/workbench-replay-visibility.ts b/web/hwlab-cloud-web/src/utils/workbench-replay-visibility.ts new file mode 100644 index 00000000..3762d79c --- /dev/null +++ b/web/hwlab-cloud-web/src/utils/workbench-replay-visibility.ts @@ -0,0 +1,47 @@ +// SPEC: PJ2026-010401080313 Workbench realtime authority draft-2026-07-20-p0-kafka-sse-single-projection. +// Responsibility: Keep a retained Kafka replay invisible until its SSE live handoff is complete. + +import type { WorkbenchRealtimeEvent } from "@/api/workbench-events"; + +export interface WorkbenchReplayVisibilityFrame { + event: WorkbenchRealtimeEvent; + eventName: string; +} + +export class WorkbenchReplayVisibilityRuntime { + private scopeKey: string | null = null; + private frames: WorkbenchReplayVisibilityFrame[] = []; + + begin(scopeKey: string): void { + this.scopeKey = scopeKey; + this.frames = []; + } + + stage(scopeKey: string, frame: WorkbenchReplayVisibilityFrame): boolean { + if (scopeKey !== this.scopeKey || !isKafkaBusinessFrame(frame)) return false; + this.frames.push(frame); + return true; + } + + commit(scopeKey: string): WorkbenchReplayVisibilityFrame[] { + if (scopeKey !== this.scopeKey) return []; + const frames = this.frames; + this.scopeKey = null; + this.frames = []; + return frames; + } + + discard(scopeKey?: string | null): void { + if (scopeKey && scopeKey !== this.scopeKey) return; + this.scopeKey = null; + this.frames = []; + } +} + +export function createWorkbenchReplayVisibilityRuntime(): WorkbenchReplayVisibilityRuntime { + return new WorkbenchReplayVisibilityRuntime(); +} + +function isKafkaBusinessFrame(frame: WorkbenchReplayVisibilityFrame): boolean { + return frame.eventName === "hwlab.event.v1" || frame.event.schema === "hwlab.event.v1"; +} diff --git a/web/hwlab-cloud-web/src/utils/workbench-stream-transport.test.ts b/web/hwlab-cloud-web/src/utils/workbench-stream-transport.test.ts new file mode 100644 index 00000000..96d41a7d --- /dev/null +++ b/web/hwlab-cloud-web/src/utils/workbench-stream-transport.test.ts @@ -0,0 +1,85 @@ +// @vitest-environment jsdom + +import { afterEach, beforeEach, describe, expect, it } from "vitest"; + +import type { WorkbenchRealtimeCapabilities } from "@/config/runtime"; +import { WorkbenchStreamTransportRuntime, type WorkbenchStreamTransportState } from "./workbench-stream-transport"; + +const capabilities: WorkbenchRealtimeCapabilities = { + liveKafkaSse: true, + kafkaRefreshReplay: true, + projectionRealtime: false +}; + +describe("WorkbenchStreamTransportRuntime connection authority", () => { + const originalEventSource = globalThis.EventSource; + + beforeEach(() => { + FakeEventSource.instances = []; + globalThis.EventSource = FakeEventSource as unknown as typeof EventSource; + }); + + afterEach(() => { + globalThis.EventSource = originalEventSource; + }); + + it("ignores late callbacks from a replaced SSE connection", () => { + const runtime = new WorkbenchStreamTransportRuntime(); + const states: WorkbenchStreamTransportState[] = []; + const recoveries: string[] = []; + + runtime.restart({ + realtimeCapabilities: capabilities, + sessionId: "ses_old", + onState: (state) => states.push(state), + onRecovery: (recovery) => recoveries.push(recovery.key), + onEvent: () => undefined + }); + const oldSource = FakeEventSource.instances[0]; + expect(oldSource).toBeDefined(); + const lateOpen = oldSource?.onopen; + const lateError = oldSource?.onerror; + + runtime.restart({ + realtimeCapabilities: capabilities, + sessionId: "ses_current", + onState: (state) => states.push(state), + onRecovery: (recovery) => recoveries.push(recovery.key), + onEvent: () => undefined + }); + const currentSource = FakeEventSource.instances[1]; + expect(currentSource).toBeDefined(); + expect(states.at(-1)).toMatchObject({ sessionId: "ses_current", phase: "connecting" }); + + lateOpen?.(new Event("open")); + lateError?.(new Event("error")); + + expect(states.at(-1)).toMatchObject({ sessionId: "ses_current", phase: "connecting" }); + expect(recoveries).toEqual([]); + + currentSource?.emitOpen(); + expect(states.at(-1)).toMatchObject({ sessionId: "ses_current", phase: "open" }); + runtime.stop(); + }); +}); + +class FakeEventSource { + static instances: FakeEventSource[] = []; + onopen: ((event: Event) => void) | null = null; + onerror: ((event: Event) => void) | null = null; + onmessage: ((event: MessageEvent) => void) | null = null; + + constructor(readonly url: string | URL) { + FakeEventSource.instances.push(this); + } + + addEventListener(): void {} + removeEventListener(): void {} + + close(): void {} + + emitOpen(): void { + this.onopen?.(new Event("open")); + } + +} diff --git a/web/hwlab-cloud-web/src/utils/workbench-stream-transport.ts b/web/hwlab-cloud-web/src/utils/workbench-stream-transport.ts index cfb29b6a..eb327d63 100644 --- a/web/hwlab-cloud-web/src/utils/workbench-stream-transport.ts +++ b/web/hwlab-cloud-web/src/utils/workbench-stream-transport.ts @@ -97,6 +97,7 @@ export function workbenchRealtimeTransportEnabled(capabilities: WorkbenchRealtim export class WorkbenchStreamTransportRuntime { private stream: WorkbenchEventStream | null = null; private key = ""; + private generation = 0; private tick = 0; private wait: WorkbenchTransportWait | null = null; private readonly cursorByKey = new Map(); @@ -107,6 +108,7 @@ export class WorkbenchStreamTransportRuntime { if (key === this.key && this.stream && input.forceReconnect !== true) return { key, changed: false, streamStarted: true }; this.stop(); this.key = key; + const generation = this.generation; this.armWait(key); if (!workbenchRealtimeTransportEnabled(input.realtimeCapabilities)) return { key, changed: true, streamStarted: false }; if (!input.sessionId && !input.traceId) return { key, changed: true, streamStarted: false }; @@ -121,17 +123,20 @@ export class WorkbenchStreamTransportRuntime { flushYieldMs: input.flushYieldMs, onIngress: input.onIngress, onOpen: () => { + if (!this.isCurrentStream(key, generation)) return; this.markLive(key); this.emitState(input, "open", null); input.onOpen?.(); }, onError: (event) => { + if (!this.isCurrentStream(key, generation)) return; this.markBlocked(key); this.emitState(input, "error", null, event.type); input.onError?.(event); this.requestRecovery(input, event.type || "eventsource-error"); }, onEvent: (event, eventName) => { + if (!this.isCurrentStream(key, generation)) return; if (!input.realtimeCapabilities.liveKafkaSse && input.realtimeCapabilities.projectionRealtime) this.rememberCursor(key, event); this.markLive(key); this.emitState(input, "event", eventName); @@ -143,6 +148,7 @@ export class WorkbenchStreamTransportRuntime { stop(): void { const previousKey = this.key; + this.generation += 1; this.stream?.close(); this.stream = null; this.key = ""; @@ -157,6 +163,10 @@ export class WorkbenchStreamTransportRuntime { return this.cursorByKey.get(key) ?? { outboxSeq: null, traceSeq: null }; } + private isCurrentStream(key: string, generation: number): boolean { + return this.stream !== null && this.key === key && this.generation === generation; + } + clear(key?: string | null): void { const normalized = normalizeKey(key); if (normalized) {