fix(workbench): 统一 Kafka 回放时钟并加速会话加载
This commit is contained in:
root
@@ -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",
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -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
|
||||
}));
|
||||
|
||||
@@ -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
|
||||
},
|
||||
|
||||
@@ -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<typeof record>[], 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() {} };
|
||||
}
|
||||
@@ -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");
|
||||
}
|
||||
@@ -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
|
||||
}));
|
||||
|
||||
@@ -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 }
|
||||
}
|
||||
}
|
||||
},
|
||||
|
||||
@@ -156,6 +156,7 @@ function workbenchConnectedContract(data: Record<string, any>) {
|
||||
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<string, any>) {
|
||||
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<string, any>) {
|
||||
consumerSubscribeMs: nullableNumber(queryTiming.consumerSubscribeMs),
|
||||
scanMs: nullableNumber(queryTiming.scanMs),
|
||||
cleanupMs: nullableNumber(queryTiming.cleanupMs),
|
||||
indexLookupMs: nullableNumber(queryTiming.indexLookupMs),
|
||||
totalMs: nullableNumber(queryTiming.totalMs),
|
||||
valuesPrinted: false
|
||||
} : null,
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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");
|
||||
});
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
|
||||
@@ -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;
|
||||
|
||||
Reference in New Issue
Block a user