From c7ff4733c0478c20867b8ef267ab4b35bda1292a Mon Sep 17 00:00:00 2001 From: root Date: Mon, 20 Jul 2026 05:24:00 +0200 Subject: [PATCH] =?UTF-8?q?fix(workbench):=20=E7=BB=9F=E4=B8=80=20Kafka=20?= =?UTF-8?q?=E5=9B=9E=E6=94=BE=E6=97=B6=E9=92=9F=E5=B9=B6=E5=8A=A0=E9=80=9F?= =?UTF-8?q?=E4=BC=9A=E8=AF=9D=E5=8A=A0=E8=BD=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/cloud/kafka-event-bridge.test.ts | 26 + internal/cloud/kafka-event-bridge.ts | 122 ++++- .../cloud/server-workbench-realtime-http.ts | 1 + .../cloud/workbench-kafka-refresh-handoff.ts | 12 + .../workbench-kafka-session-index.test.ts | 118 +++++ .../cloud/workbench-kafka-session-index.ts | 473 ++++++++++++++++++ internal/workbench/http.ts | 1 + internal/workbench/workbench.test.ts | 6 +- tools/src/workbench-cli.ts | 12 + .../stores/workbench-live-kafka-event.test.ts | 58 ++- .../src/stores/workbench-live-kafka-event.ts | 26 +- .../src/stores/workbench-session.test.ts | 25 + .../src/stores/workbench-session.ts | 4 +- web/hwlab-cloud-web/src/stores/workbench.ts | 34 +- 14 files changed, 882 insertions(+), 36 deletions(-) create mode 100644 internal/cloud/workbench-kafka-session-index.test.ts create mode 100644 internal/cloud/workbench-kafka-session-index.ts diff --git a/internal/cloud/kafka-event-bridge.test.ts b/internal/cloud/kafka-event-bridge.test.ts index 33b759e9..8fe4de74 100644 --- a/internal/cloud/kafka-event-bridge.test.ts +++ b/internal/cloud/kafka-event-bridge.test.ts @@ -30,6 +30,32 @@ test("Workbench refresh replay rejects foreign keyed values before JSON parsing" assert.equal(kafkaMessageKeyCanMatchPartitionKey("ses_target", null), true); }); +test("Workbench session index uses explicit budgets and defaults missing non-core values with warnings", () => { + const explicit = kafkaEventBridgeConfig({ + ...LIVE_REFRESH_ENV, + HWLAB_WORKBENCH_KAFKA_REFRESH_INDEX_ENABLED: "true", + HWLAB_WORKBENCH_KAFKA_REFRESH_INDEX_MAX_EVENTS: "100000", + HWLAB_WORKBENCH_KAFKA_REFRESH_INDEX_MAX_EVENTS_PER_SESSION: "10000", + HWLAB_WORKBENCH_KAFKA_REFRESH_INDEX_BOOTSTRAP_TIMEOUT_MS: "30000", + HWLAB_WORKBENCH_KAFKA_REFRESH_INDEX_REBUILD_INTERVAL_MS: "300000", + HWLAB_WORKBENCH_KAFKA_REFRESH_INDEX_REBUILD_COOLDOWN_MS: "30000" + }); + assert.deepEqual(explicit.refreshReplay.sessionIndex, { + enabled: true, + maxEvents: 100000, + maxEventsPerSession: 10000, + bootstrapTimeoutMs: 30000, + rebuildIntervalMs: 300000, + rebuildCooldownMs: 30000, + warnings: [], + valuesRedacted: true + }); + + const missing = kafkaEventBridgeConfig(LIVE_REFRESH_ENV); + assert.equal(missing.refreshReplay.sessionIndex.enabled, false); + assert.equal(missing.refreshReplay.sessionIndex.warnings[0].blocking, false); +}); + 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 1c16028e..b358f711 100644 --- a/internal/cloud/kafka-event-bridge.ts +++ b/internal/cloud/kafka-event-bridge.ts @@ -9,6 +9,7 @@ import tls from "node:tls"; import { Kafka, logLevel, Partitioners } from "kafkajs"; import { emitCodeAgentOtelSpan } from "./otel-trace.ts"; +import { createWorkbenchKafkaSessionIndex } from "./workbench-kafka-session-index.ts"; import { buildWorkbenchProjectionEventFacts } from "./workbench-projection-writer.ts"; import { WORKBENCH_REALTIME_CAPABILITY_ENVS, @@ -26,6 +27,11 @@ export const AGENTRUN_STDIO_RECONSTRUCTION_KIND = "stdio-derived partial reconst const DEFAULT_CLIENT_ID = "hwlab-v03-cloud-api"; const DEFAULT_QUERY_TIMEOUT_MS = 5000; const DEFAULT_QUERY_LIMIT = 50; +const DEFAULT_SESSION_INDEX_MAX_EVENTS = 100000; +const DEFAULT_SESSION_INDEX_MAX_EVENTS_PER_SESSION = 10000; +const DEFAULT_SESSION_INDEX_BOOTSTRAP_TIMEOUT_MS = 30000; +const DEFAULT_SESSION_INDEX_REBUILD_INTERVAL_MS = 300000; +const DEFAULT_SESSION_INDEX_REBUILD_COOLDOWN_MS = 30000; const LIVE_KAFKA_COMMAND_OUTPUT_OTEL_SAMPLE_MODULUS = 32; const REQUIRED_KAFKA_ENV = Object.freeze([ "HWLAB_KAFKA_BOOTSTRAP_SERVERS", @@ -89,7 +95,8 @@ export function kafkaEventBridgeConfig(env = process.env) { timeoutMs: strictPositiveInteger(env.HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_TIMEOUT_MS, "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_TIMEOUT_MS"), scanLimit: strictPositiveInteger(env.HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_SCAN_LIMIT, "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_SCAN_LIMIT"), matchedEventLimit: strictPositiveInteger(env.HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_MATCHED_EVENT_LIMIT, "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_MATCHED_EVENT_LIMIT"), - liveBufferLimit: strictPositiveInteger(env.HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_LIVE_BUFFER_LIMIT, "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_LIVE_BUFFER_LIMIT") + liveBufferLimit: strictPositiveInteger(env.HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_LIVE_BUFFER_LIMIT, "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_LIVE_BUFFER_LIMIT"), + sessionIndex: workbenchKafkaSessionIndexConfig(env) } : null; if (relay && relay.sendTimeoutMs + 5000 >= relay.leaseMs) { const error = new Error("HWLAB Kafka relay lease must exceed producer send timeout by at least 5000ms."); @@ -273,6 +280,26 @@ function startLiveHwlabKafkaEventBridge({ config, env = process.env, logger = co let startupReady = false; const subscribers = new Set(); const lagByPartition = new Map(); + const sessionIndex = config.refreshReplay?.sessionIndex?.enabled === true + ? createWorkbenchKafkaSessionIndex({ + topic: config.hwlabTopic, + ...config.refreshReplay.sessionIndex, + scanLimit: config.refreshReplay.scanLimit, + logger, + queryAll: (params) => queryKafkaEventStream({ + ...params, + env, + stream: "hwlab", + topic: config.hwlabTopic, + fromBeginning: true, + groupIdPrefix: `${config.refreshReplay.groupIdPrefix}-session-index`, + kafkaFactory + }) + }) + : null; + for (const warning of config.refreshReplay?.sessionIndex?.warnings ?? []) { + logWarn(logger, "workbench-kafka-session-index-config-warning", warning); + } const publishLiveEnvelope = (envelope, transport) => { for (const listener of [...subscribers]) { try { @@ -306,6 +333,7 @@ function startLiveHwlabKafkaEventBridge({ config, env = process.env, logger = co } else { const transport = { topic: batch.topic, partition: batch.partition, offset: message.offset }; emitLiveKafkaOtelSpan("hwlab.kafka.live.fanout_receive", envelope, transport, { env, otelSpanEmitter }); + sessionIndex?.observeLive(envelope, transport); publishLiveEnvelope(envelope, transport); } resolveOffset(message.offset); @@ -321,9 +349,11 @@ function startLiveHwlabKafkaEventBridge({ config, env = process.env, logger = co await publishAgentRunKafkaMessageLive({ topic, partition, message }, { producer, config, env, logger, otelSpanEmitter }); } }); + void sessionIndex?.start(); startupReady = true; logInfo(logger, "hwlab-live-kafka-event-bridge-started", { capabilities: config.capabilities, + sessionIndex: sessionIndex?.status() ?? { enabled: false, blocking: false, valuesRedacted: true }, agentRunTopic: config.agentRunTopic, hwlabTopic: config.hwlabTopic, clientId: config.clientId, @@ -349,12 +379,13 @@ function startLiveHwlabKafkaEventBridge({ config, env = process.env, logger = co async stop() { stopped = true; subscribers.clear(); + sessionIndex?.stop(); await Promise.allSettled([bridgeConsumer?.stop?.(), fanoutConsumer?.stop?.()]); await Promise.allSettled([bridgeConsumer?.disconnect?.(), fanoutConsumer?.disconnect?.(), producer?.disconnect?.()]); }, async status() { if (startupError) return { status: "blocked", capabilities: config.capabilities, errorCode: startupError.code ?? "hwlab_live_kafka_start_failed", message: errorMessage(startupError), valuesRedacted: true }; - return { status: startupReady ? "ready" : "initializing", capabilities: config.capabilities, subscriberCount: subscribers.size, consumerLag: liveKafkaLagSummary(lagByPartition), lossPossible: true, valuesRedacted: true }; + return { status: startupReady ? "ready" : "initializing", capabilities: config.capabilities, subscriberCount: subscribers.size, consumerLag: liveKafkaLagSummary(lagByPartition), sessionIndex: sessionIndex?.status() ?? { enabled: false, blocking: false, valuesRedacted: true }, lossPossible: true, valuesRedacted: true }; }, startupStatus() { if (startupError) return { status: "blocked", capabilities: config.capabilities, errorCode: startupError.code ?? "hwlab_live_kafka_start_failed", message: errorMessage(startupError), valuesRedacted: true }; @@ -366,8 +397,24 @@ function startLiveHwlabKafkaEventBridge({ config, env = process.env, logger = co subscribers.add(listener); return () => subscribers.delete(listener); }, - queryHwlabEventRetention(params = {}) { - return queryKafkaEventStream({ ...params, env, stream: "hwlab", topic: config.hwlabTopic, kafkaFactory }); + async queryHwlabEventRetention(params = {}) { + const indexed = sessionIndex?.query(params) ?? { + hit: false, + fallbackReason: "index-disabled", + status: { enabled: false, ready: false, blocking: false, valuesRedacted: true } + }; + if (indexed.hit) return indexed.result; + const result = await queryKafkaEventStream({ ...params, env, stream: "hwlab", topic: config.hwlabTopic, kafkaFactory }); + return { + ...result, + index: { + hit: false, + fallbackReason: indexed.fallbackReason, + status: indexed.status, + warning: indexed.warning ?? null, + valuesRedacted: true + } + }; }, liveSubscriberCount() { return subscribers.size; }, valuesPrinted: false @@ -1857,6 +1904,73 @@ function explicitBoolean(value) { return typeof value === "boolean" ? value : null; } +function workbenchKafkaSessionIndexConfig(env) { + const enabledValue = stringValue(env.HWLAB_WORKBENCH_KAFKA_REFRESH_INDEX_ENABLED); + if (!enabledValue) { + return { + enabled: false, + warnings: [{ + code: "workbench_kafka_session_index_config_missing", + field: "HWLAB_WORKBENCH_KAFKA_REFRESH_INDEX_ENABLED", + blocking: false, + valuesPrinted: false + }], + valuesRedacted: true + }; + } + const enabled = truthy(env.HWLAB_WORKBENCH_KAFKA_REFRESH_INDEX_ENABLED); + if (!enabled) return { enabled: false, warnings: [], valuesRedacted: true }; + const warnings = []; + return { + enabled: true, + maxEvents: nonBlockingPositiveInteger( + env.HWLAB_WORKBENCH_KAFKA_REFRESH_INDEX_MAX_EVENTS, + "HWLAB_WORKBENCH_KAFKA_REFRESH_INDEX_MAX_EVENTS", + DEFAULT_SESSION_INDEX_MAX_EVENTS, + warnings + ), + maxEventsPerSession: nonBlockingPositiveInteger( + env.HWLAB_WORKBENCH_KAFKA_REFRESH_INDEX_MAX_EVENTS_PER_SESSION, + "HWLAB_WORKBENCH_KAFKA_REFRESH_INDEX_MAX_EVENTS_PER_SESSION", + DEFAULT_SESSION_INDEX_MAX_EVENTS_PER_SESSION, + warnings + ), + bootstrapTimeoutMs: nonBlockingPositiveInteger( + env.HWLAB_WORKBENCH_KAFKA_REFRESH_INDEX_BOOTSTRAP_TIMEOUT_MS, + "HWLAB_WORKBENCH_KAFKA_REFRESH_INDEX_BOOTSTRAP_TIMEOUT_MS", + DEFAULT_SESSION_INDEX_BOOTSTRAP_TIMEOUT_MS, + warnings + ), + rebuildIntervalMs: nonBlockingPositiveInteger( + env.HWLAB_WORKBENCH_KAFKA_REFRESH_INDEX_REBUILD_INTERVAL_MS, + "HWLAB_WORKBENCH_KAFKA_REFRESH_INDEX_REBUILD_INTERVAL_MS", + DEFAULT_SESSION_INDEX_REBUILD_INTERVAL_MS, + warnings + ), + rebuildCooldownMs: nonBlockingPositiveInteger( + env.HWLAB_WORKBENCH_KAFKA_REFRESH_INDEX_REBUILD_COOLDOWN_MS, + "HWLAB_WORKBENCH_KAFKA_REFRESH_INDEX_REBUILD_COOLDOWN_MS", + DEFAULT_SESSION_INDEX_REBUILD_COOLDOWN_MS, + warnings + ), + warnings, + valuesRedacted: true + }; +} + +function nonBlockingPositiveInteger(value, name, fallback, warnings) { + const parsed = Number(value); + if (Number.isInteger(parsed) && parsed > 0) return parsed; + warnings.push({ + code: "workbench_kafka_session_index_config_defaulted", + field: name, + fallback, + blocking: false, + valuesPrinted: false + }); + return fallback; +} + function nextKafkaOffset(value) { try { return (BigInt(String(value)) + 1n).toString(); diff --git a/internal/cloud/server-workbench-realtime-http.ts b/internal/cloud/server-workbench-realtime-http.ts index 359715ab..a8246ba9 100644 --- a/internal/cloud/server-workbench-realtime-http.ts +++ b/internal/cloud/server-workbench-realtime-http.ts @@ -784,6 +784,7 @@ async function handleKafkaRefreshReplayWorkbenchRealtimeHttp(request, response, keyRejectedCount: result?.keyRejectedCount ?? null, partitionKeyScoped: result?.partitionKeyScoped === true, targetPartition: result?.targetPartition ?? null, + index: result?.index ?? null, timing: result?.timing ?? null, valuesPrinted: false })); diff --git a/internal/cloud/workbench-kafka-refresh-handoff.ts b/internal/cloud/workbench-kafka-refresh-handoff.ts index b0d70e88..f609dd1f 100644 --- a/internal/cloud/workbench-kafka-refresh-handoff.ts +++ b/internal/cloud/workbench-kafka-refresh-handoff.ts @@ -399,6 +399,7 @@ export function createWorkbenchKafkaRefreshHandoff(options = {}) { function boundedQueryDiagnostics(result) { const timing = result?.timing && typeof result.timing === "object" ? result.timing : {}; + const index = result?.index && typeof result.index === "object" ? result.index : {}; return { completionReason: textValue(result?.completion?.reason ?? result?.completionReason), complete: result?.completion?.complete === true, @@ -409,6 +410,16 @@ function boundedQueryDiagnostics(result) { keyRejectedCount: numericValue(result?.keyRejectedCount), partitionKeyScoped: result?.partitionKeyScoped === true, targetPartition: partitionValue(result?.targetPartition), + index: { + hit: index.hit === true, + source: textValue(index.source), + fallbackReason: textValue(index.fallbackReason), + indexedEventCount: numericValue(index.indexedEventCount ?? index.status?.indexedEventCount), + indexedSessionCount: numericValue(index.indexedSessionCount ?? index.status?.indexedSessionCount), + incompleteSessionCount: numericValue(index.incompleteSessionCount ?? index.status?.incompleteSessionCount), + blocking: false, + valuesRedacted: true + }, timing: { endOffsetSnapshotMs: numericValue(timing.endOffsetSnapshotMs), groupOffsetsMs: numericValue(timing.groupOffsetsMs), @@ -416,6 +427,7 @@ function boundedQueryDiagnostics(result) { consumerSubscribeMs: numericValue(timing.consumerSubscribeMs), scanMs: numericValue(timing.scanMs), cleanupMs: numericValue(timing.cleanupMs), + indexLookupMs: numericValue(timing.indexLookupMs), totalMs: numericValue(timing.totalMs), valuesPrinted: false }, diff --git a/internal/cloud/workbench-kafka-session-index.test.ts b/internal/cloud/workbench-kafka-session-index.test.ts new file mode 100644 index 00000000..5d0018f7 --- /dev/null +++ b/internal/cloud/workbench-kafka-session-index.test.ts @@ -0,0 +1,118 @@ +import assert from "node:assert/strict"; +import { test } from "bun:test"; + +import { createWorkbenchKafkaSessionIndex } from "./workbench-kafka-session-index.ts"; + +test("process session index bootstraps once and serves repeated scoped queries from memory", async () => { + let queryCount = 0; + let now = 1000; + const index = createWorkbenchKafkaSessionIndex({ + topic: "hwlab.event.v1", + maxEvents: 20, + maxEventsPerSession: 10, + bootstrapTimeoutMs: 5000, + rebuildIntervalMs: 60000, + rebuildCooldownMs: 1000, + scanLimit: 100, + nowMs: () => now++, + logger: quietLogger(), + queryAll: async () => { + queryCount += 1; + return completeResult([ + record(0, "ses_a", "trc_a"), + record(1, "ses_b", "trc_b"), + record(2, "ses_a", "trc_a") + ], "3"); + } + }); + + assert.equal(await index.start(), true); + const first = index.query({ sessionId: "ses_a", partitionKey: "ses_a", limit: 10, scanLimit: 100, timeoutMs: 5000 }); + const second = index.query({ sessionId: "ses_a", partitionKey: "ses_a", limit: 10, scanLimit: 100, timeoutMs: 5000 }); + assert.equal(first.hit, true); + assert.equal(second.hit, true); + assert.equal(first.result.events.length, 2); + assert.equal(first.result.scannedCount, 2); + assert.equal(first.result.index.hit, true); + assert.equal(first.result.index.indexedEventCount, 3); + assert.equal(queryCount, 1); + + index.observeLive(envelope("ses_a", "trc_a", 4), { topic: "hwlab.event.v1", partition: 0, offset: "3" }); + const live = index.query({ sessionId: "ses_a", limit: 10 }); + assert.equal(live.hit, true); + assert.equal(live.result.events.length, 3); + assert.equal(live.result.endOffsets[0].endOffset, "4"); + index.stop(); +}); + +test("process session index falls back instead of returning an incomplete bounded bucket", async () => { + const index = createWorkbenchKafkaSessionIndex({ + topic: "hwlab.event.v1", + maxEvents: 10, + maxEventsPerSession: 1, + bootstrapTimeoutMs: 5000, + rebuildIntervalMs: 60000, + rebuildCooldownMs: 1000, + scanLimit: 100, + logger: quietLogger(), + queryAll: async () => completeResult([ + record(0, "ses_overflow", "trc_overflow"), + record(1, "ses_overflow", "trc_overflow") + ], "2") + }); + + assert.equal(await index.start(), true); + const result = index.query({ sessionId: "ses_overflow", limit: 10 }); + assert.equal(result.hit, false); + assert.equal(result.fallbackReason, "session-capacity-exceeded"); + assert.equal(result.warning.blocking, false); + index.stop(); +}); + +function completeResult(events: ReturnType[], endOffset: string) { + return { + completionReason: "end-offset", + completion: { + complete: true, + barrierReached: true, + retentionStartVerified: true + }, + reachedEndOffsets: true, + endOffsetsAvailable: true, + endOffsets: [{ partition: 0, startOffset: "0", endOffset }], + events + }; +} + +function record(offset: number, sessionId: string, traceId: string) { + const value = envelope(sessionId, traceId, offset + 1); + return { + topic: "hwlab.event.v1", + partition: 0, + offset: String(offset), + key: sessionId, + timestamp: value.event.createdAt, + valueSha256: `sha-${offset}`, + value + }; +} + +function envelope(sessionId: string, traceId: string, seq: number) { + return { + schema: "hwlab.event.v1", + eventId: `evt_${sessionId}_${seq}`, + sourceEventId: `src_${sessionId}_${seq}`, + sessionId, + traceId, + event: { + type: seq === 1 ? "user" : "assistant", + sessionId, + traceId, + createdAt: `2026-07-20T00:00:0${seq}.000Z` + } + }; +} + +function quietLogger() { + return { log() {}, warn() {} }; +} diff --git a/internal/cloud/workbench-kafka-session-index.ts b/internal/cloud/workbench-kafka-session-index.ts new file mode 100644 index 00000000..f34935f0 --- /dev/null +++ b/internal/cloud/workbench-kafka-session-index.ts @@ -0,0 +1,473 @@ +// SPEC: PJ2026-010401080313 Workbench实时权威 draft-2026-07-20-p0-process-session-index. +// Responsibility: accelerate retained hwlab.event.v1 session queries without creating another business authority. + +import { createHash } from "node:crypto"; + +export function createWorkbenchKafkaSessionIndex(options = {}) { + const topic = text(options.topic); + const maxEvents = positiveInteger(options.maxEvents); + const maxEventsPerSession = positiveInteger(options.maxEventsPerSession); + const bootstrapTimeoutMs = positiveInteger(options.bootstrapTimeoutMs); + const rebuildIntervalMs = positiveInteger(options.rebuildIntervalMs); + const rebuildCooldownMs = positiveInteger(options.rebuildCooldownMs); + const scanLimit = positiveInteger(options.scanLimit); + const logger = options.logger ?? console; + const nowMs = typeof options.nowMs === "function" ? options.nowMs : Date.now; + if (!topic || !maxEvents || !maxEventsPerSession || !bootstrapTimeoutMs || !rebuildIntervalMs || !rebuildCooldownMs || !scanLimit || typeof options.queryAll !== "function") { + throw indexError("workbench_kafka_session_index_config_invalid", "Workbench Kafka session index requires complete bounded configuration."); + } + + let ready = false; + let stopped = false; + let rebuilding = false; + let rebuildTimer = null; + let lastRebuildStartedAtMs = null; + let lastRebuildCompletedAtMs = null; + let lastFailure = null; + let bootstrapPromise = null; + let pendingLiveOverflow = false; + let state = emptyState(); + const pendingLive = []; + + function start() { + if (bootstrapPromise) return bootstrapPromise; + bootstrapPromise = rebuild("bootstrap"); + rebuildTimer = setInterval(() => { + void rebuild("interval"); + }, rebuildIntervalMs); + rebuildTimer.unref?.(); + return bootstrapPromise; + } + + function stop() { + stopped = true; + if (rebuildTimer) clearInterval(rebuildTimer); + rebuildTimer = null; + pendingLive.length = 0; + } + + function observeLive(envelope, transport) { + if (stopped || envelope?.schema !== "hwlab.event.v1") return; + const record = normalizeRecord({ topic, partition: transport?.partition, offset: transport?.offset, key: sessionId(envelope), timestamp: null, value: envelope }); + if (!record) return; + if (rebuilding) { + if (pendingLive.length >= maxEvents) { + pendingLiveOverflow = true; + lastFailure = warning("workbench_kafka_session_index_live_buffer_overflow", "Session index rebuild live buffer exceeded its YAML-owned bound."); + } else { + pendingLive.push(record); + } + } + if (ready) addRecord(state, record); + } + + async function rebuild(reason) { + if (stopped || rebuilding) return false; + const startedAtMs = nowMs(); + if (lastRebuildStartedAtMs !== null && reason !== "bootstrap" && startedAtMs - lastRebuildStartedAtMs < rebuildCooldownMs) return false; + rebuilding = true; + lastRebuildStartedAtMs = startedAtMs; + pendingLive.length = 0; + pendingLiveOverflow = false; + try { + const result = await options.queryAll({ + limit: maxEvents, + scanLimit, + timeoutMs: bootstrapTimeoutMs + }); + if (!completeQuery(result)) { + lastFailure = warning("workbench_kafka_session_index_bootstrap_incomplete", `Session index ${reason} did not reach the Kafka retention barrier.`); + logWarning(lastFailure, { reason, completionReason: result?.completionReason ?? null }); + return false; + } + const candidate = emptyState(); + candidate.endOffsets = normalizeEndOffsets(result.endOffsets); + for (const raw of Array.isArray(result.events) ? result.events : []) { + const record = normalizeRecord(raw); + if (!record) continue; + addRecord(candidate, record); + } + if (pendingLiveOverflow) { + throw indexError("workbench_kafka_session_index_live_buffer_overflow", "Session index rebuild cannot prove completeness after its live buffer overflowed."); + } + for (const record of pendingLive.splice(0)) addRecord(candidate, record); + state = candidate; + ready = true; + lastFailure = null; + lastRebuildCompletedAtMs = nowMs(); + logger.log?.(JSON.stringify({ + code: "workbench-kafka-session-index-ready", + reason, + indexedEventCount: state.records.length, + indexedSessionCount: state.bySession.size, + incompleteSessionCount: state.incompleteSessions.size, + durationMs: lastRebuildCompletedAtMs - startedAtMs, + valuesPrinted: false + })); + return true; + } catch (error) { + lastFailure = warning(error?.code ?? "workbench_kafka_session_index_bootstrap_failed", errorMessage(error)); + logWarning(lastFailure, { reason }); + return false; + } finally { + rebuilding = false; + } + } + + function query(params = {}) { + const startedAtMs = nowMs(); + const requestedSessionId = text(params.sessionId); + const requestedTraceId = text(params.traceId); + const fallbackReason = !ready + ? "index-not-ready" + : requestedSessionId && state.incompleteSessions.has(requestedSessionId) + ? "session-capacity-exceeded" + : !requestedSessionId && !state.globalComplete + ? "global-capacity-exceeded" + : null; + if (fallbackReason) return { + hit: false, + fallbackReason, + status: status(), + warning: warning("workbench_kafka_session_index_fallback", `Session index cannot prove a complete result (${fallbackReason}).`) + }; + const source = requestedSessionId ? state.bySession.get(requestedSessionId) ?? [] : state.records; + const events = source.filter((record) => matchesTrace(record.value, requestedTraceId)); + const limit = positiveInteger(params.limit); + if (limit && events.length > limit) { + return { + hit: false, + fallbackReason: "matched-event-limit", + status: status(), + warning: warning("workbench_kafka_session_index_fallback", "Session index result exceeds the request matched-event limit.") + }; + } + const elapsedMs = Math.max(0, nowMs() - startedAtMs); + const firstScannedOffsets = partitionOffsets(events, "first"); + const lastScannedOffsets = partitionOffsets(events, "last"); + return { + hit: true, + result: { + ok: true, + stream: "hwlab", + topic, + groupId: null, + count: events.length, + scannedCount: events.length, + parsedCount: events.length, + matchedCount: events.length, + invalidJsonCount: 0, + filterRejectedCount: 0, + keyRejectedCount: 0, + excludedPostBarrierCount: 0, + completionReason: "session-index", + completion: { + reason: "session-index", + complete: true, + barrierReached: true, + retentionStartVerified: true, + retentionStartMoved: false, + timeout: false, + matchedEventLimitReached: false, + scanLimitReached: false, + aborted: false, + valuesRedacted: true + }, + reachedEndOffsets: true, + endOffsets: [...state.endOffsets.values()].sort((left, right) => left.partition - right.partition), + endOffsetsAvailable: state.endOffsets.size > 0, + endOffsetsError: null, + firstScannedOffsets, + lastScannedOffsets, + limit: limit ?? null, + scanLimit: positiveInteger(params.scanLimit), + timeoutMs: positiveInteger(params.timeoutMs), + filters: compact({ sessionId: requestedSessionId, traceId: requestedTraceId }), + partitionKeyScoped: Boolean(text(params.partitionKey)), + targetPartition: events.length > 0 ? events[0].partition : null, + timing: { + endOffsetSnapshotMs: 0, + groupOffsetsMs: 0, + consumerConnectMs: 0, + consumerSubscribeMs: 0, + scanMs: 0, + cleanupMs: 0, + indexLookupMs: elapsedMs, + totalMs: elapsedMs, + valuesPrinted: false + }, + index: { + hit: true, + source: "process-session-index", + indexedEventCount: state.records.length, + indexedSessionCount: state.bySession.size, + incompleteSessionCount: state.incompleteSessions.size, + valuesRedacted: true + }, + events, + valuesPrinted: false + } + }; + } + + function status() { + return { + enabled: true, + ready, + rebuilding, + indexedEventCount: state.records.length, + indexedSessionCount: state.bySession.size, + incompleteSessionCount: state.incompleteSessions.size, + globalComplete: state.globalComplete, + lastRebuildStartedAt: iso(lastRebuildStartedAtMs), + lastRebuildCompletedAt: iso(lastRebuildCompletedAtMs), + warning: lastFailure, + blocking: false, + valuesRedacted: true + }; + } + + function emptyState() { + return { + records: [], + byTransport: new Map(), + bySession: new Map(), + incompleteSessions: new Set(), + endOffsets: new Map(), + globalComplete: true + }; + } + + function addRecord(target, record) { + const transportKey = `${record.topic}:${record.partition}:${record.offset}`; + if (target.byTransport.has(transportKey)) return; + target.byTransport.set(transportKey, record); + target.records.push(record); + updateEndOffset(target.endOffsets, record.partition, record.offset); + const id = indexedSessionId(record); + if (id) { + const bucket = target.bySession.get(id) ?? []; + bucket.push(record); + target.bySession.set(id, bucket); + if (bucket.length > maxEventsPerSession) { + const removed = bucket.shift(); + if (removed) removeRecord(target, removed); + target.incompleteSessions.add(id); + } + } + if (target.records.length > maxEvents) { + const removed = target.records[0]; + removeRecord(target, removed); + const removedSessionId = indexedSessionId(removed); + if (removedSessionId) target.incompleteSessions.add(removedSessionId); + target.globalComplete = false; + } + } + + function removeRecord(target, record) { + if (!record) return; + const transportKey = `${record.topic}:${record.partition}:${record.offset}`; + target.byTransport.delete(transportKey); + const recordIndex = target.records.indexOf(record); + if (recordIndex >= 0) target.records.splice(recordIndex, 1); + const id = indexedSessionId(record); + const bucket = id ? target.bySession.get(id) : null; + if (bucket) { + const bucketIndex = bucket.indexOf(record); + if (bucketIndex >= 0) bucket.splice(bucketIndex, 1); + if (bucket.length === 0) target.bySession.delete(id); + } + } + + function logWarning(entry, fields = {}) { + logger.warn?.(JSON.stringify({ + code: entry.code, + component: "workbench-kafka-session-index", + message: entry.message, + blocking: false, + ...fields, + valuesPrinted: false + })); + } + + return { + start, + stop, + observeLive, + query, + status, + rebuild, + valuesPrinted: false + }; +} + +function normalizeRecord(raw) { + const recordTopic = text(raw?.topic); + const partition = integer(raw?.partition); + const offset = offsetValue(raw?.offset); + const value = raw?.value; + if (!recordTopic || partition === null || offset === null || value?.schema !== "hwlab.event.v1") return null; + return { + topic: recordTopic, + partition, + offset, + key: text(raw?.key), + timestamp: text(raw?.timestamp), + valueSha256: text(raw?.valueSha256) ?? sha256(JSON.stringify(value)), + value + }; +} + +function completeQuery(result) { + return result?.completion?.complete === true + && result?.completion?.barrierReached === true + && result?.completion?.retentionStartVerified === true + && result?.reachedEndOffsets === true + && result?.endOffsetsAvailable === true; +} + +function normalizeEndOffsets(value) { + const result = new Map(); + for (const entry of Array.isArray(value) ? value : []) { + const partition = integer(entry?.partition); + const endOffset = offsetValue(entry?.endOffset); + if (partition === null || endOffset === null) continue; + result.set(partition, { + partition, + startOffset: offsetValue(entry?.startOffset), + endOffset + }); + } + return result; +} + +function updateEndOffset(endOffsets, partition, offset) { + const next = nextOffset(offset); + if (next === null) return; + const existing = endOffsets.get(partition); + if (!existing || compareOffset(next, existing.endOffset) > 0) { + endOffsets.set(partition, { + partition, + startOffset: existing?.startOffset ?? null, + endOffset: next + }); + } +} + +function sessionId(value) { + return firstText( + value?.sessionId, + value?.hwlabSessionId, + value?.event?.sessionId, + value?.event?.hwlabSessionId, + value?.context?.sessionId, + value?.context?.hwlabSessionId + ); +} + +function indexedSessionId(record) { + const envelopeSessionId = sessionId(record?.value); + const key = text(record?.key); + if (!envelopeSessionId) return null; + if (key && key !== envelopeSessionId) return null; + return envelopeSessionId; +} + +function matchesTrace(value, traceId) { + if (!traceId) return true; + return [ + value?.traceId, + value?.event?.traceId, + value?.context?.traceId, + value?.sourceEvent?.traceId + ].some((candidate) => text(candidate) === traceId); +} + +function partitionOffsets(records, mode) { + const byPartition = new Map(); + for (const record of records) { + const existing = byPartition.get(record.partition); + if (existing === undefined || (mode === "first" ? compareOffset(record.offset, existing) < 0 : compareOffset(record.offset, existing) > 0)) { + byPartition.set(record.partition, record.offset); + } + } + return [...byPartition.entries()] + .map(([partition, offset]) => ({ partition, offset })) + .sort((left, right) => left.partition - right.partition); +} + +function warning(code, message) { + return { code, message, blocking: false, valuesRedacted: true }; +} + +function indexError(code, message) { + const error = new Error(message); + error.code = code; + return error; +} + +function errorMessage(error) { + return error instanceof Error ? error.message : String(error ?? "unknown error"); +} + +function compact(value) { + return Object.fromEntries(Object.entries(value).filter(([, item]) => item !== null && item !== undefined && item !== "")); +} + +function firstText(...values) { + for (const value of values) { + const result = text(value); + if (result) return result; + } + return null; +} + +function text(value) { + return typeof value === "string" && value.trim() ? value.trim() : null; +} + +function integer(value) { + const result = Number(value); + return Number.isInteger(result) && result >= 0 ? result : null; +} + +function positiveInteger(value) { + const result = Number(value); + return Number.isInteger(result) && result > 0 ? result : null; +} + +function offsetValue(value) { + const result = text(String(value ?? "")); + if (!result) return null; + try { + return BigInt(result) >= 0n ? result : null; + } catch { + return null; + } +} + +function nextOffset(value) { + try { + return String(BigInt(String(value)) + 1n); + } catch { + return null; + } +} + +function compareOffset(left, right) { + try { + const leftValue = BigInt(String(left)); + const rightValue = BigInt(String(right)); + return leftValue < rightValue ? -1 : leftValue > rightValue ? 1 : 0; + } catch { + return String(left).localeCompare(String(right)); + } +} + +function iso(value) { + return Number.isFinite(value) ? new Date(value).toISOString() : null; +} + +function sha256(value) { + return createHash("sha256").update(value).digest("hex"); +} diff --git a/internal/workbench/http.ts b/internal/workbench/http.ts index 2c664286..d28f85c4 100644 --- a/internal/workbench/http.ts +++ b/internal/workbench/http.ts @@ -176,6 +176,7 @@ function nativeKafkaRefreshEventStream(bridge: any, sessionId: string, traceId: keyRejectedCount: result?.keyRejectedCount ?? null, partitionKeyScoped: result?.partitionKeyScoped === true, targetPartition: result?.targetPartition ?? null, + index: result?.index ?? null, timing: result?.timing ?? null, valuesPrinted: false })); diff --git a/internal/workbench/workbench.test.ts b/internal/workbench/workbench.test.ts index 863516a4..97931278 100644 --- a/internal/workbench/workbench.test.ts +++ b/internal/workbench/workbench.test.ts @@ -191,7 +191,8 @@ describe("Workbench native HTTP adapter", () => { keyRejectedCount: 0, partitionKeyScoped: true, targetPartition: 0, - timing: { endOffsetSnapshotMs: 2, groupOffsetsMs: 0, consumerConnectMs: 3, consumerSubscribeMs: 1, scanMs: 7, cleanupMs: 1, totalMs: 14 } + timing: { endOffsetSnapshotMs: 0, groupOffsetsMs: 0, consumerConnectMs: 0, consumerSubscribeMs: 0, scanMs: 0, cleanupMs: 0, indexLookupMs: 1, totalMs: 1 }, + index: { hit: true, source: "process-session-index", indexedEventCount: 300, indexedSessionCount: 3, incompleteSessionCount: 0, valuesRedacted: true } }; } }; @@ -231,7 +232,8 @@ describe("Workbench native HTTP adapter", () => { keyRejectedCount: 0, partitionKeyScoped: true, targetPartition: 0, - timing: { scanMs: 7, totalMs: 14 } + index: { hit: true, source: "process-session-index", indexedEventCount: 300, indexedSessionCount: 3, incompleteSessionCount: 0, blocking: false }, + timing: { scanMs: 0, indexLookupMs: 1, totalMs: 1 } } } }, diff --git a/tools/src/workbench-cli.ts b/tools/src/workbench-cli.ts index f97c95a4..9d44ef64 100644 --- a/tools/src/workbench-cli.ts +++ b/tools/src/workbench-cli.ts @@ -156,6 +156,7 @@ function workbenchConnectedContract(data: Record) { const counts = record(refreshReplay?.counts); const query = record(refreshReplay?.query); const queryTiming = record(query?.timing); + const queryIndex = record(query?.index); return { deliverySemantics: String(data.deliverySemantics ?? "").trim() || null, realtimeSource: String(data.realtimeSource ?? "").trim() || null, @@ -184,6 +185,16 @@ function workbenchConnectedContract(data: Record) { keyRejectedCount: nullableNumber(query.keyRejectedCount), partitionKeyScoped: query.partitionKeyScoped === true, targetPartition: nullableNumber(query.targetPartition), + index: queryIndex ? { + hit: queryIndex.hit === true, + source: String(queryIndex.source ?? "").trim() || null, + fallbackReason: String(queryIndex.fallbackReason ?? "").trim() || null, + indexedEventCount: nullableNumber(queryIndex.indexedEventCount), + indexedSessionCount: nullableNumber(queryIndex.indexedSessionCount), + incompleteSessionCount: nullableNumber(queryIndex.incompleteSessionCount), + blocking: false, + valuesPrinted: false + } : null, timing: queryTiming ? { endOffsetSnapshotMs: nullableNumber(queryTiming.endOffsetSnapshotMs), groupOffsetsMs: nullableNumber(queryTiming.groupOffsetsMs), @@ -191,6 +202,7 @@ function workbenchConnectedContract(data: Record) { consumerSubscribeMs: nullableNumber(queryTiming.consumerSubscribeMs), scanMs: nullableNumber(queryTiming.scanMs), cleanupMs: nullableNumber(queryTiming.cleanupMs), + indexLookupMs: nullableNumber(queryTiming.indexLookupMs), totalMs: nullableNumber(queryTiming.totalMs), valuesPrinted: false } : null, 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 32b6dad9..1cf98267 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 @@ -38,7 +38,7 @@ test("live Kafka user event restores and idempotently updates the user bubble", assert.equal(first?.text, "persisted user input"); assert.equal(second?.id, first?.id); assert.equal(second?.createdAt, first?.createdAt); - assert.equal(second?.updatedAt, "2026-07-10T10:00:06.000Z"); + assert.equal(second?.updatedAt, event.createdAt); }); test("HTTP-first and Kafka-first admission converge on one stable optimistic user bubble", () => { @@ -230,7 +230,7 @@ test("refresh rebuilds the completed turn from an empty store using only Kafka S assert.equal(afterRefresh[1]?.threadId, threadId); }); -test("live Kafka projection refreshes lastEventAt from each ingress receipt", () => { +test("live Kafka projection uses business event time instead of replay receipt time", () => { const traceId = "trc_live_timing"; const sessionId = "ses_live_timing"; const firstReceivedAt = "2026-07-10T10:00:05.000Z"; @@ -250,12 +250,13 @@ test("live Kafka projection refreshes lastEventAt from each ingress receipt", () event: traceEvent(2, "assistant", { assistantText: "visible response" }) }); - assert.equal(first.lastEventAt, firstReceivedAt); - assert.equal(second.lastEventAt, secondReceivedAt); - assert.equal(second.timing?.lastEventAt, secondReceivedAt); + assert.equal(first.lastEventAt, "2026-07-10T10:00:01.000Z"); + assert.equal(second.lastEventAt, "2026-07-10T10:00:02.000Z"); + assert.equal(second.timing?.lastEventAt, "2026-07-10T10:00:02.000Z"); assert.equal(second.timing?.lastEventAgeMs, 0); - assert.equal(second.runnerTrace?.lastEventAt, secondReceivedAt); - assert.equal(second.runnerTrace?.timing?.lastEventAt, secondReceivedAt); + assert.equal(second.timing?.observedAt, secondReceivedAt); + assert.equal(second.runnerTrace?.lastEventAt, "2026-07-10T10:00:02.000Z"); + assert.equal(second.runnerTrace?.timing?.lastEventAt, "2026-07-10T10:00:02.000Z"); assert.equal(second.runnerTrace?.eventCount, 2); }); @@ -339,21 +340,56 @@ test("live Kafka projection keeps terminal lifecycle while accepting a later eve event: traceEvent(3, "tool") }); - assert.equal(terminal.finishedAt, terminalAt); + assert.equal(terminal.finishedAt, "2026-07-10T10:00:02.000Z"); const terminalFinalResponse = terminal.finalResponse as { text?: string } | null | undefined; assert.equal(terminalFinalResponse?.text, "final answer"); assert.equal(running.status, "completed"); - assert.equal(running.lastEventAt, acceptedAt); + assert.equal(running.lastEventAt, "2026-07-10T10:00:03.000Z"); assert.equal(running.timing?.lastEventAgeMs, 0); - assert.equal(running.finishedAt, terminalAt); + assert.equal(running.finishedAt, "2026-07-10T10:00:02.000Z"); assert.equal(running.durationMs, terminal.durationMs); assert.equal((running.finalResponse as { text?: string } | null)?.text, "final answer"); - assert.equal(running.runnerTrace?.finishedAt, terminalAt); + assert.equal(running.runnerTrace?.finishedAt, "2026-07-10T10:00:02.000Z"); assert.equal(running.runnerTrace?.durationMs, terminal.durationMs); assert.equal(running.runnerTrace?.eventCount, 3); assert.equal(running.traceAutoLifecycle, "terminal"); }); +test("live and retained Kafka terminal duration share the user-send-to-terminal business clock", () => { + const turnStartedAt = "2026-07-10T00:00:00.000Z"; + const event = traceEvent(2, "result", { + traceId: "trc_replay_duration", + sessionId: "ses_replay_duration", + terminal: true, + status: "completed", + createdAt: "2026-07-10T00:01:03.000Z" + }); + const live = projectWorkbenchLiveKafkaMessage({ + previous: null, + traceId: "trc_replay_duration", + sessionId: "ses_replay_duration", + turnStartedAt, + receivedAt: "2026-07-10T00:01:03.010Z", + event + }); + const replay = projectWorkbenchLiveKafkaMessage({ + previous: null, + traceId: "trc_replay_duration", + sessionId: "ses_replay_duration", + turnStartedAt, + receivedAt: "2026-07-10T10:30:00.000Z", + event + }); + + assert.equal(live.startedAt, turnStartedAt); + assert.equal(live.finishedAt, "2026-07-10T00:01:03.000Z"); + assert.equal(live.durationMs, 63000); + assert.equal(replay.startedAt, live.startedAt); + assert.equal(replay.finishedAt, live.finishedAt); + assert.equal(replay.durationMs, live.durationMs); + assert.equal(replay.timing?.observedAt, "2026-07-10T10:30:00.000Z"); +}); + function seededMessage(traceId: string, sessionId: string): ChatMessage { return projectWorkbenchLiveKafkaMessage({ previous: null, 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 e74786c3..b7983493 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 @@ -18,6 +18,7 @@ export interface WorkbenchLiveKafkaProjectionInput { sessionId: string; event: TraceEvent; receivedAt: string; + turnStartedAt?: string | null; title?: string; threadId?: string | null; } @@ -96,7 +97,7 @@ export function projectWorkbenchLiveKafkaUserMessage(input: WorkbenchLiveKafkaUs sessionId: input.sessionId, threadId: firstNonEmptyString(input.threadId, input.previous?.threadId), createdAt: input.previous?.createdAt ?? createdAt, - updatedAt: receivedAt + updatedAt: createdAt } as ChatMessage; } @@ -124,6 +125,7 @@ export function projectWorkbenchLiveKafkaMessage(input: WorkbenchLiveKafkaProjec const receivedAt = timestamp(input.receivedAt); if (!receivedAt) throw new Error("receivedAt must be a valid timestamp for live Kafka projection"); const previous = input.previous; + const eventAt = timestamp(input.event.createdAt) ?? receivedAt; const liveState = reduceWorkbenchLiveKafkaMessageState({ text: previous?.text ?? "", status: previous?.status ?? "running", @@ -131,19 +133,19 @@ export function projectWorkbenchLiveKafkaMessage(input: WorkbenchLiveKafkaProjec }, input.event, Array.isArray(previous?.runnerTrace?.events) ? previous.runnerTrace.events : []); const startedAt = timestamp(previous?.timing?.startedAt) ?? timestamp(previous?.startedAt) - ?? timestamp(input.event.createdAt) - ?? receivedAt; + ?? timestamp(input.turnStartedAt) + ?? eventAt; const previousFinishedAt = timestamp(previous?.timing?.finishedAt) ?? timestamp(previous?.finishedAt); - const finishedAt = liveState.terminal ? previousFinishedAt ?? receivedAt : null; + const finishedAt = liveState.terminal ? previousFinishedAt ?? eventAt : null; const previousDurationMs = nonNegativeNumber(previous?.timing?.durationMs) ?? nonNegativeNumber(previous?.durationMs); const durationMs = liveState.terminal ? previousFinishedAt && previousDurationMs !== null ? previousDurationMs - : Math.max(0, Date.parse(finishedAt ?? receivedAt) - Date.parse(startedAt)) + : Math.max(0, Date.parse(finishedAt ?? eventAt) - Date.parse(startedAt)) : null; const timing: WorkbenchTurnTimingProjection = { startedAt, - lastEventAt: receivedAt, + lastEventAt: eventAt, finishedAt, durationMs, observedAt: receivedAt, @@ -160,10 +162,10 @@ export function projectWorkbenchLiveKafkaMessage(input: WorkbenchLiveKafkaProjec eventSource: "hwlab-kafka-sse", timing, startedAt, - lastEventAt: receivedAt, + lastEventAt: eventAt, finishedAt, durationMs, - updatedAt: receivedAt + updatedAt: eventAt }); const runnerTrace = { ...mergedRunnerTrace, @@ -171,10 +173,10 @@ export function projectWorkbenchLiveKafkaMessage(input: WorkbenchLiveKafkaProjec traceStatus: liveState.status, timing, startedAt, - lastEventAt: receivedAt, + lastEventAt: eventAt, finishedAt, durationMs, - updatedAt: receivedAt + updatedAt: eventAt }; const messageId = workbenchAgentMessageIdForTrace(input.traceId); return { @@ -190,9 +192,9 @@ export function projectWorkbenchLiveKafkaMessage(input: WorkbenchLiveKafkaProjec sessionId: input.sessionId, threadId: firstNonEmptyString(input.threadId, previous?.threadId), createdAt: previous?.createdAt ?? startedAt, - updatedAt: receivedAt, + updatedAt: eventAt, startedAt, - lastEventAt: receivedAt, + lastEventAt: eventAt, finishedAt, durationMs, runnerTrace, 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 a254ae6c..71806cfd 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-session.test.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-session.test.ts @@ -140,3 +140,28 @@ test("session rail status prefers newer running trace over stale completed autho assert.equal(tab.running, true); assert.equal(tab.lastTraceId, current.traceId); }); + +test("session rail update time uses the last user send time instead of replay receipt time", () => { + const sessionId = "ses_session_rail_user_time"; + const user = { + id: "msg_session_rail_user_time", + messageId: "msg_session_rail_user_time", + role: "user", + title: "用户", + text: "hi", + status: "sent", + traceId: "trc_session_rail_user_time", + sessionId, + createdAt: "2026-07-20T00:00:00.000Z", + updatedAt: "2026-07-20T10:30:00.000Z" + } as ChatMessage; + const tab = sessionToSessionTab({ + sessionId, + status: "completed", + lastUserMessageAt: "2026-07-20T10:30:00.000Z", + updatedAt: "2026-07-20T10:30:00.000Z", + messages: [user] + }, sessionId); + + assert.equal(tab.updatedAt, "2026-07-20T00:00:00.000Z"); +}); diff --git a/web/hwlab-cloud-web/src/stores/workbench-session.ts b/web/hwlab-cloud-web/src/stores/workbench-session.ts index bffad4dd..eb5bc5f4 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-session.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-session.ts @@ -441,9 +441,9 @@ function firstReadableSentence(...values: unknown[]): string | null { export function sessionDisplayUpdatedAt(session: WorkbenchSessionRecord): string | null { return firstNonEmptyString( + latestUserMessageAtFromMessages(session.messages ?? []), session.lastUserMessageAt, session.snapshot?.lastUserMessageAt, - latestUserMessageAtFromMessages(session.messages ?? []), session.startedAt, session.updatedAt, session.snapshot?.updatedAt @@ -453,7 +453,7 @@ export function sessionDisplayUpdatedAt(session: WorkbenchSessionRecord): string export function latestUserMessageAtFromMessages(messages: ChatMessage[] | undefined): string | null { const userMessageTimes = (messages ?? []) .filter((message) => message.role === "user") - .flatMap((message) => [message.createdAt, message.updatedAt]); + .map((message) => firstNonEmptyString(message.createdAt, message.updatedAt)); return latestTimestamp(...userMessageTimes); } diff --git a/web/hwlab-cloud-web/src/stores/workbench.ts b/web/hwlab-cloud-web/src/stores/workbench.ts index 8a3b9ec9..b0c3a713 100644 --- a/web/hwlab-cloud-web/src/stores/workbench.ts +++ b/web/hwlab-cloud-web/src/stores/workbench.ts @@ -1072,12 +1072,26 @@ export const useWorkbenchStore = defineStore("workbench", () => { ? (serverState.value.messagesBySessionId[ownerSessionId] ?? []).find((message) => (message.messageId ?? message.id) === messageId) ?? null : null; const userMessage = projectWorkbenchLiveKafkaUserMessage({ previous: previousUserMessage, traceId, sessionId: ownerSessionId, event, receivedAt, threadId }); - if (userMessage) reduceServerState({ type: "message.upsert", sessionId: ownerSessionId, message: userMessage }); + if (userMessage) { + reduceServerState({ type: "message.upsert", sessionId: ownerSessionId, message: userMessage }); + const existing = sessions.value.find((session) => session.sessionId === ownerSessionId) ?? null; + if (existing) { + const lastUserMessageAt = firstNonEmptyString(userMessage.createdAt, userMessage.updatedAt); + rememberSessionList(mergeSessionIntoList(sessions.value, { + ...existing, + lastUserMessageAt, + updatedAt: lastUserMessageAt ?? existing.updatedAt + })); + } + } else error.value = "workbench_live_user_message_invalid"; return; } - const previousMessage = (serverState.value.messagesBySessionId[ownerSessionId] ?? []).find((message) => messageMatchesTraceAuthority(message, traceId, authoritySessionId ?? eventSessionId, ownerSessionId, true)) ?? null; - const message = projectWorkbenchLiveKafkaMessage({ previous: previousMessage, traceId, sessionId: ownerSessionId, event, receivedAt, threadId }); + const sessionMessages = serverState.value.messagesBySessionId[ownerSessionId] ?? []; + const previousMessage = sessionMessages.find((message) => messageMatchesTraceAuthority(message, traceId, authoritySessionId ?? eventSessionId, ownerSessionId, true)) ?? null; + const turnUserMessage = [...sessionMessages].reverse().find((candidate) => candidate.role === "user" && firstNonEmptyString(candidate.traceId, candidate.targetTraceId) === traceId) ?? null; + const turnStartedAt = firstNonEmptyString(turnUserMessage?.createdAt, turnUserMessage?.updatedAt); + const message = projectWorkbenchLiveKafkaMessage({ previous: previousMessage, traceId, sessionId: ownerSessionId, event, receivedAt, turnStartedAt, threadId }); reduceServerState({ type: "message.upsert", sessionId: ownerSessionId, message }); if (message.runnerTrace) rememberTraceAuthority(message.runnerTrace); markWorkbenchTraceProjected(traceId); @@ -1096,10 +1110,20 @@ export const useWorkbenchStore = defineStore("workbench", () => { lastEventAt: message.timing?.lastEventAt, finishedAt: message.timing?.finishedAt, durationMs: message.timing?.durationMs, - updatedAt: receivedAt + updatedAt: message.updatedAt } as AgentChatResultResponse); const existing = sessions.value.find((session) => session.sessionId === ownerSessionId) ?? null; - if (existing) rememberSessionList(mergeSessionIntoList(sessions.value, { ...existing, threadId: message.threadId ?? existing.threadId, status, lastTraceId: traceId, updatedAt: receivedAt })); + if (existing) { + const lastUserMessageAt = firstNonEmptyString(existing.lastUserMessageAt, turnStartedAt); + rememberSessionList(mergeSessionIntoList(sessions.value, { + ...existing, + threadId: message.threadId ?? existing.threadId, + status, + lastTraceId: traceId, + lastUserMessageAt, + updatedAt: lastUserMessageAt ?? existing.updatedAt + })); + } if (terminal && currentRequest.value?.traceId === traceId) { chatPending.value = false; currentRequest.value = null;