From 57af108d40a0628bc42114efca87ee65765a36e8 Mon Sep 17 00:00:00 2001 From: root Date: Mon, 20 Jul 2026 04:45:15 +0200 Subject: [PATCH] fix(workbench): show and accelerate Kafka replay --- internal/cloud/kafka-event-bridge.test.ts | 9 +++- internal/cloud/kafka-event-bridge.ts | 47 ++++++++++++------- .../cloud/server-workbench-realtime-http.ts | 3 ++ .../cloud/workbench-kafka-refresh-handoff.ts | 3 ++ internal/workbench/http.ts | 3 ++ internal/workbench/workbench.test.ts | 6 +++ tools/src/workbench-cli.ts | 3 ++ .../src/stores/workbench-session.test.ts | 23 ++++++++- .../src/stores/workbench-session.ts | 13 +++++ web/hwlab-cloud-web/src/stores/workbench.ts | 18 +++++-- 10 files changed, 107 insertions(+), 21 deletions(-) diff --git a/internal/cloud/kafka-event-bridge.test.ts b/internal/cloud/kafka-event-bridge.test.ts index 1077c4f0..33b759e9 100644 --- a/internal/cloud/kafka-event-bridge.test.ts +++ b/internal/cloud/kafka-event-bridge.test.ts @@ -6,7 +6,7 @@ import { test } from "bun:test"; import { createCloudApiBunServer } from "./bun-server.ts"; import { buildCloudApiReadiness } from "./health-contract.ts"; -import { decodeCanonicalAgentRunKafkaMessage, kafkaDnsLookup, kafkaEventBridgeConfig, kafkaPartitionForKey, projectAgentRunKafkaEventToHwlabEvent, publishAgentRunKafkaMessageLive, relayHwlabKafkaOutboxOnce, startHwlabKafkaEventBridge } from "./kafka-event-bridge.ts"; +import { decodeCanonicalAgentRunKafkaMessage, kafkaDnsLookup, kafkaEventBridgeConfig, kafkaMessageKeyCanMatchPartitionKey, kafkaPartitionForKey, projectAgentRunKafkaEventToHwlabEvent, publishAgentRunKafkaMessageLive, relayHwlabKafkaOutboxOnce, startHwlabKafkaEventBridge } from "./kafka-event-bridge.ts"; test("Workbench refresh replay resolves the same Kafka partition as the session-keyed producer", () => { const offsets = [ @@ -23,6 +23,13 @@ test("Workbench refresh replay resolves the same Kafka partition as the session- assert.equal(kafkaPartitionForKey(null, "hwlab.event.v1", offsets), null); }); +test("Workbench refresh replay rejects foreign keyed values before JSON parsing", () => { + assert.equal(kafkaMessageKeyCanMatchPartitionKey("ses_target", "ses_target"), true); + assert.equal(kafkaMessageKeyCanMatchPartitionKey("ses_foreign", "ses_target"), false); + assert.equal(kafkaMessageKeyCanMatchPartitionKey(null, "ses_target"), true); + assert.equal(kafkaMessageKeyCanMatchPartitionKey("ses_target", null), true); +}); + const PROJECTOR_ENV = Object.freeze({ HWLAB_KAFKA_DIRECT_PUBLISH_ENABLED: "false", HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED: "false", diff --git a/internal/cloud/kafka-event-bridge.ts b/internal/cloud/kafka-event-bridge.ts index 0979ea1f..1c16028e 100644 --- a/internal/cloud/kafka-event-bridge.ts +++ b/internal/cloud/kafka-event-bridge.ts @@ -973,6 +973,7 @@ export async function queryKafkaEventStream({ env = process.env, stream = "hwlab const clientId = stringValue(env.HWLAB_KAFKA_CLIENT_ID) || DEFAULT_CLIENT_ID; const resolvedTopic = stringValue(topic) || kafkaTopicForStream(stream, env); const maxEvents = Math.max(1, integerValue(limit) || DEFAULT_QUERY_LIMIT); + const resolvedPartitionKey = stringValue(partitionKey); const maxScannedRecords = scanLimit === null || scanLimit === undefined ? null : strictPositiveInteger(scanLimit, "scanLimit"); @@ -989,7 +990,7 @@ export async function queryKafkaEventStream({ env = process.env, stream = "hwlab timing.endOffsetSnapshotMs = Date.now() - stageStartedAtMs; const endOffsetByPartition = new Map(endOffsetSnapshot.offsets.map((entry) => [entry.partition, entry.endOffset])); const startOffsetByPartition = new Map(endOffsetSnapshot.offsets.map((entry) => [entry.partition, entry.startOffset])); - const targetPartition = kafkaPartitionForKey(partitionKey, resolvedTopic, endOffsetSnapshot.offsets); + const targetPartition = kafkaPartitionForKey(resolvedPartitionKey, resolvedTopic, endOffsetSnapshot.offsets); const skippedPartitions = targetPartition === null ? [] : endOffsetSnapshot.offsets.filter((entry) => entry.partition !== targetPartition).map((entry) => entry.partition); const requiresScopedGroupOffsets = skippedPartitions.length > 0; const reachedEndPartitions = new Set([...skippedPartitions, ...endOffsetSnapshot.offsets.filter(kafkaOffsetPartitionIsEmpty).map((entry) => entry.partition)]); @@ -1009,6 +1010,7 @@ export async function queryKafkaEventStream({ env = process.env, stream = "hwlab let parsedCount = 0; let invalidJsonCount = 0; let filterRejectedCount = 0; + let keyRejectedCount = 0; let excludedPostBarrierCount = 0; let completionReason = "timeout"; let timer = null; @@ -1090,22 +1092,28 @@ export async function queryKafkaEventStream({ env = process.env, stream = "hwlab } scannedCount += 1; lastScannedOffsetByPartition.set(partition, stringValue(message.offset)); - const valueText = message.value ? Buffer.from(message.value).toString("utf8") : ""; - const value = parseJson(valueText); - if (!value || typeof value !== "object" || Array.isArray(value)) invalidJsonCount += 1; - else { - parsedCount += 1; - if (!eventMatchesFilters(value, { traceId, sessionId, runId, commandId })) filterRejectedCount += 1; + const messageKey = message.key ? Buffer.from(message.key).toString("utf8") : null; + if (!kafkaMessageKeyCanMatchPartitionKey(messageKey, resolvedPartitionKey)) { + filterRejectedCount += 1; + keyRejectedCount += 1; + } else { + const valueText = message.value ? Buffer.from(message.value).toString("utf8") : ""; + const value = parseJson(valueText); + if (!value || typeof value !== "object" || Array.isArray(value)) invalidJsonCount += 1; else { - events.push({ - topic: messageTopic, - partition, - offset: message.offset, - key: message.key ? Buffer.from(message.key).toString("utf8") : null, - timestamp: message.timestamp ?? null, - valueSha256: sha256(valueText), - value - }); + parsedCount += 1; + if (!eventMatchesFilters(value, { traceId, sessionId, runId, commandId })) filterRejectedCount += 1; + else { + events.push({ + topic: messageTopic, + partition, + offset: message.offset, + key: messageKey, + timestamp: message.timestamp ?? null, + valueSha256: sha256(valueText), + value + }); + } } } if (fromBeginning !== false && endOffset && kafkaOffsetReached(message.offset, endOffset)) reachedEndPartitions.add(partition); @@ -1153,6 +1161,7 @@ export async function queryKafkaEventStream({ env = process.env, stream = "hwlab matchedCount: events.length, invalidJsonCount, filterRejectedCount, + keyRejectedCount, excludedPostBarrierCount, completionReason, completion: { @@ -1190,6 +1199,12 @@ export async function queryKafkaEventStream({ env = process.env, stream = "hwlab } } +export function kafkaMessageKeyCanMatchPartitionKey(messageKey, partitionKey) { + const expected = stringValue(partitionKey); + const actual = stringValue(messageKey); + return !expected || !actual || actual === expected; +} + export function kafkaPartitionForKey(partitionKey, topic, offsets = []) { const key = stringValue(partitionKey); const partitionMetadata = offsets diff --git a/internal/cloud/server-workbench-realtime-http.ts b/internal/cloud/server-workbench-realtime-http.ts index cd32aa3f..359715ab 100644 --- a/internal/cloud/server-workbench-realtime-http.ts +++ b/internal/cloud/server-workbench-realtime-http.ts @@ -778,7 +778,10 @@ async function handleKafkaRefreshReplayWorkbenchRealtimeHttp(request, response, completionReason: result?.completionReason ?? null, complete: result?.completion?.complete === true, scannedCount: result?.scannedCount ?? null, + parsedCount: result?.parsedCount ?? null, matchedCount: result?.matchedCount ?? null, + filterRejectedCount: result?.filterRejectedCount ?? null, + keyRejectedCount: result?.keyRejectedCount ?? null, partitionKeyScoped: result?.partitionKeyScoped === true, targetPartition: result?.targetPartition ?? null, timing: result?.timing ?? null, diff --git a/internal/cloud/workbench-kafka-refresh-handoff.ts b/internal/cloud/workbench-kafka-refresh-handoff.ts index ddcf1b75..b0d70e88 100644 --- a/internal/cloud/workbench-kafka-refresh-handoff.ts +++ b/internal/cloud/workbench-kafka-refresh-handoff.ts @@ -403,7 +403,10 @@ function boundedQueryDiagnostics(result) { completionReason: textValue(result?.completion?.reason ?? result?.completionReason), complete: result?.completion?.complete === true, scannedCount: numericValue(result?.scannedCount), + parsedCount: numericValue(result?.parsedCount), matchedCount: numericValue(result?.matchedCount), + filterRejectedCount: numericValue(result?.filterRejectedCount), + keyRejectedCount: numericValue(result?.keyRejectedCount), partitionKeyScoped: result?.partitionKeyScoped === true, targetPartition: partitionValue(result?.targetPartition), timing: { diff --git a/internal/workbench/http.ts b/internal/workbench/http.ts index fd1ddaf8..2c664286 100644 --- a/internal/workbench/http.ts +++ b/internal/workbench/http.ts @@ -170,7 +170,10 @@ function nativeKafkaRefreshEventStream(bridge: any, sessionId: string, traceId: completionReason: result?.completionReason ?? null, complete: result?.completion?.complete === true, scannedCount: result?.scannedCount ?? null, + parsedCount: result?.parsedCount ?? null, matchedCount: result?.matchedCount ?? null, + filterRejectedCount: result?.filterRejectedCount ?? null, + keyRejectedCount: result?.keyRejectedCount ?? null, partitionKeyScoped: result?.partitionKeyScoped === true, targetPartition: result?.targetPartition ?? null, timing: result?.timing ?? null, diff --git a/internal/workbench/workbench.test.ts b/internal/workbench/workbench.test.ts index a06cbeca..863516a4 100644 --- a/internal/workbench/workbench.test.ts +++ b/internal/workbench/workbench.test.ts @@ -185,7 +185,10 @@ describe("Workbench native HTTP adapter", () => { endOffsetsAvailable: true, endOffsets: [{ partition: 0, startOffset: "0", endOffset: "100" }], scannedCount: 100, + parsedCount: 100, matchedCount: 100, + filterRejectedCount: 0, + keyRejectedCount: 0, partitionKeyScoped: true, targetPartition: 0, timing: { endOffsetSnapshotMs: 2, groupOffsetsMs: 0, consumerConnectMs: 3, consumerSubscribeMs: 1, scanMs: 7, cleanupMs: 1, totalMs: 14 } @@ -222,7 +225,10 @@ describe("Workbench native HTTP adapter", () => { query: { complete: true, scannedCount: 100, + parsedCount: 100, matchedCount: 100, + filterRejectedCount: 0, + keyRejectedCount: 0, partitionKeyScoped: true, targetPartition: 0, timing: { scanMs: 7, totalMs: 14 } diff --git a/tools/src/workbench-cli.ts b/tools/src/workbench-cli.ts index c9744adf..f97c95a4 100644 --- a/tools/src/workbench-cli.ts +++ b/tools/src/workbench-cli.ts @@ -178,7 +178,10 @@ function workbenchConnectedContract(data: Record) { completionReason: String(query.completionReason ?? "").trim() || null, complete: query.complete === true, scannedCount: nullableNumber(query.scannedCount), + parsedCount: nullableNumber(query.parsedCount), matchedCount: nullableNumber(query.matchedCount), + filterRejectedCount: nullableNumber(query.filterRejectedCount), + keyRejectedCount: nullableNumber(query.keyRejectedCount), partitionKeyScoped: query.partitionKeyScoped === true, targetPartition: nullableNumber(query.targetPartition), timing: queryTiming ? { diff --git a/web/hwlab-cloud-web/src/stores/workbench-session.test.ts b/web/hwlab-cloud-web/src/stores/workbench-session.test.ts index 86d69f48..a254ae6c 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-session.test.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-session.test.ts @@ -2,7 +2,7 @@ import assert from "node:assert/strict"; import { test } from "bun:test"; import type { ChatMessage, WorkbenchSessionRecord } from "../types"; -import { resolveCancelableAgentMessage, resolveComposerState, sessionToSessionTab, shouldReadTerminalTraceDetail } from "./workbench-session"; +import { resolveCancelableAgentMessage, resolveComposerState, sessionToSessionTab, shouldReadTerminalTraceDetail, shouldShowSessionDetailLoading } from "./workbench-session"; import type { TurnStatusAuthority } from "./workbench-session"; function agentMessage(input: Partial & Pick): ChatMessage { @@ -76,6 +76,27 @@ test("composer blocks turn submission until the live session stream is connected assert.equal(connected.disabled, false); }); +test("session detail keeps a visible loading state until Kafka replay connects", () => { + assert.equal(shouldShowSessionDetailLoading({ + loading: false, + messageCount: 0, + replayLoadingSessionId: "ses_refresh", + activeSessionId: "ses_refresh" + }), true); + assert.equal(shouldShowSessionDetailLoading({ + loading: false, + messageCount: 2, + replayLoadingSessionId: null, + activeSessionId: "ses_refresh" + }), false); + assert.equal(shouldShowSessionDetailLoading({ + loading: false, + messageCount: 0, + replayLoadingSessionId: "ses_previous", + activeSessionId: "ses_refresh" + }), false); +}); + test("cancel target remains available for completed message until final response exists", () => { const traceId = "trc_cancel_completed_without_final"; const sessionId = "ses_cancel_completed_without_final"; diff --git a/web/hwlab-cloud-web/src/stores/workbench-session.ts b/web/hwlab-cloud-web/src/stores/workbench-session.ts index 48b31726..bffad4dd 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-session.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-session.ts @@ -130,6 +130,19 @@ export function shouldShowSessionListLoading(input: { loading: boolean; sessions return input.loading && !input.sessionsReady; } +export function shouldShowSessionDetailLoading(input: { + loading: boolean; + messageCount: number; + sessionDetailLoadingId?: string | null; + switchingSessionId?: string | null; + replayLoadingSessionId?: string | null; + activeSessionId?: string | null; +}): boolean { + if (input.sessionDetailLoadingId || input.switchingSessionId) return true; + if (input.loading && input.messageCount === 0) return true; + return Boolean(input.replayLoadingSessionId && input.replayLoadingSessionId === input.activeSessionId); +} + export function sessionToSessionTab(session: WorkbenchSessionRecord, activeSessionId: string | null, sessionStatusAuthority: SessionStatusAuthorityMap = {}): SessionTab { const sessionId = session.sessionId; const updatedAt = sessionDisplayUpdatedAt(session); diff --git a/web/hwlab-cloud-web/src/stores/workbench.ts b/web/hwlab-cloud-web/src/stores/workbench.ts index 1ce81a45..8a3b9ec9 100644 --- a/web/hwlab-cloud-web/src/stores/workbench.ts +++ b/web/hwlab-cloud-web/src/stores/workbench.ts @@ -16,7 +16,7 @@ import type { AgentChatResponse, AgentChatResultResponse, ApiError, ApiResult, C import { firstNonEmptyString, nextProtocolId, normalizeWorkbenchSessionId, normalizeWorkbenchSessionRouteId } from "@/utils"; import { composeWorkbenchScopedKey, workbenchRealtimeScopeKey } from "@/utils/workbench-key"; import { failWorkbenchSessionSwitch, failWorkbenchSubmitJourney, finishWorkbenchSessionSwitchFullLoad, markWorkbenchSubmitApiAccepted, markWorkbenchTraceEventsReceived, markWorkbenchTraceProjected, recordWorkbenchLoadingState, recordWorkbenchRuntimeDiagnostic, startWorkbenchSessionSwitch, startWorkbenchSubmitJourney } from "@/utils/workbench-performance"; -import { RECENT_DRAFTS_STORAGE_KEY, appendSessionPage, defaultProviderProfileOptions, isArchivedSession, mergeSessionIntoList, normalizeChatMessageStatus, normalizeRecentDrafts, normalizeWorkbenchMessageTitle, preferredProviderProfile, providerProfileOptionsFromPayload, recordRecentDraft, resolveCancelableAgentMessage, resolveComposerState, shouldReadTerminalTraceDetail, shouldShowSessionListLoading, sortSessionTabs, stableSessionList, type DraftEntry, type ProviderProfileOption, type TurnStatusAuthority } from "./workbench-session"; +import { RECENT_DRAFTS_STORAGE_KEY, appendSessionPage, defaultProviderProfileOptions, isArchivedSession, mergeSessionIntoList, normalizeChatMessageStatus, normalizeRecentDrafts, normalizeWorkbenchMessageTitle, preferredProviderProfile, providerProfileOptionsFromPayload, recordRecentDraft, resolveCancelableAgentMessage, resolveComposerState, shouldReadTerminalTraceDetail, shouldShowSessionDetailLoading, shouldShowSessionListLoading, sortSessionTabs, stableSessionList, type DraftEntry, type ProviderProfileOption, type TurnStatusAuthority } from "./workbench-session"; 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"; @@ -113,6 +113,7 @@ export const useWorkbenchStore = defineStore("workbench", () => { const activityRef = ref({ lastActivityAt: Date.now(), lastActivityIso: new Date().toISOString(), waitingFor: "idle", lastEventLabel: null as string | null }); const currentRequest = ref<{ traceId: string; sessionId: string | null; threadId: string | null; status?: string | null } | null>(null); const liveRealtimeReadySessionId = ref(null); + const realtimeReplayLoadingSessionId = ref(null); const explicitSessionId = ref(initialWorkbenchSessionIdFromLocation()); const activeSelectionSource = ref(explicitSessionId.value ? "route" : "system"); const selectionEpoch = ref(0); @@ -138,7 +139,14 @@ export const useWorkbenchStore = defineStore("workbench", () => { const sessionTabs = computed(() => sortSessionTabs(routeActiveSession.value ? mergeSessionIntoList(sessions.value, routeActiveSession.value) : sessions.value, activeSessionId.value, sessionStatusAuthority.value)); const sessionListLoadedCount = computed(() => sessionTabs.value.length); const sessionListLoading = computed(() => shouldShowSessionListLoading({ loading: loading.value, sessionsReady: sessionsReady.value })); - const sessionDetailLoading = computed(() => Boolean(sessionDetailLoadingId.value || switchingSessionId.value || (loading.value && messages.value.length === 0))); + const sessionDetailLoading = computed(() => shouldShowSessionDetailLoading({ + loading: loading.value, + messageCount: messages.value.length, + sessionDetailLoadingId: sessionDetailLoadingId.value, + switchingSessionId: switchingSessionId.value, + replayLoadingSessionId: realtimeReplayLoadingSessionId.value, + activeSessionId: activeSessionId.value + })); const composer = computed(() => resolveComposerState({ messages: messages.value, sessions: sessions.value, activeSessionId: activeSessionId.value, chatPending: chatPending.value, realtimeReady: !realtimeCapabilities.liveKafkaSse || liveRealtimeReadySessionId.value === activeSessionId.value, currentRequest: currentRequest.value, turnStatusAuthority: turnStatusAuthority.value })); function recordActivity(label = "user-activity"): void { @@ -839,7 +847,10 @@ export const useWorkbenchStore = defineStore("workbench", () => { void reason; const realtimeScopeKey = workbenchRealtimeScopeKey(sessionId, traceId); const scopeChanged = realtimeTransport.currentKey() !== realtimeScopeKey; - if (realtimeCapabilities.liveKafkaSse && scopeChanged) liveRealtimeReadySessionId.value = null; + if (realtimeCapabilities.liveKafkaSse && scopeChanged) { + liveRealtimeReadySessionId.value = null; + realtimeReplayLoadingSessionId.value = realtimeCapabilities.kafkaRefreshReplay ? sessionId : null; + } if (debugCapabilities.rawHwlabEventWindow.enabled && scopeChanged) rawHwlabIngress.value = createRawHwlabIngressState(realtimeScopeKey); realtimeTransport.restart({ realtimeCapabilities, @@ -864,6 +875,7 @@ export const useWorkbenchStore = defineStore("workbench", () => { if (realtimeCapabilities.liveKafkaSse && eventName === "workbench.connected") { const assessment = assessWorkbenchRealtimeConnected(event, sessionId, realtimeCapabilities.kafkaRefreshReplay); liveRealtimeReadySessionId.value = assessment.ready ? assessment.connectedSessionId : null; + if (assessment.ready && realtimeReplayLoadingSessionId.value === assessment.connectedSessionId) realtimeReplayLoadingSessionId.value = null; if (assessment.warning) { recordWorkbenchRuntimeDiagnostic({ module: "workbench-stream-contract",