fix: 修复纯 Kafka Trace 实时可见性
This commit is contained in:
root
@@ -30,4 +30,19 @@
|
||||
修复隔离 debug replay 的 producer/barrier 与分层结果合同：保留 source lineage、replayId、offset range，报告 scanned/matched/delivered/decoded/applied/terminal，区分 producer-not-invoked、trace missing、filter mismatch、decoder/reducer rejection 与 terminal missing；仅写 debug topic。上下文：[隔离根因 #2476](https://github.com/pikasTech/HWLAB/issues/2476#issuecomment-4935398125)、[主方案 #2474](https://github.com/pikasTech/HWLAB/issues/2474#issuecomment-4935408473)。依赖：无。验证：empty、matched-no-terminal、reducer-rejected、terminal-complete 四类 typed 结果及 semantic internal web-probe command，完成任务后将详细报告写入[任务报告](./details/pure-kafka-live-single-step-debug/R2.3_Task_Report.md)。
|
||||
### R2.4 [in_progress]
|
||||
|
||||
集成前三个子任务，受控部署到 NC01/v03 并用原入口验证运行中及结束后 Trace 可读、同一 ingress raw/reducer 计数可解释、隔离 debug 命中 terminal；保持 OTel 三服务完整、error=0、projection=0 且产品链无补洞路径。上下文：[验收 #2474](https://github.com/pikasTech/HWLAB/issues/2474)、[OTel 基线 #2477](https://github.com/pikasTech/HWLAB/issues/2477#issuecomment-4935385934)。依赖：Trace 序列权威修复、单一 ingress raw 观察窗、隔离 debug replay 分层诊断。验证：定向测试、YAML/render/runtime contract、受控 CI/CD、semantic --origin internal web-probe 与 Kafka/SSE/DOM/OTel bounded counts，完成任务后将详细报告写入[任务报告](./details/pure-kafka-live-single-step-debug/R2.4_Task_Report.md)。
|
||||
以 [HWLAB #2474](https://github.com/pikasTech/HWLAB/issues/2474) 和固定 trace trc_fb6031c28d79495e 完成纯 Kafka 集成闭环：先增强 CLI 直接读取持久 hwlab.event 并复用前端 decode、merge、row model、summary 输出同构 Markdown，再修复 Trace 内 final response 重复、tool-call summary 可见性和每个 HWLAB SSE ingress 刷新 lastEventAt；最后受控部署 NC01/v03，以 Kafka 重放、semantic internal web-probe、DOM 与 OTel 验收 projection=0 且无补洞路径，完成任务后将详细报告写入[任务报告](./details/pure-kafka-live-single-step-debug/R2.4_Task_Report.md)。
|
||||
## R3
|
||||
|
||||
在纯 Kafka SSE 权威链内实现页面刷新后的可重放引导，明确区别于 projector/read model/snapshot/sync/gap-fill/finalizer 补洞；承接上下文：[HWLAB #2474](https://github.com/pikasTech/HWLAB/issues/2474)，完成任务后将详细报告写入[任务报告](./details/pure-kafka-live-single-step-debug/R3_Task_Report.md)。
|
||||
### R3.1
|
||||
|
||||
深入分析 Kafka retention、SSE 刷新连接、offset barrier、live handoff 与去重身份，形成不引入第二事实源的方案和失败可见性，完成任务后将详细报告写入[任务报告](./details/pure-kafka-live-single-step-debug/R3.1_Task_Report.md)。
|
||||
### R3.2
|
||||
|
||||
实现 trace/session scoped Kafka refresh bootstrap 与 live fanout 无缝交接，全部开关、group、容量和时限保持 YAML-first，完成任务后将详细报告写入[任务报告](./details/pure-kafka-live-single-step-debug/R3.2_Task_Report.md)。
|
||||
### R3.3
|
||||
|
||||
优先使用既有 trace 通过 CLI 单步重放验证顺序、terminal、final 唯一性、tool summary 和 timing，不启动新真实运行，完成任务后将详细报告写入[任务报告](./details/pure-kafka-live-single-step-debug/R3.3_Task_Report.md)。
|
||||
### R3.4
|
||||
|
||||
完成定向测试、受控 CI/CD 与 semantic internal web-probe 页面刷新验收，确认 projection=0 且无 snapshot/sync/gap-fill/finalizer 路径，完成任务后将详细报告写入[任务报告](./details/pure-kafka-live-single-step-debug/R3.4_Task_Report.md)。
|
||||
@@ -1,5 +1,50 @@
|
||||
# Kafka 源码直连调试
|
||||
|
||||
- `hwlab-cli kafka render trace` 用于把持久 HWLAB Kafka 事件直接渲染为与 Web 同构的 Markdown Trace：
|
||||
- 命令为 `hwlab-cli kafka render trace --from kafka --trace-id <traceId> --format markdown`；
|
||||
- 默认输入 topic 优先读取 owning YAML 注入的 `HWLAB_KAFKA_EVENT_TOPIC`；
|
||||
- 未注入时使用 canonical `hwlab.event.v1`；
|
||||
- Kafka 读取必须使用 `--group-prefix` 或 owning YAML 注入的 `HWLAB_KAFKA_HWLAB_DEBUG_GROUP_PREFIX`；
|
||||
- group prefix 必须标识独立 debug group，不复用产品 consumer group；
|
||||
- 命令只读取 Kafka，不发布事件，不启动 Cloud API，不访问数据库，也不依赖 projector。
|
||||
|
||||
- Trace 渲染与 Web 共享最后一个展示分叉之前的生产管线：
|
||||
- JSON frame 使用 `decodeWorkbenchRealtimeEventFrame` 解码；
|
||||
- 事件使用 `reduceWorkbenchRealtimeEvent` 与 `planWorkbenchRealtimeApply` 分类；
|
||||
- 卡片使用 `projectWorkbenchLiveKafkaMessage` 增量归约；
|
||||
- Trace 行使用 `traceDisplayRows` 生成同一 row model；
|
||||
- Web 最后渲染 HTML，CLI 最后使用 `renderTraceRowsMarkdown` 渲染 Markdown；
|
||||
- Final Response 只放在 Trace 外层，Trace 内保留非最终的助手进展消息。
|
||||
|
||||
- Kafka Trace 查询必须披露有界扫描事实：
|
||||
- `--run-id` 与 `--command-id` 用于在复用或异常混合的 traceId 内精确隔离一轮；
|
||||
- `--command-id` 先按 trace、session 与 run 读取基础流，再在 CLI 内形成 command 生命周期；
|
||||
- 只有基础流中至少存在一条精确 commandId 事件，才允许吸附同一 run 的无 commandId final/terminal；
|
||||
- 精确 commandId 不存在时必须返回 `source_command_missing`，不得由 commandless 生命周期冒充成功；
|
||||
- 精确 command 事件保留；
|
||||
- 同一 run 中缺少 commandId 的 final/terminal 生命周期事件保留；
|
||||
- 其他 command 和无法归属的 commandless 事件排除；
|
||||
- 禁止把 commandId 直接下推后丢失 run-scoped terminal/final；
|
||||
- identity 输出必须同时披露观测到的 session、run、command 和 source event 去重计数；
|
||||
- command scope 输出必须披露每个 commandId 的计数、缺失 commandId 数、可归属生命周期数、歧义数与排除数；
|
||||
- command 生命周期只有在 `scanComplete=true` 时才可判定完整；
|
||||
- `limit` 或 `timeout` 只能作为部分流证据，`commandLifecyclePreserved` 必须保持 false；
|
||||
- 输出 `scannedCount`、`parsedCount`、`matchedCount`、`invalidJsonCount` 与 `filterRejectedCount`；
|
||||
- 输出 topic end offsets、最后扫描 offsets 与 `completionReason`；
|
||||
- `completionReason=end-offset` 表示已经扫描到命令启动时捕获的 Kafka 上界；
|
||||
- `completionReason=limit` 表示达到显式事件上限，不能冒充完整 trace；
|
||||
- `completionReason=timeout` 表示未在 YAML 或 CLI 时限内证明到达上界；
|
||||
- 没有匹配事件时以 `source_trace_missing` 失败，并保留上述扫描计数。
|
||||
|
||||
- Trace Markdown artifact 默认写入 `.state/hwlab-cli/kafka-render/`：
|
||||
- `--output-markdown` 可显式选择路径；
|
||||
- `--from jsonl --jsonl-file <path>` 可完全离线复用同一管线；
|
||||
- 默认文本输出保持 compact；
|
||||
- 默认 JSON 只返回 `--row-limit` 指定的尾部行，并披露 returned/omitted/window；
|
||||
- `--full` 仅用于显式下钻完整 row payload；
|
||||
- `--format markdown` 输出 Markdown 正文；
|
||||
- `--json` 输出身份、投影、row model、终态、计数和 artifact SHA 的结构化证据。
|
||||
|
||||
- `hwlab-cli kafka regenerate hwlab` 用于把指定 AgentRun session 的事件直接映射为 HWLAB debug 事件：
|
||||
- 调用生产路径同源的 AgentRun 解码与 HWLAB 事件映射函数；
|
||||
- 不启动 Cloud API；
|
||||
|
||||
@@ -117,6 +117,50 @@ test("projects final AgentRun assistant text without sealing before terminal_sta
|
||||
assert.equal(projected.event.finalResponse.traceId, "trc_final");
|
||||
});
|
||||
|
||||
test("projects structured AgentRun commandExecution visibility into HWLAB tool events", () => {
|
||||
const projected = projectAgentRunKafkaEventToHwlabEvent({
|
||||
schema: "agentrun.event.v1",
|
||||
eventType: "agentrun.run.event",
|
||||
run: { runId: "run_tool", sessionId: "ses_tool" },
|
||||
command: { commandId: "cmd_tool" },
|
||||
event: {
|
||||
seq: 10,
|
||||
type: "tool_call",
|
||||
payload: {
|
||||
traceId: "trc_tool",
|
||||
type: "commandExecution",
|
||||
toolName: "commandExecution",
|
||||
method: "item/completed",
|
||||
itemId: "tool_rg",
|
||||
status: "completed",
|
||||
command: "/bin/bash -lc 'cd /workspace && rg -n target src'",
|
||||
cwd: "/workspace",
|
||||
exitCode: 0,
|
||||
durationMs: 1234,
|
||||
stdoutSummary: "src/main.ts:42:target",
|
||||
stderrSummary: "warning summary",
|
||||
outputSummary: "1 match",
|
||||
summary: {
|
||||
outputBytes: 24,
|
||||
outputTruncated: true
|
||||
}
|
||||
}
|
||||
}
|
||||
}, { source: "hwlab-test" });
|
||||
|
||||
assert.equal(projected.event.toolName, "commandExecution");
|
||||
assert.equal(projected.event.toolType, "commandExecution");
|
||||
assert.equal(projected.event.command, "/bin/bash -lc 'cd /workspace && rg -n target src'");
|
||||
assert.equal(projected.event.cwd, "/workspace");
|
||||
assert.equal(projected.event.exitCode, 0);
|
||||
assert.equal(projected.event.durationMs, 1234);
|
||||
assert.equal(projected.event.stdoutSummary, "src/main.ts:42:target");
|
||||
assert.equal(projected.event.stderrSummary, "warning summary");
|
||||
assert.equal(projected.event.outputSummary, "1 match");
|
||||
assert.equal(projected.event.outputBytes, 24);
|
||||
assert.equal(projected.event.outputTruncated, true);
|
||||
});
|
||||
|
||||
test("projects AgentRun terminal_status Kafka event into terminal HWLAB event", () => {
|
||||
const projected = projectAgentRunKafkaEventToHwlabEvent({
|
||||
schema: "agentrun.event.v1",
|
||||
|
||||
@@ -13,7 +13,7 @@ import {
|
||||
|
||||
const TRUE_VALUES = new Set(["1", "true", "yes", "on"]);
|
||||
export const DEFAULT_AGENTRUN_EVENT_TOPIC = "agentrun.event.v1";
|
||||
const DEFAULT_HWLAB_EVENT_TOPIC = "hwlab.event.v1";
|
||||
export const DEFAULT_HWLAB_EVENT_TOPIC = "hwlab.event.v1";
|
||||
const DEFAULT_STDIO_TOPIC = "codex-stdio.raw.v1";
|
||||
export const DEFAULT_AGENTRUN_DEBUG_EVENT_TOPIC = "agentrun.event.debug.v1";
|
||||
export const DEFAULT_HWLAB_DEBUG_EVENT_TOPIC = "hwlab.event.debug.v1";
|
||||
@@ -933,61 +933,166 @@ export async function queryKafkaEventStream({ env = process.env, stream = "hwlab
|
||||
const resolvedTopic = stringValue(topic) || kafkaTopicForStream(stream, env);
|
||||
const maxEvents = Math.max(1, integerValue(limit) || DEFAULT_QUERY_LIMIT);
|
||||
const budgetMs = Math.max(250, integerValue(timeoutMs) || DEFAULT_QUERY_TIMEOUT_MS);
|
||||
const deadlineMs = Date.now() + budgetMs;
|
||||
const kafka = kafkaFactory({ brokers, clientId: `${clientId}-query` });
|
||||
const endOffsetSnapshot = await kafkaTopicEndOffsetSnapshot(kafka, resolvedTopic, deadlineMs);
|
||||
const endOffsetByPartition = new Map(endOffsetSnapshot.offsets.map((entry) => [entry.partition, entry.endOffset]));
|
||||
const reachedEndPartitions = new Set(endOffsetSnapshot.offsets.filter(kafkaOffsetPartitionIsEmpty).map((entry) => entry.partition));
|
||||
const resolvedGroupIdPrefix = stringValue(groupIdPrefix) || `${clientId}-query`;
|
||||
const groupId = `${resolvedGroupIdPrefix}-${Date.now()}-${randomUUID().slice(0, 8)}`;
|
||||
const consumer = kafka.consumer({ groupId, allowAutoTopicCreation: false });
|
||||
const events = [];
|
||||
const lastScannedOffsetByPartition = new Map();
|
||||
let scannedCount = 0;
|
||||
let parsedCount = 0;
|
||||
let invalidJsonCount = 0;
|
||||
let filterRejectedCount = 0;
|
||||
let completionReason = "timeout";
|
||||
let timer = null;
|
||||
let running = false;
|
||||
await consumer.connect();
|
||||
await consumer.subscribe({ topic: resolvedTopic, fromBeginning: fromBeginning !== false });
|
||||
try {
|
||||
if (remainingKafkaQueryBudget(deadlineMs) <= 0) return kafkaQueryResult();
|
||||
await withinKafkaQueryBudget(consumer.connect(), deadlineMs, "consumer-connect");
|
||||
await withinKafkaQueryBudget(consumer.subscribe({ topic: resolvedTopic, fromBeginning: fromBeginning !== false }), deadlineMs, "consumer-subscribe");
|
||||
const alreadyAtBoundedEnd = fromBeginning !== false && endOffsetSnapshot.available && reachedEndPartitions.size === endOffsetByPartition.size;
|
||||
if (alreadyAtBoundedEnd) completionReason = "end-offset";
|
||||
if (alreadyAtBoundedEnd) return kafkaQueryResult();
|
||||
await new Promise((resolve, reject) => {
|
||||
let finished = false;
|
||||
const finish = (value = null) => {
|
||||
const finish = (reason = "timeout") => {
|
||||
if (finished) return;
|
||||
finished = true;
|
||||
resolve(value);
|
||||
completionReason = reason;
|
||||
resolve(reason);
|
||||
};
|
||||
timer = setTimeout(finish, budgetMs);
|
||||
timer = setTimeout(() => finish("timeout"), Math.max(1, remainingKafkaQueryBudget(deadlineMs)));
|
||||
timer.unref?.();
|
||||
running = true;
|
||||
consumer.run({
|
||||
eachMessage: async ({ topic: messageTopic, partition, message }) => {
|
||||
scannedCount += 1;
|
||||
lastScannedOffsetByPartition.set(partition, stringValue(message.offset));
|
||||
const valueText = message.value ? Buffer.from(message.value).toString("utf8") : "";
|
||||
const value = parseJson(valueText);
|
||||
if (!value || !eventMatchesFilters(value, { traceId, sessionId, runId, commandId })) return;
|
||||
events.push({
|
||||
topic: messageTopic,
|
||||
partition,
|
||||
offset: message.offset,
|
||||
key: message.key ? Buffer.from(message.key).toString("utf8") : null,
|
||||
timestamp: message.timestamp ?? null,
|
||||
valueSha256: sha256(valueText),
|
||||
value
|
||||
});
|
||||
if (events.length >= maxEvents) finish();
|
||||
if (!value || typeof value !== "object" || Array.isArray(value)) invalidJsonCount += 1;
|
||||
else {
|
||||
parsedCount += 1;
|
||||
if (!eventMatchesFilters(value, { traceId, sessionId, runId, commandId })) filterRejectedCount += 1;
|
||||
else {
|
||||
events.push({
|
||||
topic: messageTopic,
|
||||
partition,
|
||||
offset: message.offset,
|
||||
key: message.key ? Buffer.from(message.key).toString("utf8") : null,
|
||||
timestamp: message.timestamp ?? null,
|
||||
valueSha256: sha256(valueText),
|
||||
value
|
||||
});
|
||||
}
|
||||
}
|
||||
if (events.length >= maxEvents) return finish("limit");
|
||||
const endOffset = endOffsetByPartition.get(partition);
|
||||
if (fromBeginning !== false && endOffset && kafkaOffsetReached(message.offset, endOffset)) reachedEndPartitions.add(partition);
|
||||
if (endOffsetSnapshot.available && reachedEndPartitions.size === endOffsetByPartition.size) finish("end-offset");
|
||||
}
|
||||
}).catch(reject);
|
||||
});
|
||||
} finally {
|
||||
if (timer) clearTimeout(timer);
|
||||
if (running) await consumer.stop().catch(() => undefined);
|
||||
await boundedDisconnect(consumer);
|
||||
if (running) await boundedKafkaPromise(consumer.stop(), remainingKafkaQueryCleanupBudget(deadlineMs)).catch(() => undefined);
|
||||
await boundedDisconnect(consumer, remainingKafkaQueryCleanupBudget(deadlineMs));
|
||||
}
|
||||
return kafkaQueryResult();
|
||||
|
||||
function kafkaQueryResult() {
|
||||
return {
|
||||
ok: true,
|
||||
stream,
|
||||
topic: resolvedTopic,
|
||||
groupId,
|
||||
count: events.length,
|
||||
scannedCount,
|
||||
parsedCount,
|
||||
matchedCount: events.length,
|
||||
invalidJsonCount,
|
||||
filterRejectedCount,
|
||||
completionReason,
|
||||
reachedEndOffsets: completionReason === "end-offset",
|
||||
endOffsets: endOffsetSnapshot.offsets,
|
||||
endOffsetsAvailable: endOffsetSnapshot.available,
|
||||
endOffsetsError: endOffsetSnapshot.error,
|
||||
lastScannedOffsets: [...lastScannedOffsetByPartition.entries()].map(([partition, offset]) => ({ partition, offset })),
|
||||
limit: maxEvents,
|
||||
timeoutMs: budgetMs,
|
||||
filters: compactObject({ traceId, sessionId, runId, commandId }),
|
||||
events,
|
||||
valuesPrinted: false
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
async function kafkaTopicEndOffsetSnapshot(kafka, topic, deadlineMs) {
|
||||
if (typeof kafka?.admin !== "function") return { available: false, offsets: [], error: { code: "kafka_admin_unavailable", valuesRedacted: true } };
|
||||
const admin = kafka.admin();
|
||||
try {
|
||||
await withinKafkaQueryBudget(admin.connect(), deadlineMs, "admin-connect");
|
||||
const fetched = await withinKafkaQueryBudget(admin.fetchTopicOffsets(topic), deadlineMs, "fetch-topic-offsets");
|
||||
const offsets = Array.isArray(fetched)
|
||||
? fetched.map((entry) => ({
|
||||
partition: integerValue(entry.partition),
|
||||
startOffset: kafkaOffsetValue(entry.low),
|
||||
endOffset: kafkaOffsetValue(entry.high ?? entry.offset)
|
||||
})).filter((entry) => entry.partition !== null && entry.endOffset !== null)
|
||||
: [];
|
||||
if (offsets.length === 0) return { available: false, offsets: [], error: { code: "kafka_end_offsets_empty", valuesRedacted: true } };
|
||||
return {
|
||||
available: true,
|
||||
offsets,
|
||||
error: null
|
||||
};
|
||||
} catch (error) {
|
||||
return { available: false, offsets: [], error: { code: error?.code ?? "kafka_end_offsets_failed", message: errorMessage(error), valuesRedacted: true } };
|
||||
} finally {
|
||||
await boundedDisconnect(admin, remainingKafkaQueryCleanupBudget(deadlineMs));
|
||||
}
|
||||
}
|
||||
|
||||
function kafkaOffsetPartitionIsEmpty(entry) {
|
||||
if (entry.startOffset !== null) return entry.startOffset === entry.endOffset;
|
||||
return entry.endOffset === "0";
|
||||
}
|
||||
|
||||
function remainingKafkaQueryBudget(deadlineMs) {
|
||||
return Math.max(0, deadlineMs - Date.now());
|
||||
}
|
||||
|
||||
function remainingKafkaQueryCleanupBudget(deadlineMs) {
|
||||
return Math.max(1, remainingKafkaQueryBudget(deadlineMs));
|
||||
}
|
||||
|
||||
async function withinKafkaQueryBudget(promise, deadlineMs, stage) {
|
||||
const remainingMs = remainingKafkaQueryBudget(deadlineMs);
|
||||
if (remainingMs <= 0) throw contractError("kafka_query_timeout", `Kafka query timed out before ${stage}.`);
|
||||
let timer = null;
|
||||
try {
|
||||
return await Promise.race([
|
||||
promise,
|
||||
new Promise((_, reject) => {
|
||||
timer = setTimeout(() => reject(contractError("kafka_query_timeout", `Kafka query timed out during ${stage}.`)), remainingMs);
|
||||
timer.unref?.();
|
||||
})
|
||||
]);
|
||||
} finally {
|
||||
if (timer) clearTimeout(timer);
|
||||
}
|
||||
}
|
||||
|
||||
function kafkaOffsetReached(offset, endOffset) {
|
||||
try {
|
||||
return BigInt(String(offset)) + 1n >= BigInt(String(endOffset));
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
return {
|
||||
ok: true,
|
||||
stream,
|
||||
topic: resolvedTopic,
|
||||
groupId,
|
||||
count: events.length,
|
||||
limit: maxEvents,
|
||||
timeoutMs: budgetMs,
|
||||
filters: compactObject({ traceId, sessionId, runId, commandId }),
|
||||
events,
|
||||
valuesPrinted: false
|
||||
};
|
||||
}
|
||||
|
||||
export async function openKafkaEventStream({ env = process.env, stream = "hwlab", topic = null, groupIdPrefix = null, traceId = null, sessionId = null, runId = null, commandId = null, replayId = null, offsetRange = null, fromBeginning = false, onEvent, onRecord = null, onBarrierReady = null, onBarrierComplete = null, onError = null, kafkaFactory = defaultKafkaFactory } = {}) {
|
||||
@@ -1122,8 +1227,13 @@ function mapAgentRunSourceEventToHwlabEvent({ input, sourceEvent, payload, run,
|
||||
assistantText,
|
||||
finalResponse: authoritative ? { text: assistantText, status: "completed", traceId, source: "agentrun.kafka", valuesPrinted: false } : null,
|
||||
itemId: firstText(payload.itemId),
|
||||
assistantSource: firstText(payload.source),
|
||||
messageIndex: integerValue(payload.messageIndex),
|
||||
messageCount: integerValue(payload.messageCount),
|
||||
replyAuthority: payload.replyAuthority === true,
|
||||
final: payload.final === true,
|
||||
textTruncated: payload.textTruncated === true,
|
||||
outputTruncated: payload.outputTruncated === true,
|
||||
terminal: false
|
||||
};
|
||||
}
|
||||
@@ -1146,11 +1256,20 @@ function mapAgentRunSourceEventToHwlabEvent({ input, sourceEvent, payload, run,
|
||||
label: `agentrun:tool:${toolName}`,
|
||||
message: textPayload(payload, `AgentRun tool ${toolName}`),
|
||||
toolName,
|
||||
toolType: firstText(payload.type),
|
||||
method: firstText(payload.method),
|
||||
itemId: firstText(payload.itemId),
|
||||
command: firstText(payload.command, payload.commandLine),
|
||||
cwd: firstText(payload.cwd),
|
||||
processId: firstText(payload.processId),
|
||||
exitCode: integerValue(payload.exitCode),
|
||||
durationMs: integerValue(payload.durationMs),
|
||||
outputSummary: firstText(payload.outputSummary, payload.summary?.text),
|
||||
stdoutSummary: firstText(payload.stdoutSummary),
|
||||
stderrSummary: firstText(payload.stderrSummary),
|
||||
commandTruncated: payload.commandTruncated === true,
|
||||
outputBytes: integerValue(payload.outputBytes ?? payload.summary?.outputBytes),
|
||||
outputTruncated: payload.outputTruncated === true || payload.summary?.outputTruncated === true,
|
||||
terminal: false
|
||||
};
|
||||
}
|
||||
@@ -1244,12 +1363,16 @@ function defaultKafkaFactory(config) {
|
||||
}
|
||||
|
||||
async function boundedDisconnect(consumer, timeoutMs = 2000) {
|
||||
return boundedKafkaPromise(consumer.disconnect(), timeoutMs);
|
||||
}
|
||||
|
||||
async function boundedKafkaPromise(promise, timeoutMs) {
|
||||
let timer = null;
|
||||
try {
|
||||
await Promise.race([
|
||||
consumer.disconnect(),
|
||||
promise,
|
||||
new Promise((resolve) => {
|
||||
timer = setTimeout(resolve, timeoutMs);
|
||||
timer = setTimeout(resolve, Math.max(1, timeoutMs));
|
||||
timer.unref?.();
|
||||
})
|
||||
]);
|
||||
|
||||
@@ -7,6 +7,7 @@ import { test } from "bun:test";
|
||||
|
||||
import { queryKafkaEventStream } from "../../internal/cloud/kafka-event-bridge.ts";
|
||||
import { mapAgentRunRecordsToHwlabDebugEvents, runKafkaCli } from "../src/hwlab-cli/kafka-regenerate.ts";
|
||||
import { traceDisplayRows } from "../src/hwlab-cli/trace-renderer.ts";
|
||||
|
||||
const SESSION_ID = "ses_agentrun_5ec4e141_6abc_466d_9afe_049f7c0ac105";
|
||||
const HWLAB_SESSION_ID = "ses_5ec4e141-6abc-466d-9afe-049f7c0ac105";
|
||||
@@ -16,8 +17,294 @@ const RUN_ID = "run_a78cbcb05f2c4afaa748a3158db44998";
|
||||
test("Kafka help shows canonical default input and explicit reconstruction debug input", async () => {
|
||||
const result = await runKafkaCli(["help", "--json"], { env: {}, now: () => "2026-07-10T12:00:00.000Z" });
|
||||
assert.equal(result.exitCode, 0);
|
||||
assert.match(result.payload.commands[0], /--input-topic agentrun\.event\.v1/u);
|
||||
assert.match(result.payload.commands[1], /--input-topic agentrun\.event\.debug\.v1/u);
|
||||
assert.ok(result.payload.commands.some((command: string) => /regenerate hwlab.*--input-topic agentrun\.event\.v1/u.test(command)));
|
||||
assert.ok(result.payload.commands.some((command: string) => /regenerate hwlab.*--input-topic agentrun\.event\.debug\.v1/u.test(command)));
|
||||
assert.ok(result.payload.commands.some((command: string) => /render trace.*--input-topic hwlab\.event\.v1/u.test(command)));
|
||||
});
|
||||
|
||||
test("offline HWLAB JSONL uses the shared Web pipeline and renders final response only outside Trace", async () => {
|
||||
const cwd = await mkdtemp(path.join(os.tmpdir(), "hwlab-kafka-trace-render-"));
|
||||
const inputFile = path.join(cwd, "hwlab-events.jsonl");
|
||||
const outputFile = path.join(cwd, "trace.md");
|
||||
const finalText = "Hi. What do you want to work on?";
|
||||
const records = [
|
||||
hwlabRecord(1, { type: "tool", eventType: "tool", status: "started", label: "agentrun:tool:commandExecution", toolName: "commandExecution", itemId: "tool_rg", command: "/bin/bash -lc 'cd /workspace && rg -n target src'" }),
|
||||
hwlabRecord(2, { type: "tool", eventType: "tool", status: "completed", label: "agentrun:tool:commandExecution", toolName: "commandExecution", itemId: "tool_rg", command: "/bin/bash -lc 'cd /workspace && rg -n target src'", durationMs: 1234, exitCode: 0, stdoutSummary: "src/main.ts:42:target" }),
|
||||
hwlabRecord(3, { type: "assistant", eventType: "assistant", status: "running", label: "agentrun:assistant:message", itemId: "assistant_progress", message: "I am checking the workspace.", assistantText: "I am checking the workspace.", assistantSource: "agent-message" }),
|
||||
hwlabRecord(4, { type: "assistant", eventType: "assistant", status: "running", label: "agentrun:assistant:message", itemId: "assistant_final", message: finalText, assistantText: finalText, assistantSource: "completed-agent-message" }),
|
||||
hwlabRecord(5, { type: "assistant", eventType: "assistant", status: "running", label: "agentrun:assistant:message", itemId: "assistant_final", message: finalText, assistantText: finalText, assistantSource: "completed-agent-message-final", replyAuthority: true, final: true, finalResponse: { text: finalText } }),
|
||||
hwlabRecord(6, { type: "result", eventType: "terminal", status: "completed", label: "agentrun:terminal:completed", terminal: true, message: "AgentRun completed" })
|
||||
];
|
||||
await writeFile(inputFile, records.map((record) => JSON.stringify(record)).join("\n") + "\n", "utf8");
|
||||
|
||||
const result = await runKafkaCli([
|
||||
"render", "trace",
|
||||
"--from", "jsonl",
|
||||
"--trace-id", TRACE_ID,
|
||||
"--session-id", HWLAB_SESSION_ID,
|
||||
"--run-id", RUN_ID,
|
||||
"--command-id", "cmd_render_trace",
|
||||
"--jsonl-file", inputFile,
|
||||
"--output-markdown", outputFile,
|
||||
"--format", "markdown"
|
||||
], { cwd, env: {}, now: () => "2026-07-10T12:00:00.000Z" });
|
||||
|
||||
assert.equal(result.exitCode, 0);
|
||||
assert.equal(result.payload.input.scannedCount, 6);
|
||||
assert.equal(result.payload.input.matchedCount, 6);
|
||||
assert.equal(result.payload.projection.decodedCount, 6);
|
||||
assert.equal(result.payload.projection.appliedCount, 6);
|
||||
assert.equal(result.payload.projection.terminalObserved, true);
|
||||
assert.equal(result.payload.projection.lastEventAt, "2026-07-10T12:00:00.005Z");
|
||||
assert.deepEqual(result.payload.identity.observedRunIds, [RUN_ID]);
|
||||
assert.deepEqual(result.payload.identity.observedCommandIds, ["cmd_render_trace"]);
|
||||
assert.equal(result.payload.identity.sourceEventIdentityDuplicates, 0);
|
||||
assert.equal(result.payload.render.toolRowCount, 1);
|
||||
assert.equal(result.payload.render.assistantRowCount, 1);
|
||||
assert.equal(result.payload.render.finalResponsePresent, true);
|
||||
assert.equal(result.payload.render.finalResponseInTrace, false);
|
||||
assert.equal(result.payload.render.rowsReturned, result.payload.render.rowCount);
|
||||
assert.equal(result.payload.render.rowsOmitted, 0);
|
||||
assert.equal(result.payload.validation.allMatchedEventsApplied, true);
|
||||
assert.equal(result.payload.validation.finalResponseOutsideTrace, true);
|
||||
const toolRow = result.payload.render.rows.find((row: any) => row.rowId.startsWith("tool:"));
|
||||
assert.match(toolRow.header, /ok rg duration=00:00:01 exit=0/u);
|
||||
assert.match(toolRow.preview, /rg -n target src/u);
|
||||
assert.match(toolRow.preview, /src\/main\.ts:42:target/u);
|
||||
const markdown = await readFile(outputFile, "utf8");
|
||||
assert.equal((markdown.match(new RegExp(finalText.replace(/[.*+?^${}()|[\]\\]/gu, "\\$&"), "gu")) ?? []).length, 1);
|
||||
assert.match(markdown, /## 运行记录/u);
|
||||
assert.match(markdown, /## 最终回复/u);
|
||||
assert.equal(result.markdownOutput?.trim(), markdown.trim());
|
||||
});
|
||||
|
||||
test("outer final response stays out of the Trace fallback when no completion event exists", () => {
|
||||
const finalText = "Hi. What do you want to work on?";
|
||||
const rows = traceDisplayRows({ eventSource: "hwlab-kafka-sse" }, [
|
||||
{
|
||||
sourceSeq: 1,
|
||||
sourceEventId: "evt_same_text",
|
||||
type: "assistant",
|
||||
eventType: "assistant",
|
||||
status: "running",
|
||||
label: "agentrun:assistant:message",
|
||||
itemId: "assistant_same_text",
|
||||
message: finalText,
|
||||
assistantText: finalText,
|
||||
assistantSource: "completed-agent-message"
|
||||
},
|
||||
{
|
||||
sourceSeq: 2,
|
||||
sourceEventId: "evt_final",
|
||||
type: "assistant",
|
||||
eventType: "assistant",
|
||||
status: "running",
|
||||
label: "agentrun:assistant:message",
|
||||
itemId: "assistant_final",
|
||||
message: finalText,
|
||||
assistantText: finalText,
|
||||
assistantSource: "completed-agent-message-final",
|
||||
finalResponse: { text: finalText },
|
||||
final: true,
|
||||
replyAuthority: true
|
||||
}
|
||||
], {
|
||||
finalResponsePlacement: "outer",
|
||||
outerFinalResponseText: finalText
|
||||
});
|
||||
|
||||
assert.deepEqual(rows, []);
|
||||
});
|
||||
|
||||
test("trace JSON defaults to a bounded tail while the Markdown artifact stays complete", async () => {
|
||||
const cwd = await mkdtemp(path.join(os.tmpdir(), "hwlab-kafka-trace-window-"));
|
||||
const inputFile = path.join(cwd, "hwlab-events.jsonl");
|
||||
const records = Array.from({ length: 25 }, (_, index) => hwlabRecord(index + 1, {
|
||||
type: "assistant",
|
||||
eventType: "assistant",
|
||||
status: "running",
|
||||
label: "agentrun:assistant:message",
|
||||
itemId: `assistant_${index + 1}`,
|
||||
message: `progress ${index + 1}`,
|
||||
assistantText: `progress ${index + 1}`
|
||||
}));
|
||||
records.push(hwlabRecord(26, { type: "result", eventType: "terminal", status: "completed", label: "agentrun:terminal:completed", terminal: true }));
|
||||
await writeFile(inputFile, records.map((record) => JSON.stringify(record)).join("\n") + "\n", "utf8");
|
||||
|
||||
const result = await runKafkaCli([
|
||||
"render", "trace",
|
||||
"--from", "jsonl",
|
||||
"--trace-id", TRACE_ID,
|
||||
"--jsonl-file", inputFile,
|
||||
"--row-limit", "5",
|
||||
"--format", "markdown"
|
||||
], { cwd, env: {}, now: () => "2026-07-10T12:00:00.000Z" });
|
||||
|
||||
assert.equal(result.exitCode, 0);
|
||||
assert.equal(result.payload.render.rowCount, 25);
|
||||
assert.equal(result.payload.render.rowsReturned, 5);
|
||||
assert.equal(result.payload.render.rowsOmitted, 20);
|
||||
assert.equal(result.payload.render.rowWindow, "tail");
|
||||
assert.match(result.markdownOutput, /progress 1/u);
|
||||
assert.match(result.markdownOutput, /progress 24/u);
|
||||
});
|
||||
|
||||
test("command-scoped trace keeps attributable commandless final and terminal lifecycle events", async () => {
|
||||
const cwd = await mkdtemp(path.join(os.tmpdir(), "hwlab-kafka-command-scope-"));
|
||||
const finalText = "Hi. What do you want to work on?";
|
||||
const exactTool = hwlabRecord(1, {
|
||||
type: "tool",
|
||||
eventType: "tool",
|
||||
status: "completed",
|
||||
label: "agentrun:tool:commandExecution",
|
||||
toolName: "commandExecution",
|
||||
itemId: "tool_command_scope",
|
||||
command: "rg -n target src",
|
||||
exitCode: 0
|
||||
});
|
||||
const exactProgress = hwlabRecord(2, {
|
||||
type: "assistant",
|
||||
eventType: "assistant",
|
||||
status: "running",
|
||||
itemId: "assistant_progress",
|
||||
assistantText: "Checking the workspace."
|
||||
});
|
||||
const unscopedFinal = withoutCommandId(hwlabRecord(3, {
|
||||
type: "assistant",
|
||||
eventType: "assistant",
|
||||
status: "running",
|
||||
itemId: "assistant_final",
|
||||
assistantText: finalText,
|
||||
finalResponse: { text: finalText },
|
||||
final: true,
|
||||
replyAuthority: true
|
||||
}));
|
||||
const unscopedTerminal = withoutCommandId(hwlabRecord(4, {
|
||||
type: "result",
|
||||
eventType: "terminal",
|
||||
status: "completed",
|
||||
terminal: true,
|
||||
message: "AgentRun completed"
|
||||
}));
|
||||
const runnerShutdown = withCommandId(hwlabRecord(5, {
|
||||
type: "backend",
|
||||
eventType: "backend",
|
||||
status: "running",
|
||||
message: "runner shutdown"
|
||||
}), "runner-shutdown");
|
||||
const unscopedBackend = withoutCommandId(hwlabRecord(6, {
|
||||
type: "backend",
|
||||
eventType: "backend",
|
||||
status: "running",
|
||||
message: "unscoped heartbeat"
|
||||
}));
|
||||
let readerInput: any = null;
|
||||
const result = await runKafkaCli([
|
||||
"render", "trace",
|
||||
"--from", "kafka",
|
||||
"--trace-id", TRACE_ID,
|
||||
"--session-id", HWLAB_SESSION_ID,
|
||||
"--run-id", RUN_ID,
|
||||
"--command-id", "cmd_render_trace",
|
||||
"--group-prefix", "hwlab-v03-workbench-isolated-debug",
|
||||
"--row-limit", "2",
|
||||
"--format", "markdown"
|
||||
], {
|
||||
cwd,
|
||||
env: {},
|
||||
now: () => "2026-07-10T12:00:00.000Z",
|
||||
async readKafka(input) {
|
||||
readerInput = input;
|
||||
return {
|
||||
events: [exactTool, exactProgress, unscopedFinal, unscopedTerminal, runnerShutdown, unscopedBackend],
|
||||
scannedCount: 6,
|
||||
parsedCount: 6,
|
||||
matchedCount: 6,
|
||||
filterRejectedCount: 0,
|
||||
completionReason: "end-offset",
|
||||
reachedEndOffsets: true
|
||||
};
|
||||
}
|
||||
});
|
||||
|
||||
assert.equal(result.exitCode, 0);
|
||||
assert.equal(readerInput.commandId, undefined, "commandId must not be pushed down before lifecycle attribution");
|
||||
assert.equal(readerInput.runId, RUN_ID);
|
||||
assert.equal(result.payload.input.queryMatchedCount, 6);
|
||||
assert.equal(result.payload.input.matchedCount, 4);
|
||||
assert.deepEqual(result.payload.input.commandScope.commandIdCounts, [
|
||||
{ commandId: "cmd_render_trace", count: 2 },
|
||||
{ commandId: "runner-shutdown", count: 1 }
|
||||
]);
|
||||
assert.equal(result.payload.input.commandScope.missingCommandIdCount, 3);
|
||||
assert.equal(result.payload.input.commandScope.unscopedLifecycleCount, 2);
|
||||
assert.equal(result.payload.input.commandScope.selectedUnscopedLifecycleCount, 2);
|
||||
assert.equal(result.payload.input.commandScope.unscopedNonLifecycleCount, 1);
|
||||
assert.equal(result.payload.input.commandScope.otherCommandRecordCount, 1);
|
||||
assert.equal(result.payload.projection.eventCount, 4);
|
||||
assert.equal(result.payload.projection.terminalObserved, true);
|
||||
assert.equal(result.payload.render.finalResponseInTrace, false);
|
||||
assert.equal(result.payload.validation.commandLifecyclePreserved, true);
|
||||
assert.equal((result.markdownOutput?.match(/Hi\. What do you want to work on\?/gu) ?? []).length, 1);
|
||||
});
|
||||
|
||||
test("missing commandId cannot be satisfied by commandless lifecycle events from the same run", async () => {
|
||||
const otherCommand = withCommandId(hwlabRecord(1, {
|
||||
type: "tool",
|
||||
eventType: "tool",
|
||||
status: "completed",
|
||||
label: "agentrun:tool:commandExecution",
|
||||
itemId: "tool_other_command",
|
||||
command: "rg -n target src",
|
||||
exitCode: 0
|
||||
}), "cmd_other");
|
||||
const commandlessFinal = withoutCommandId(hwlabRecord(2, {
|
||||
type: "assistant",
|
||||
eventType: "assistant",
|
||||
status: "running",
|
||||
itemId: "assistant_final",
|
||||
assistantText: "This belongs to the existing command.",
|
||||
finalResponse: { text: "This belongs to the existing command." },
|
||||
final: true,
|
||||
replyAuthority: true
|
||||
}));
|
||||
const commandlessTerminal = withoutCommandId(hwlabRecord(3, {
|
||||
type: "result",
|
||||
eventType: "terminal",
|
||||
status: "completed",
|
||||
terminal: true
|
||||
}));
|
||||
|
||||
const result = await runKafkaCli([
|
||||
"render", "trace",
|
||||
"--from", "kafka",
|
||||
"--trace-id", TRACE_ID,
|
||||
"--session-id", HWLAB_SESSION_ID,
|
||||
"--run-id", RUN_ID,
|
||||
"--command-id", "cmd_missing",
|
||||
"--group-prefix", "hwlab-v03-workbench-isolated-debug",
|
||||
"--format", "markdown"
|
||||
], {
|
||||
env: {},
|
||||
now: () => "2026-07-10T12:00:00.000Z",
|
||||
async readKafka() {
|
||||
return {
|
||||
events: [otherCommand, commandlessFinal, commandlessTerminal],
|
||||
scannedCount: 3,
|
||||
parsedCount: 3,
|
||||
matchedCount: 3,
|
||||
filterRejectedCount: 0,
|
||||
completionReason: "end-offset",
|
||||
reachedEndOffsets: true
|
||||
};
|
||||
}
|
||||
});
|
||||
|
||||
assert.equal(result.exitCode, 1);
|
||||
assert.equal(result.payload.error.code, "source_command_missing");
|
||||
assert.equal(result.payload.error.details.commandScope.exactCommandCount, 0);
|
||||
assert.equal(result.payload.error.details.commandScope.selectedCount, 0);
|
||||
assert.equal(result.payload.error.details.commandScope.selectedUnscopedLifecycleCount, 0);
|
||||
});
|
||||
|
||||
test("offline JSONL maps 35 stdio reconstructions to 35 ordered HWLAB debug events without runtime dependencies", async () => {
|
||||
@@ -325,6 +612,170 @@ test("Kafka query exposes the isolated generated group and original value hash",
|
||||
assert.match(queried.groupId, /^hwlab-v03-workbench-isolated-debug-\d+-[a-f0-9]{8}$/u);
|
||||
assert.equal(consumerConfigs[0].groupId, queried.groupId);
|
||||
assert.equal(queried.events[0].valueSha256, createSha256(valueText));
|
||||
assert.equal(queried.scannedCount, 1);
|
||||
assert.equal(queried.parsedCount, 1);
|
||||
assert.equal(queried.matchedCount, 1);
|
||||
assert.equal(queried.invalidJsonCount, 0);
|
||||
assert.equal(queried.completionReason, "limit");
|
||||
assert.equal(queried.endOffsetsAvailable, false);
|
||||
});
|
||||
|
||||
test("Kafka query reports invalid JSON, filter rejection, and bounded end-offset completion", async () => {
|
||||
const matching = canonicalRecord(1).value;
|
||||
const mismatch = structuredClone(matching);
|
||||
mismatch.sessionId = "ses_agentrun_other";
|
||||
mismatch.run.sessionId = "ses_agentrun_other";
|
||||
mismatch.event.payload.sessionId = "ses_agentrun_other";
|
||||
const consumer = {
|
||||
async connect() {},
|
||||
async subscribe() {},
|
||||
async run({ eachMessage }: any) {
|
||||
await eachMessage({ topic: "agentrun.event.v1", partition: 0, message: { offset: "0", value: Buffer.from("{invalid") } });
|
||||
await eachMessage({ topic: "agentrun.event.v1", partition: 0, message: { offset: "1", value: Buffer.from(JSON.stringify(mismatch)) } });
|
||||
await eachMessage({ topic: "agentrun.event.v1", partition: 0, message: { offset: "2", value: Buffer.from(JSON.stringify(matching)) } });
|
||||
},
|
||||
async stop() {},
|
||||
async disconnect() {}
|
||||
};
|
||||
const queried = await queryKafkaEventStream({
|
||||
env: { HWLAB_KAFKA_BOOTSTRAP_SERVERS: "kafka.test:9092", HWLAB_KAFKA_CLIENT_ID: "hwlab-test" },
|
||||
topic: "agentrun.event.v1",
|
||||
sessionId: SESSION_ID,
|
||||
limit: 2,
|
||||
groupIdPrefix: "hwlab-v03-workbench-isolated-debug",
|
||||
kafkaFactory() {
|
||||
return {
|
||||
admin() { return { async connect() {}, async fetchTopicOffsets() { return [{ partition: 0, low: "0", high: "3" }]; }, async disconnect() {} }; },
|
||||
consumer() { return consumer; }
|
||||
};
|
||||
}
|
||||
});
|
||||
assert.equal(queried.scannedCount, 3);
|
||||
assert.equal(queried.parsedCount, 2);
|
||||
assert.equal(queried.invalidJsonCount, 1);
|
||||
assert.equal(queried.filterRejectedCount, 1);
|
||||
assert.equal(queried.matchedCount, 1);
|
||||
assert.equal(queried.completionReason, "end-offset");
|
||||
assert.equal(queried.reachedEndOffsets, true);
|
||||
assert.deepEqual(queried.endOffsets, [{ partition: 0, startOffset: "0", endOffset: "3" }]);
|
||||
assert.deepEqual(queried.lastScannedOffsets, [{ partition: 0, offset: "2" }]);
|
||||
});
|
||||
|
||||
test("Kafka query treats a retained empty partition as a completed bounded scan", async () => {
|
||||
let runCalled = false;
|
||||
let disconnected = false;
|
||||
const queried = await queryKafkaEventStream({
|
||||
env: { HWLAB_KAFKA_BOOTSTRAP_SERVERS: "kafka.test:9092", HWLAB_KAFKA_CLIENT_ID: "hwlab-test" },
|
||||
topic: "hwlab.event.v1",
|
||||
traceId: TRACE_ID,
|
||||
timeoutMs: 250,
|
||||
groupIdPrefix: "hwlab-v03-workbench-isolated-debug",
|
||||
kafkaFactory() {
|
||||
return {
|
||||
admin() { return { async connect() {}, async fetchTopicOffsets() { return [{ partition: 0, low: "7", high: "7" }]; }, async disconnect() {} }; },
|
||||
consumer() {
|
||||
return {
|
||||
async connect() {},
|
||||
async subscribe() {},
|
||||
async run() { runCalled = true; },
|
||||
async stop() {},
|
||||
async disconnect() { disconnected = true; }
|
||||
};
|
||||
}
|
||||
};
|
||||
}
|
||||
});
|
||||
assert.equal(queried.completionReason, "end-offset");
|
||||
assert.equal(queried.reachedEndOffsets, true);
|
||||
assert.equal(queried.scannedCount, 0);
|
||||
assert.equal(runCalled, false);
|
||||
assert.equal(disconnected, true);
|
||||
assert.deepEqual(queried.endOffsets, [{ partition: 0, startOffset: "7", endOffset: "7" }]);
|
||||
});
|
||||
|
||||
test("Kafka query does not report an empty offset response as a completed scan", async () => {
|
||||
const queried = await queryKafkaEventStream({
|
||||
env: { HWLAB_KAFKA_BOOTSTRAP_SERVERS: "kafka.test:9092", HWLAB_KAFKA_CLIENT_ID: "hwlab-test" },
|
||||
topic: "hwlab.event.v1",
|
||||
traceId: TRACE_ID,
|
||||
timeoutMs: 250,
|
||||
groupIdPrefix: "hwlab-v03-workbench-isolated-debug",
|
||||
kafkaFactory() {
|
||||
return {
|
||||
admin() { return { async connect() {}, async fetchTopicOffsets() { return []; }, async disconnect() {} }; },
|
||||
consumer() { return { async connect() {}, async subscribe() {}, async run() {}, async stop() {}, async disconnect() {} }; }
|
||||
};
|
||||
}
|
||||
});
|
||||
assert.equal(queried.completionReason, "timeout");
|
||||
assert.equal(queried.reachedEndOffsets, false);
|
||||
assert.equal(queried.endOffsetsAvailable, false);
|
||||
assert.equal(queried.endOffsetsError?.code, "kafka_end_offsets_empty");
|
||||
});
|
||||
|
||||
test("Kafka query disconnects a connected consumer when subscribe fails", async () => {
|
||||
let disconnected = 0;
|
||||
await assert.rejects(queryKafkaEventStream({
|
||||
env: { HWLAB_KAFKA_BOOTSTRAP_SERVERS: "kafka.test:9092", HWLAB_KAFKA_CLIENT_ID: "hwlab-test" },
|
||||
topic: "hwlab.event.v1",
|
||||
traceId: TRACE_ID,
|
||||
timeoutMs: 250,
|
||||
groupIdPrefix: "hwlab-v03-workbench-isolated-debug",
|
||||
kafkaFactory() {
|
||||
return {
|
||||
consumer() {
|
||||
return {
|
||||
async connect() {},
|
||||
async subscribe() { throw new Error("subscribe failed"); },
|
||||
async run() {},
|
||||
async stop() {},
|
||||
async disconnect() { disconnected += 1; }
|
||||
};
|
||||
}
|
||||
};
|
||||
}
|
||||
}), /subscribe failed/u);
|
||||
assert.equal(disconnected, 1);
|
||||
});
|
||||
|
||||
test("Kafka query setup and cleanup stay inside the command budget", async () => {
|
||||
let consumerConnectCalled = false;
|
||||
let consumerDisconnected = false;
|
||||
const startedAt = Date.now();
|
||||
const queried = await queryKafkaEventStream({
|
||||
env: { HWLAB_KAFKA_BOOTSTRAP_SERVERS: "kafka.test:9092", HWLAB_KAFKA_CLIENT_ID: "hwlab-test" },
|
||||
topic: "hwlab.event.v1",
|
||||
traceId: TRACE_ID,
|
||||
timeoutMs: 250,
|
||||
groupIdPrefix: "hwlab-v03-workbench-isolated-debug",
|
||||
kafkaFactory() {
|
||||
return {
|
||||
admin() {
|
||||
return {
|
||||
async connect() { await new Promise(() => undefined); },
|
||||
async fetchTopicOffsets() { throw new Error("must not fetch after setup timeout"); },
|
||||
async disconnect() {}
|
||||
};
|
||||
},
|
||||
consumer() {
|
||||
return {
|
||||
async connect() { consumerConnectCalled = true; },
|
||||
async subscribe() {},
|
||||
async run() {},
|
||||
async stop() {},
|
||||
async disconnect() { consumerDisconnected = true; }
|
||||
};
|
||||
}
|
||||
};
|
||||
}
|
||||
});
|
||||
const elapsedMs = Date.now() - startedAt;
|
||||
assert.equal(queried.completionReason, "timeout");
|
||||
assert.equal(queried.endOffsetsAvailable, false);
|
||||
assert.equal(queried.endOffsetsError?.code, "kafka_query_timeout");
|
||||
assert.equal(consumerConnectCalled, false);
|
||||
assert.equal(consumerDisconnected, true);
|
||||
assert.ok(elapsedMs < 500, `query exceeded bounded setup budget: ${elapsedMs}ms`);
|
||||
});
|
||||
|
||||
test("non-reversible AgentRun session identity stays only in debug lineage", () => {
|
||||
@@ -465,6 +916,51 @@ function canonicalRecord(seq: number): any {
|
||||
return { topic: "agentrun.event.v1", partition: 0, offset: String(1600 + seq), key: SESSION_ID, value };
|
||||
}
|
||||
|
||||
function hwlabRecord(sourceSeq: number, event: Record<string, unknown>): any {
|
||||
return {
|
||||
topic: "hwlab.event.v1",
|
||||
partition: 0,
|
||||
offset: String(1900 + sourceSeq),
|
||||
key: HWLAB_SESSION_ID,
|
||||
value: {
|
||||
schema: "hwlab.event.v1",
|
||||
eventType: "hwlab.trace.event.projected",
|
||||
eventId: `hwlab:evt_render_${sourceSeq}`,
|
||||
sourceEventId: `evt_render_${sourceSeq}`,
|
||||
traceId: TRACE_ID,
|
||||
hwlabSessionId: HWLAB_SESSION_ID,
|
||||
sessionId: HWLAB_SESSION_ID,
|
||||
runId: RUN_ID,
|
||||
commandId: "cmd_render_trace",
|
||||
serverSentAt: `2026-07-10T11:00:0${sourceSeq}.000Z`,
|
||||
event: {
|
||||
traceId: TRACE_ID,
|
||||
sessionId: HWLAB_SESSION_ID,
|
||||
source: "agentrun.kafka",
|
||||
sourceSeq,
|
||||
sourceEventId: `evt_render_${sourceSeq}`,
|
||||
runId: RUN_ID,
|
||||
commandId: "cmd_render_trace",
|
||||
createdAt: `2026-07-10T11:00:0${sourceSeq}.000Z`,
|
||||
...event
|
||||
},
|
||||
valuesPrinted: false
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
function withoutCommandId(record: any) {
|
||||
delete record.value.commandId;
|
||||
delete record.value.event.commandId;
|
||||
return record;
|
||||
}
|
||||
|
||||
function withCommandId(record: any, commandId: string) {
|
||||
record.value.commandId = commandId;
|
||||
record.value.event.commandId = commandId;
|
||||
return record;
|
||||
}
|
||||
|
||||
function createSha256(value: string) {
|
||||
return new Bun.CryptoHasher("sha256").update(value).digest("hex");
|
||||
}
|
||||
|
||||
@@ -11,12 +11,18 @@ import {
|
||||
DEFAULT_AGENTRUN_DEBUG_EVENT_TOPIC,
|
||||
DEFAULT_AGENTRUN_EVENT_TOPIC,
|
||||
DEFAULT_HWLAB_DEBUG_EVENT_TOPIC,
|
||||
DEFAULT_HWLAB_EVENT_TOPIC,
|
||||
projectAgentRunKafkaMessageToHwlabDebugEvent,
|
||||
queryKafkaEventStream
|
||||
} from "../../../internal/cloud/kafka-event-bridge.ts";
|
||||
import { decodeWorkbenchRealtimeEventFrame } from "../../../web/hwlab-cloud-web/src/api/workbench-realtime-codec.ts";
|
||||
import { reduceWorkbenchRealtimeEvent } from "../../../web/hwlab-cloud-web/src/stores/workbench-event-reducer.ts";
|
||||
import { projectWorkbenchLiveKafkaMessage } from "../../../web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.ts";
|
||||
import { planWorkbenchRealtimeApply } from "../../../web/hwlab-cloud-web/src/stores/workbench-realtime-plan.ts";
|
||||
import { renderTraceRowsMarkdown, traceDisplayRows } from "./trace-renderer.ts";
|
||||
|
||||
const CLI_NAME = "hwlab-cli";
|
||||
const VERSION = "0.3.1-kafka-debug-replay";
|
||||
const VERSION = "0.3.3-kafka-command-lifecycle-scope";
|
||||
const DEFAULT_LIMIT = 500;
|
||||
const DEFAULT_TIMEOUT_MS = 5000;
|
||||
const DEBUG_OUTPUT_PARTITION = 0;
|
||||
@@ -36,12 +42,14 @@ type KafkaCliOptions = {
|
||||
export async function mainKafkaCli(argv = process.argv.slice(3), options: KafkaCliOptions = {}) {
|
||||
const result = await runKafkaCli(argv, options);
|
||||
const json = argv.includes("--json");
|
||||
process.stdout.write(json ? `${JSON.stringify(result.payload, null, 2)}\n` : `${renderKafkaCliText(result.payload)}\n`);
|
||||
const markdown = requestedFormat(argv) === "markdown" && !json ? result.markdownOutput : null;
|
||||
process.stdout.write(json ? `${JSON.stringify(result.payload, null, 2)}\n` : markdown ? `${markdown.trimEnd()}\n` : `${renderKafkaCliText(result.payload)}\n`);
|
||||
process.exitCode = result.exitCode;
|
||||
}
|
||||
|
||||
export async function runKafkaCli(argv: string[], options: KafkaCliOptions = {}) {
|
||||
const now = options.now ?? (() => new Date().toISOString());
|
||||
let action = "kafka.help";
|
||||
try {
|
||||
const parsed = parseOptions(argv);
|
||||
const command = parsed._[0] || "help";
|
||||
@@ -49,9 +57,19 @@ export async function runKafkaCli(argv: string[], options: KafkaCliOptions = {})
|
||||
if (["help", "--help", "-h"].includes(command) || parsed.help === true) {
|
||||
return result(0, kafkaHelp(), now);
|
||||
}
|
||||
if (command !== "regenerate" || resource !== "hwlab") {
|
||||
throw cliError("unsupported_kafka_command", "supported command: kafka regenerate hwlab", { command, resource });
|
||||
if (command === "render" && resource === "trace") {
|
||||
action = "kafka.render.trace";
|
||||
const rendered = await renderHwlabKafkaTrace(parsed, {
|
||||
cwd: options.cwd ?? process.cwd(),
|
||||
env: options.env ?? process.env,
|
||||
now,
|
||||
readKafka: options.readKafka ?? defaultKafkaReader,
|
||||
createProducer: options.createProducer ?? defaultProducerFactory
|
||||
});
|
||||
return { ...result(0, rendered.payload, now), markdownOutput: rendered.markdown };
|
||||
}
|
||||
if (command !== "regenerate" || resource !== "hwlab") throw cliError("unsupported_kafka_command", "supported commands: kafka regenerate hwlab | kafka render trace", { command, resource });
|
||||
action = "kafka.regenerate.hwlab";
|
||||
const payload = await regenerateHwlabDebugEvents(parsed, {
|
||||
cwd: options.cwd ?? process.cwd(),
|
||||
env: options.env ?? process.env,
|
||||
@@ -61,7 +79,7 @@ export async function runKafkaCli(argv: string[], options: KafkaCliOptions = {})
|
||||
});
|
||||
return result(0, payload, now);
|
||||
} catch (error) {
|
||||
return result(1, failure(error), now);
|
||||
return result(1, failure(error, action), now);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -256,6 +274,321 @@ export async function regenerateHwlabDebugEvents(parsed: ParsedArgs, dependencie
|
||||
};
|
||||
}
|
||||
|
||||
export async function renderHwlabKafkaTrace(parsed: ParsedArgs, dependencies: Required<KafkaCliOptions>) {
|
||||
const traceId = requiredTraceId(parsed.traceId);
|
||||
const sessionId = optionalSessionId(parsed.sessionId);
|
||||
const runId = optionalRunId(parsed.runId);
|
||||
const commandId = optionalCommandId(parsed.commandId);
|
||||
const sourceMode = enumValue(parsed.from ?? "kafka", "from", ["kafka", "jsonl"]);
|
||||
const format = enumValue(parsed.format ?? "text", "format", ["text", "markdown"]);
|
||||
const inputTopicResolution = resolveConfig(
|
||||
parsed.inputTopic,
|
||||
dependencies.env.HWLAB_KAFKA_EVENT_TOPIC,
|
||||
DEFAULT_HWLAB_EVENT_TOPIC,
|
||||
"--input-topic",
|
||||
"env:HWLAB_KAFKA_EVENT_TOPIC"
|
||||
);
|
||||
const inputTopic = inputTopicResolution.value;
|
||||
const limit = boundedInteger(parsed.limit, DEFAULT_LIMIT, 1, 5000, "limit");
|
||||
const timeoutMs = boundedInteger(parsed.timeoutMs, DEFAULT_TIMEOUT_MS, 250, 60000, "timeoutMs");
|
||||
const expectedCount = optionalPositiveInteger(parsed.expectCount, "expectCount");
|
||||
const readLimit = commandId || expectedCount === undefined ? limit : Math.min(5000, expectedCount + 1);
|
||||
const groupPrefixResolution = resolveKafkaGroupConfig(
|
||||
parsed.groupPrefix,
|
||||
dependencies.env.HWLAB_KAFKA_HWLAB_DEBUG_GROUP_PREFIX,
|
||||
sourceMode === "kafka"
|
||||
);
|
||||
const groupPrefix = text(groupPrefixResolution.value);
|
||||
if (groupPrefix) assertDebugGroup(groupPrefix);
|
||||
|
||||
const readResult = sourceMode === "jsonl"
|
||||
? await readJsonlInput(requiredText(parsed.jsonlFile, "jsonlFile"), dependencies.cwd, inputTopic)
|
||||
: await dependencies.readKafka({
|
||||
env: dependencies.env,
|
||||
stream: "hwlab",
|
||||
topic: inputTopic,
|
||||
traceId,
|
||||
sessionId: sessionId || null,
|
||||
runId: runId || null,
|
||||
limit: readLimit,
|
||||
timeoutMs,
|
||||
fromBeginning: parsed.fromEnd !== true,
|
||||
groupIdPrefix: groupPrefix
|
||||
});
|
||||
const records = Array.isArray(readResult.events) ? readResult.events : [];
|
||||
const scoped = scopeHwlabTraceRecords(records, { traceId, sessionId, runId, commandId });
|
||||
const commandScopeScanComplete = sourceMode === "jsonl" || readResult.completionReason === "end-offset";
|
||||
const matchingRecords = scoped.records;
|
||||
if (matchingRecords.length === 0) {
|
||||
throw cliError(commandId && scoped.summary.baseMatchedCount > 0 ? "source_command_missing" : "source_trace_missing", commandId
|
||||
? "no HWLAB Kafka events matched the requested command lifecycle"
|
||||
: "no HWLAB Kafka events matched the requested traceId/sessionId", {
|
||||
traceId,
|
||||
sessionId: sessionId || null,
|
||||
runId: runId || null,
|
||||
commandId: commandId || null,
|
||||
scannedCount: readResult.scannedCount ?? records.length,
|
||||
parsedCount: readResult.parsedCount ?? records.length,
|
||||
queryMatchedCount: readResult.matchedCount ?? records.length,
|
||||
commandScope: { ...scoped.summary, scanComplete: commandScopeScanComplete },
|
||||
completionReason: readResult.completionReason ?? null,
|
||||
reachedEndOffsets: readResult.reachedEndOffsets ?? null
|
||||
});
|
||||
}
|
||||
if (expectedCount !== undefined && matchingRecords.length !== expectedCount) {
|
||||
throw cliError("hwlab_trace_event_count_mismatch", "matched HWLAB event count does not equal --expect-count", {
|
||||
expectedCount,
|
||||
actualCount: matchingRecords.length,
|
||||
traceId,
|
||||
sessionId: sessionId || null
|
||||
});
|
||||
}
|
||||
|
||||
const projection = projectHwlabRecordsThroughWorkbench(matchingRecords, {
|
||||
traceId,
|
||||
sessionId,
|
||||
observedAt: dependencies.now()
|
||||
});
|
||||
if (!projection.message?.runnerTrace) {
|
||||
throw cliError("workbench_trace_projection_empty", "HWLAB events did not produce a Workbench runner trace", {
|
||||
traceId,
|
||||
decodedCount: projection.decodedCount,
|
||||
plannedCount: projection.plannedCount,
|
||||
appliedCount: projection.appliedCount,
|
||||
rejectedCount: projection.rejections.length,
|
||||
firstRejection: projection.rejections[0] ?? null
|
||||
});
|
||||
}
|
||||
const finalResponseText = projection.message.traceAutoLifecycle === "terminal" ? text(projection.message.text) : "";
|
||||
const trace = projection.message.runnerTrace as Record<string, any>;
|
||||
const events = Array.isArray(trace.events) ? trace.events : [];
|
||||
const rows = traceDisplayRows(trace, events, {
|
||||
finalResponsePlacement: "outer",
|
||||
outerFinalResponseText: finalResponseText || null
|
||||
});
|
||||
const markdown = renderWorkbenchKafkaTraceMarkdown({
|
||||
traceId,
|
||||
sessionId: text(projection.message.sessionId),
|
||||
status: text(projection.message.status),
|
||||
sourceEventCount: events.length,
|
||||
rows,
|
||||
finalResponseText: finalResponseText || null
|
||||
});
|
||||
const stateId = `hwlab-trace-${Date.now().toString(36)}-${randomUUID().slice(0, 8)}`;
|
||||
const artifactPath = path.resolve(
|
||||
dependencies.cwd,
|
||||
text(parsed.outputMarkdown) || path.join(".state", "hwlab-cli", "kafka-render", stateId, `${traceId}.md`)
|
||||
);
|
||||
await mkdir(path.dirname(artifactPath), { recursive: true });
|
||||
await writeFile(artifactPath, `${markdown.trimEnd()}\n`, "utf8");
|
||||
const fullRows = parsed.full === true;
|
||||
const rowPreviewLimit = boundedInteger(parsed.rowLimit, 20, 1, 200, "rowLimit");
|
||||
const selectedRows = fullRows ? rows : rows.slice(-rowPreviewLimit);
|
||||
const sessionIds = uniqueText(matchingRecords.flatMap((record) => hwlabSessionCandidates(valueFromRecord(record))));
|
||||
const runIds = uniqueText(matchingRecords.flatMap((record) => runCandidates(valueFromRecord(record))));
|
||||
const commandIds = uniqueText(matchingRecords.flatMap((record) => commandCandidates(valueFromRecord(record))));
|
||||
const sourceEventIdentityValues = matchingRecords.map((record) => text(valueFromRecord(record)?.sourceEventId ?? valueFromRecord(record)?.event?.sourceEventId));
|
||||
const sourceEventIds = uniqueText(sourceEventIdentityValues);
|
||||
const sourceEventIdentityPresentCount = sourceEventIdentityValues.filter(Boolean).length;
|
||||
const rowPayload = selectedRows.map((row) => ({
|
||||
rowId: row.rowId,
|
||||
seq: row.seq,
|
||||
sequenceAuthority: row.sequenceAuthority,
|
||||
tone: row.tone,
|
||||
header: row.header,
|
||||
preview: row.preview ?? null,
|
||||
body: fullRows ? row.body : boundedText(row.body, 500),
|
||||
bodyTruncated: !fullRows && text(row.body).length > 500,
|
||||
terminal: row.terminal === true,
|
||||
bodyFormat: row.bodyFormat ?? null
|
||||
}));
|
||||
const toolRowCount = rows.filter((row) => row.rowId.startsWith("tool:")).length;
|
||||
const assistantRowCount = rows.filter((row) => /助手/u.test(row.header)).length;
|
||||
const finalResponseFingerprint = traceTextFingerprint(finalResponseText);
|
||||
const finalResponseInTrace = Boolean(finalResponseFingerprint && rows.some((row) =>
|
||||
/助手/u.test(row.header) && traceTextFingerprint(row.body) === finalResponseFingerprint
|
||||
));
|
||||
return {
|
||||
markdown,
|
||||
payload: {
|
||||
ok: true,
|
||||
action: "kafka.render.trace",
|
||||
status: "succeeded",
|
||||
format,
|
||||
input: {
|
||||
mode: sourceMode,
|
||||
topic: inputTopic,
|
||||
topicSource: inputTopicResolution.source,
|
||||
groupId: readResult.groupId ?? null,
|
||||
groupPrefix: groupPrefix || null,
|
||||
groupPrefixSource: groupPrefixResolution.source,
|
||||
scannedCount: readResult.scannedCount ?? records.length,
|
||||
parsedCount: readResult.parsedCount ?? records.length,
|
||||
queryMatchedCount: readResult.matchedCount ?? records.length,
|
||||
matchedCount: matchingRecords.length,
|
||||
invalidJsonCount: (readResult.invalidJsonCount ?? 0) + scoped.summary.invalidTransportRecordCount,
|
||||
filterRejectedCount: readResult.filterRejectedCount ?? 0,
|
||||
scopeRejectedCount: scoped.summary.baseRejectedCount + scoped.summary.commandScopeRejectedCount,
|
||||
commandScope: { ...scoped.summary, scanComplete: commandScopeScanComplete },
|
||||
completionReason: readResult.completionReason ?? (sourceMode === "jsonl" ? "file-end" : null),
|
||||
reachedEndOffsets: readResult.reachedEndOffsets ?? (sourceMode === "jsonl"),
|
||||
endOffsetsAvailable: readResult.endOffsetsAvailable ?? false,
|
||||
endOffsets: readResult.endOffsets ?? [],
|
||||
lastScannedOffsets: readResult.lastScannedOffsets ?? [],
|
||||
sourceFile: readResult.sourceFile ?? null,
|
||||
valuesPrinted: false
|
||||
},
|
||||
identity: {
|
||||
requestedTraceId: traceId,
|
||||
requestedSessionId: sessionId || null,
|
||||
requestedRunId: runId || null,
|
||||
requestedCommandId: commandId || null,
|
||||
observedTraceId: text(projection.message.traceId),
|
||||
observedSessionIds: sessionIds,
|
||||
observedRunIds: runIds,
|
||||
observedCommandIds: commandIds,
|
||||
baseObservedRunIds: scoped.summary.baseObservedRunIds,
|
||||
baseCommandIdCounts: scoped.summary.commandIdCounts,
|
||||
missingCommandIdCount: scoped.summary.missingCommandIdCount,
|
||||
selectedSessionId: text(projection.message.sessionId),
|
||||
sourceEventIdentityCount: sourceEventIds.length,
|
||||
sourceEventIdentityMissingCount: matchingRecords.length - sourceEventIdentityPresentCount,
|
||||
sourceEventIdentityDuplicates: Math.max(0, sourceEventIdentityPresentCount - sourceEventIds.length),
|
||||
valuesPrinted: false
|
||||
},
|
||||
projection: {
|
||||
pipeline: ["decodeWorkbenchRealtimeEventFrame", "reduceWorkbenchRealtimeEvent", "planWorkbenchRealtimeApply", "projectWorkbenchLiveKafkaMessage", "traceDisplayRows"],
|
||||
decodedCount: projection.decodedCount,
|
||||
plannedCount: projection.plannedCount,
|
||||
appliedCount: projection.appliedCount,
|
||||
rejectedCount: projection.rejections.length,
|
||||
rejections: projection.rejections.slice(0, 20),
|
||||
eventCount: events.length,
|
||||
terminalObserved: projection.message.traceAutoLifecycle === "terminal",
|
||||
status: projection.message.status,
|
||||
lastEventAt: projection.message.timing?.lastEventAt ?? null,
|
||||
valuesPrinted: false
|
||||
},
|
||||
render: {
|
||||
rowModel: "tools/src/hwlab-cli/trace-renderer:traceDisplayRows",
|
||||
markdownRenderer: "tools/src/hwlab-cli/trace-renderer:renderTraceRowsMarkdown",
|
||||
finalResponsePlacement: "outer",
|
||||
rowCount: rows.length,
|
||||
rowsReturned: rowPayload.length,
|
||||
rowsOmitted: Math.max(0, rows.length - rowPayload.length),
|
||||
rowWindow: fullRows ? "full" : "tail",
|
||||
rowPreviewLimit: fullRows ? null : rowPreviewLimit,
|
||||
toolRowCount,
|
||||
assistantRowCount,
|
||||
finalResponsePresent: Boolean(finalResponseText),
|
||||
finalResponseInTrace,
|
||||
rowModelSha256: sha256(JSON.stringify(rows)),
|
||||
rows: rowPayload,
|
||||
valuesPrinted: false
|
||||
},
|
||||
output: {
|
||||
artifact: { path: artifactPath, byteCount: Buffer.byteLength(`${markdown.trimEnd()}\n`, "utf8"), sha256: sha256(`${markdown.trimEnd()}\n`) },
|
||||
valuesPrinted: false
|
||||
},
|
||||
validation: {
|
||||
expectedCount: expectedCount ?? null,
|
||||
allMatchedEventsApplied: projection.appliedCount === matchingRecords.length,
|
||||
commandLifecyclePreserved: !commandId || (commandScopeScanComplete && scoped.summary.exactCommandCount > 0 && scoped.summary.ambiguousUnscopedLifecycleCount === 0),
|
||||
finalResponseOutsideTrace: !finalResponseInTrace && Boolean(finalResponseText),
|
||||
sharedRowModel: true,
|
||||
valuesPrinted: false
|
||||
},
|
||||
runtimeDependencies: {
|
||||
cloudApi: false,
|
||||
database: false,
|
||||
transactionalProjector: false,
|
||||
kafka: sourceMode === "kafka"
|
||||
},
|
||||
next: {
|
||||
command: `hwlab-cli kafka render trace --from ${sourceMode} --trace-id ${traceId}${sessionId ? ` --session-id ${sessionId}` : ""}${runId ? ` --run-id ${runId}` : ""}${commandId ? ` --command-id ${commandId}` : ""}${sourceMode === "jsonl" ? ` --jsonl-file ${shellArg(readResult.sourceFile)}` : ""} --format markdown`,
|
||||
reason: "使用同一持久事件和共享 row model 直接检查 Markdown Trace；浏览器验收只需核对最后的 HTML 分叉。"
|
||||
},
|
||||
valuesPrinted: false
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
export function projectHwlabRecordsThroughWorkbench(records: JsonRecord[], options: { traceId: string; sessionId?: string; observedAt: string }) {
|
||||
let message: any = null;
|
||||
let decodedCount = 0;
|
||||
let plannedCount = 0;
|
||||
let appliedCount = 0;
|
||||
const rejections: Array<{ index: number; code: string; detail?: string }> = [];
|
||||
const observedBaseMs = Date.parse(options.observedAt);
|
||||
const baseMs = Number.isFinite(observedBaseMs) ? observedBaseMs : 0;
|
||||
for (const [index, record] of records.entries()) {
|
||||
const value = valueFromRecord(record);
|
||||
const decoded = decodeWorkbenchRealtimeEventFrame(JSON.stringify(value));
|
||||
if (!decoded.payload) {
|
||||
rejections.push({ index, code: `decode-${decoded.status}` });
|
||||
continue;
|
||||
}
|
||||
decodedCount += 1;
|
||||
const schema = text(decoded.payload.schema);
|
||||
const realtimeEvent = schema === "hwlab.event.debug.v1"
|
||||
? { ...decoded.payload, schema: "hwlab.event.v1" }
|
||||
: decoded.payload;
|
||||
if (realtimeEvent.schema !== "hwlab.event.v1") {
|
||||
rejections.push({ index, code: "schema-mismatch", detail: text(realtimeEvent.schema) });
|
||||
continue;
|
||||
}
|
||||
const reduced = reduceWorkbenchRealtimeEvent(realtimeEvent as any, "hwlab.event.v1");
|
||||
const plan = planWorkbenchRealtimeApply(reduced.action);
|
||||
const traceStep = plan.steps.find((step) => step.type === "apply-trace-event");
|
||||
if (!traceStep || traceStep.type !== "apply-trace-event" || !traceStep.event) {
|
||||
rejections.push({ index, code: reduced.action.type === "ignore" ? reduced.action.reason : "trace-step-missing" });
|
||||
continue;
|
||||
}
|
||||
plannedCount += 1;
|
||||
const eventTraceId = text(realtimeEvent.traceId ?? traceStep.event.traceId);
|
||||
if (eventTraceId !== options.traceId) {
|
||||
rejections.push({ index, code: "trace-mismatch", detail: eventTraceId });
|
||||
continue;
|
||||
}
|
||||
const eventSessionId = text(realtimeEvent.hwlabSessionId ?? realtimeEvent.sessionId ?? traceStep.event.sessionId ?? options.sessionId);
|
||||
if (!eventSessionId) {
|
||||
rejections.push({ index, code: "session-missing" });
|
||||
continue;
|
||||
}
|
||||
if (options.sessionId && !hwlabSessionCandidates(realtimeEvent).includes(options.sessionId)) {
|
||||
rejections.push({ index, code: "session-mismatch", detail: eventSessionId });
|
||||
continue;
|
||||
}
|
||||
message = projectWorkbenchLiveKafkaMessage({
|
||||
previous: message,
|
||||
traceId: options.traceId,
|
||||
sessionId: eventSessionId,
|
||||
event: traceStep.event as any,
|
||||
receivedAt: new Date(baseMs + index).toISOString()
|
||||
});
|
||||
appliedCount += 1;
|
||||
}
|
||||
return { message, decodedCount, plannedCount, appliedCount, rejections };
|
||||
}
|
||||
|
||||
function renderWorkbenchKafkaTraceMarkdown(input: { traceId: string; sessionId: string; status: string; sourceEventCount: number; rows: any[]; finalResponseText: string | null }) {
|
||||
const lines = [
|
||||
`# Workbench Trace ${input.traceId}`,
|
||||
"",
|
||||
`- session: ${input.sessionId || "unknown"}`,
|
||||
`- status: ${input.status || "unknown"}`,
|
||||
`- source events: ${input.sourceEventCount}`,
|
||||
`- rendered rows: ${input.rows.length}`,
|
||||
`- final response placement: outer`,
|
||||
"",
|
||||
"## 运行记录",
|
||||
"",
|
||||
renderTraceRowsMarkdown(input.rows)
|
||||
];
|
||||
if (input.finalResponseText) lines.push("", "## 最终回复", "", input.finalResponseText);
|
||||
return lines.join("\n");
|
||||
}
|
||||
|
||||
export function mapAgentRunRecordsToHwlabDebugEvents(records: JsonRecord[], options: { sessionId: string; traceId?: string; expectedCount?: number; producedAt?: string; replayId?: string }) {
|
||||
const matchingRecords = records.filter((record) => recordMatches(record, options.sessionId, options.traceId));
|
||||
if (matchingRecords.length === 0) {
|
||||
@@ -318,12 +651,15 @@ function kafkaHelp() {
|
||||
action: "kafka.help",
|
||||
status: "succeeded",
|
||||
commands: [
|
||||
"hwlab-cli kafka render trace --from kafka --trace-id trc_... [--session-id ses_...] [--run-id run_...] [--command-id cmd_...] [--input-topic hwlab.event.v1] [--group-prefix hwlab-...-debug] [--format markdown] [--row-limit 20] [--json]",
|
||||
"hwlab-cli kafka render trace --from jsonl --trace-id trc_... [--run-id run_...] [--command-id cmd_...] --jsonl-file hwlab-events.jsonl [--format markdown] [--output-markdown trace.md] [--row-limit 20] [--json]",
|
||||
"hwlab-cli kafka regenerate hwlab --from kafka --session-id ses_... [--trace-id trc_...] [--replay-id rpl_...] [--input-topic agentrun.event.v1] [--group-prefix hwlab-...-debug] [--expect-count 35] [--json]",
|
||||
"hwlab-cli kafka regenerate hwlab --from kafka --session-id ses_... --input-topic agentrun.event.debug.v1 --group-prefix hwlab-...-debug [--replay-id rpl_...] [--json]",
|
||||
"hwlab-cli kafka regenerate hwlab --from jsonl --session-id ses_... [--replay-id rpl_...] --jsonl-file agentrun-events.jsonl --no-publish [--output-jsonl hwlab-events.jsonl] [--json]"
|
||||
],
|
||||
defaults: {
|
||||
kafkaInputTopic: DEFAULT_AGENTRUN_EVENT_TOPIC,
|
||||
kafkaTraceInputTopic: DEFAULT_HWLAB_EVENT_TOPIC,
|
||||
jsonlInputTopic: DEFAULT_AGENTRUN_DEBUG_EVENT_TOPIC,
|
||||
outputTopic: DEFAULT_HWLAB_DEBUG_EVENT_TOPIC,
|
||||
groupPrefix: null,
|
||||
@@ -348,10 +684,27 @@ function kafkaHelp() {
|
||||
}
|
||||
|
||||
async function defaultKafkaReader(input: Record<string, any>) {
|
||||
return queryKafkaEventStream({
|
||||
return withBunKafkaTimerClamp(() => queryKafkaEventStream({
|
||||
...input,
|
||||
kafkaFactory(config: { brokers: string[]; clientId: string }) { return boundedKafka(config); }
|
||||
});
|
||||
}));
|
||||
}
|
||||
|
||||
async function withBunKafkaTimerClamp<T>(task: () => Promise<T>) {
|
||||
if (!process.versions.bun) return task();
|
||||
const originalSetTimeout = globalThis.setTimeout;
|
||||
// KafkaJS 2.2.4 schedules an idle queue check at 0 - Date.now(); Bun clamps it but emits an unbounded warning stack.
|
||||
const clampedSetTimeout = ((handler: Parameters<typeof globalThis.setTimeout>[0], timeout?: number, ...args: any[]) => originalSetTimeout(
|
||||
handler,
|
||||
typeof timeout === "number" && timeout < 0 ? 1 : timeout,
|
||||
...args
|
||||
)) as typeof globalThis.setTimeout;
|
||||
globalThis.setTimeout = clampedSetTimeout;
|
||||
try {
|
||||
return await task();
|
||||
} finally {
|
||||
if (globalThis.setTimeout === clampedSetTimeout) globalThis.setTimeout = originalSetTimeout;
|
||||
}
|
||||
}
|
||||
|
||||
async function defaultProducerFactory({ env, clientId }: { env: EnvLike; clientId: string }) {
|
||||
@@ -383,7 +736,21 @@ async function readJsonlInput(fileValue: string, cwd: string, defaultTopic: stri
|
||||
}
|
||||
events.push(normalizeRecord(value, defaultTopic));
|
||||
}
|
||||
return { ok: true, topic: defaultTopic, groupId: null, events, sourceFile: file };
|
||||
return {
|
||||
ok: true,
|
||||
topic: defaultTopic,
|
||||
groupId: null,
|
||||
count: events.length,
|
||||
scannedCount: events.length,
|
||||
parsedCount: events.length,
|
||||
matchedCount: events.length,
|
||||
invalidJsonCount: 0,
|
||||
filterRejectedCount: 0,
|
||||
completionReason: "file-end",
|
||||
reachedEndOffsets: true,
|
||||
events,
|
||||
sourceFile: file
|
||||
};
|
||||
}
|
||||
|
||||
function normalizeRecord(record: any, defaultTopic: string) {
|
||||
@@ -432,12 +799,145 @@ function sessionCandidates(value: any) {
|
||||
value.run?.sessionId,
|
||||
value.run?.hwlabSessionId,
|
||||
value.event?.payload?.sessionId,
|
||||
value.event?.payload?.hwlabSessionId
|
||||
value.event?.payload?.hwlabSessionId,
|
||||
value.event?.sessionId,
|
||||
value.event?.hwlabSessionId
|
||||
]);
|
||||
}
|
||||
|
||||
function traceCandidates(value: any) {
|
||||
return uniqueText([value.traceId, value.run?.traceId, value.command?.traceId, value.event?.payload?.traceId, value.event?.payload?.hwlabTraceId]);
|
||||
return uniqueText([value.traceId, value.run?.traceId, value.command?.traceId, value.event?.traceId, value.event?.payload?.traceId, value.event?.payload?.hwlabTraceId]);
|
||||
}
|
||||
|
||||
function scopeHwlabTraceRecords(records: JsonRecord[], filters: { traceId: string; sessionId?: string; runId?: string; commandId?: string }) {
|
||||
const parsed: Array<{ record: JsonRecord; value: Record<string, any> }> = [];
|
||||
let invalidTransportRecordCount = 0;
|
||||
for (const record of records) {
|
||||
try {
|
||||
parsed.push({ record, value: valueFromRecord(record) });
|
||||
} catch {
|
||||
invalidTransportRecordCount += 1;
|
||||
}
|
||||
}
|
||||
const base = parsed.filter(({ value }) => {
|
||||
if (!traceCandidates(value).includes(filters.traceId)) return false;
|
||||
if (filters.sessionId && !hwlabSessionCandidates(value).includes(filters.sessionId)) return false;
|
||||
return !filters.runId || runCandidates(value).includes(filters.runId);
|
||||
});
|
||||
const baseObservedRunIds = uniqueText(base.flatMap(({ value }) => runCandidates(value)));
|
||||
const commandIdCountMap = new Map<string, number>();
|
||||
let missingCommandIdCount = 0;
|
||||
let unscopedLifecycleCount = 0;
|
||||
let unscopedNonLifecycleCount = 0;
|
||||
for (const { value } of base) {
|
||||
const commands = commandCandidates(value);
|
||||
if (commands.length === 0) {
|
||||
missingCommandIdCount += 1;
|
||||
if (hwlabCommandLifecycleEvent(value)) unscopedLifecycleCount += 1;
|
||||
else unscopedNonLifecycleCount += 1;
|
||||
continue;
|
||||
}
|
||||
for (const candidate of commands) commandIdCountMap.set(candidate, (commandIdCountMap.get(candidate) ?? 0) + 1);
|
||||
}
|
||||
const exactCommandRecords = filters.commandId
|
||||
? base.filter(({ value }) => commandCandidates(value).includes(filters.commandId as string))
|
||||
: [];
|
||||
const soleBaseRunId = baseObservedRunIds.length === 1 ? baseObservedRunIds[0] : null;
|
||||
const explicitTargetRunIds = uniqueText([
|
||||
filters.runId,
|
||||
...exactCommandRecords.flatMap(({ value }) => runCandidates(value))
|
||||
]);
|
||||
const targetRunIds = explicitTargetRunIds.length > 0 ? explicitTargetRunIds : uniqueText([soleBaseRunId]);
|
||||
let selectedUnscopedLifecycleCount = 0;
|
||||
let attributableUnscopedLifecycleCount = 0;
|
||||
let ambiguousUnscopedLifecycleCount = 0;
|
||||
let otherRunUnscopedLifecycleCount = 0;
|
||||
let otherCommandRecordCount = 0;
|
||||
const selected = !filters.commandId
|
||||
? base
|
||||
: exactCommandRecords.length === 0
|
||||
? []
|
||||
: base.filter(({ value }) => {
|
||||
const commands = commandCandidates(value);
|
||||
if (commands.includes(filters.commandId as string)) return true;
|
||||
if (commands.length > 0) {
|
||||
otherCommandRecordCount += 1;
|
||||
return false;
|
||||
}
|
||||
if (!hwlabCommandLifecycleEvent(value)) return false;
|
||||
const runs = runCandidates(value);
|
||||
const attributable = runs.some((candidate) => targetRunIds.includes(candidate))
|
||||
|| (runs.length === 0 && Boolean(soleBaseRunId) && targetRunIds.includes(soleBaseRunId as string));
|
||||
if (!attributable) {
|
||||
if (runs.length > 0 && targetRunIds.length > 0) otherRunUnscopedLifecycleCount += 1;
|
||||
else ambiguousUnscopedLifecycleCount += 1;
|
||||
return false;
|
||||
}
|
||||
attributableUnscopedLifecycleCount += 1;
|
||||
selectedUnscopedLifecycleCount += 1;
|
||||
return true;
|
||||
});
|
||||
return {
|
||||
records: selected.map(({ record }) => record),
|
||||
summary: {
|
||||
mode: filters.commandId ? "command-lifecycle" : "trace-run",
|
||||
requestedCommandId: filters.commandId || null,
|
||||
baseMatchedCount: base.length,
|
||||
selectedCount: selected.length,
|
||||
exactCommandCount: exactCommandRecords.length,
|
||||
missingCommandIdCount,
|
||||
unscopedLifecycleCount,
|
||||
attributableUnscopedLifecycleCount,
|
||||
selectedUnscopedLifecycleCount,
|
||||
ambiguousUnscopedLifecycleCount,
|
||||
otherRunUnscopedLifecycleCount,
|
||||
unscopedNonLifecycleCount,
|
||||
otherCommandRecordCount,
|
||||
commandScopeRejectedCount: Math.max(0, base.length - selected.length),
|
||||
baseRejectedCount: Math.max(0, records.length - invalidTransportRecordCount - base.length),
|
||||
invalidTransportRecordCount,
|
||||
baseObservedRunIds,
|
||||
targetRunIds,
|
||||
commandIdCounts: [...commandIdCountMap.entries()]
|
||||
.sort(([left], [right]) => left.localeCompare(right))
|
||||
.map(([commandId, count]) => ({ commandId, count })),
|
||||
valuesPrinted: false
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
function hwlabCommandLifecycleEvent(value: Record<string, any>) {
|
||||
const event = recordObject(value.event) ?? value;
|
||||
const finalResponse = recordObject(event.finalResponse);
|
||||
const types = uniqueText([event.type, event.eventType, value.type, value.eventType]).map((candidate) => candidate.toLowerCase());
|
||||
return event.terminal === true
|
||||
|| event.final === true
|
||||
|| event.replyAuthority === true
|
||||
|| Boolean(text(finalResponse?.text))
|
||||
|| types.some((candidate) => ["terminal", "terminal_status", "result"].includes(candidate));
|
||||
}
|
||||
|
||||
function hwlabSessionCandidates(value: any) {
|
||||
return uniqueText([
|
||||
value?.hwlabSessionId,
|
||||
value?.sessionId,
|
||||
value?.event?.hwlabSessionId,
|
||||
value?.event?.sessionId,
|
||||
value?.context?.hwlabSessionId,
|
||||
value?.context?.sessionId
|
||||
]);
|
||||
}
|
||||
|
||||
function runCandidates(value: any) {
|
||||
return uniqueText([value?.runId, value?.run_id, value?.event?.runId, value?.event?.run_id, value?.context?.runId, value?.context?.run_id]);
|
||||
}
|
||||
|
||||
function commandCandidates(value: any) {
|
||||
return uniqueText([value?.commandId, value?.command_id, value?.event?.commandId, value?.event?.command_id, value?.context?.commandId, value?.context?.command_id]);
|
||||
}
|
||||
|
||||
function recordObject(value: unknown): Record<string, any> | null {
|
||||
return value && typeof value === "object" && !Array.isArray(value) ? value as Record<string, any> : null;
|
||||
}
|
||||
|
||||
function parseOptions(argv: string[]): ParsedArgs {
|
||||
@@ -462,7 +962,17 @@ function renderKafkaCliText(payload: Record<string, any>) {
|
||||
return [
|
||||
"ok",
|
||||
`action=${payload.action}`,
|
||||
payload.input ? `input=${payload.input.matchedCount}/${payload.input.readCount}` : null,
|
||||
payload.input?.scannedCount !== undefined ? `scanned=${payload.input.scannedCount}` : null,
|
||||
payload.input?.parsedCount !== undefined ? `parsed=${payload.input.parsedCount}` : null,
|
||||
payload.input?.queryMatchedCount !== undefined ? `queryMatched=${payload.input.queryMatchedCount}` : null,
|
||||
payload.input?.matchedCount !== undefined ? `matched=${payload.input.matchedCount}` : null,
|
||||
payload.input?.commandScope?.exactCommandCount !== undefined ? `commandExact=${payload.input.commandScope.exactCommandCount}` : null,
|
||||
payload.input?.commandScope?.selectedUnscopedLifecycleCount !== undefined ? `commandLifecycle=${payload.input.commandScope.selectedUnscopedLifecycleCount}` : null,
|
||||
payload.input?.commandScope?.scanComplete !== undefined ? `scopeComplete=${payload.input.commandScope.scanComplete}` : null,
|
||||
payload.input?.readCount !== undefined ? `input=${payload.input.matchedCount}/${payload.input.readCount}` : null,
|
||||
payload.projection?.appliedCount !== undefined ? `applied=${payload.projection.appliedCount}` : null,
|
||||
payload.render?.rowCount !== undefined ? `rows=${payload.render.rowCount}` : null,
|
||||
payload.projection?.terminalObserved !== undefined ? `terminal=${payload.projection.terminalObserved}` : null,
|
||||
payload.output ? `output=${payload.output.count}` : null,
|
||||
payload.output ? `published=${payload.output.publishedCount}` : null,
|
||||
validation.orderPreserved !== undefined ? `order=${validation.orderPreserved ? "preserved" : "failed"}` : null,
|
||||
@@ -478,13 +988,13 @@ function result(exitCode: number, payload: Record<string, any>, now: () => strin
|
||||
return { exitCode, payload: { generatedAt: now(), cli: CLI_NAME, version: VERSION, ...payload } };
|
||||
}
|
||||
|
||||
function failure(error: any) {
|
||||
function failure(error: any, action = "kafka.regenerate.hwlab") {
|
||||
return {
|
||||
ok: false,
|
||||
action: "kafka.regenerate.hwlab",
|
||||
action,
|
||||
status: "failed",
|
||||
error: { code: error?.code ?? "hwlab_kafka_debug_cli_failed", message: error instanceof Error ? error.message : String(error), details: error?.details ?? undefined },
|
||||
next: { command: "hwlab-cli kafka help --json", reason: "检查独立 debug topic、sessionId、JSONL 合同和 Kafka YAML 注入后重试。" },
|
||||
next: { command: "hwlab-cli kafka help --json", reason: "检查 Kafka topic/group 的 YAML 注入、trace/session identity、JSONL 合同和扫描可见性后重试。" },
|
||||
valuesPrinted: false
|
||||
};
|
||||
}
|
||||
@@ -505,6 +1015,30 @@ function requiredSessionId(value: unknown) {
|
||||
return result;
|
||||
}
|
||||
|
||||
function optionalSessionId(value: unknown) {
|
||||
const result = text(value);
|
||||
if (result && !/^ses_[A-Za-z0-9_.:-]+$/u.test(result)) throw cliError("invalid_session_id", "sessionId must start with ses_", { sessionId: result });
|
||||
return result;
|
||||
}
|
||||
|
||||
function optionalRunId(value: unknown) {
|
||||
const result = text(value);
|
||||
if (result && !/^run_[A-Za-z0-9_.:-]+$/u.test(result)) throw cliError("invalid_run_id", "runId must start with run_", { runId: result });
|
||||
return result;
|
||||
}
|
||||
|
||||
function optionalCommandId(value: unknown) {
|
||||
const result = text(value);
|
||||
if (result && !/^cmd_[A-Za-z0-9_.:-]+$/u.test(result)) throw cliError("invalid_command_id", "commandId must start with cmd_", { commandId: result });
|
||||
return result;
|
||||
}
|
||||
|
||||
function requiredTraceId(value: unknown) {
|
||||
const result = optionalTraceId(value);
|
||||
if (!result) throw cliError("trace_id_required", "--trace-id is required", { field: "traceId" });
|
||||
return result;
|
||||
}
|
||||
|
||||
function optionalTraceId(value: unknown) {
|
||||
const result = text(value);
|
||||
if (result && !/^trc_[A-Za-z0-9_.:-]+$/u.test(result)) throw cliError("invalid_trace_id", "traceId must start with trc_", { traceId: result });
|
||||
@@ -597,6 +1131,9 @@ function cliError(code: string, message: string, details: Record<string, unknown
|
||||
|
||||
function uniqueText(values: unknown[]) { return [...new Set(values.map(text).filter(Boolean))]; }
|
||||
function text(value: unknown) { return String(value ?? "").trim(); }
|
||||
function boundedText(value: unknown, limit: number) { const result = String(value ?? ""); return result.length <= limit ? result || null : `${result.slice(0, Math.max(0, limit - 3))}...`; }
|
||||
function traceTextFingerprint(value: unknown) { return text(value).replace(/\s+/gu, " "); }
|
||||
function csv(value: unknown) { return String(value ?? "").split(",").map((item) => item.trim()).filter(Boolean); }
|
||||
function sha256(value: string) { return createHash("sha256").update(value).digest("hex"); }
|
||||
function shellArg(value: unknown) { const result = String(value ?? ""); return /^[A-Za-z0-9_./:@-]+$/u.test(result) ? result : `'${result.replace(/'/gu, `'"'"'`)}'`; }
|
||||
function requestedFormat(argv: string[]) { const index = argv.findIndex((item) => item === "--format"); if (index >= 0) return text(argv[index + 1]); const inline = argv.find((item) => item.startsWith("--format=")); return inline ? text(inline.slice("--format=".length)) : "text"; }
|
||||
|
||||
@@ -13,27 +13,37 @@ export interface TraceEventRow {
|
||||
type TraceEvent = Record<string, unknown>;
|
||||
type TraceClockFormatter = (value: string) => string;
|
||||
export type TraceSequenceAuthority = "projected" | "source";
|
||||
export type TraceFinalResponsePlacement = "trace" | "outer";
|
||||
|
||||
export interface TraceDisplayRowsOptions {
|
||||
formatClock?: TraceClockFormatter;
|
||||
finalResponsePlacement?: TraceFinalResponsePlacement;
|
||||
outerFinalResponseText?: string | null;
|
||||
}
|
||||
|
||||
interface ResolvedTraceDisplayRowsOptions extends TraceDisplayRowsOptions {
|
||||
sequenceAuthority: TraceSequenceAuthority;
|
||||
finalResponsePlacement: TraceFinalResponsePlacement;
|
||||
}
|
||||
|
||||
export function traceDisplayRows(trace: Record<string, unknown> = {}, events: TraceEvent[] = [], options: TraceDisplayRowsOptions = {}): TraceEventRow[] {
|
||||
const displayOptions: ResolvedTraceDisplayRowsOptions = {
|
||||
...options,
|
||||
sequenceAuthority: traceSequenceAuthority(trace)
|
||||
sequenceAuthority: traceSequenceAuthority(trace),
|
||||
finalResponsePlacement: options.finalResponsePlacement ?? "trace"
|
||||
};
|
||||
const orderedEvents = traceEventsForDisplay(
|
||||
events.filter((event) => hasTraceSequence(event, displayOptions.sequenceAuthority)),
|
||||
displayOptions.sequenceAuthority
|
||||
);
|
||||
const effectiveTrace = traceWithInferredStart(trace, orderedEvents);
|
||||
const finalResponseText = traceFinalResponseText(effectiveTrace);
|
||||
const finalResponseText = nonEmptyString(options.outerFinalResponseText) ?? traceFinalResponseText(effectiveTrace);
|
||||
const finalResponseFingerprint = traceMessageFingerprint(finalResponseText);
|
||||
const terminalAssistantFingerprints = new Set(orderedEvents
|
||||
.filter(isTerminalAssistantTraceEvent)
|
||||
.map((event) => traceAssistantEventFingerprint(effectiveTrace, event))
|
||||
.filter((value): value is string => Boolean(value)));
|
||||
if (finalResponseFingerprint) terminalAssistantFingerprints.add(finalResponseFingerprint);
|
||||
const hasCompletionEvent = orderedEvents.some(isCompletionTraceEvent);
|
||||
const rows: TraceEventRow[] = [];
|
||||
let completionEvent: TraceEvent | null = null;
|
||||
@@ -64,6 +74,7 @@ export function traceDisplayRows(trace: Record<string, unknown> = {}, events: Tr
|
||||
continue;
|
||||
}
|
||||
if (isTerminalAssistantTraceEvent(event)) {
|
||||
if (isOuterFinalResponseTraceEvent(effectiveTrace, event, displayOptions, terminalAssistantFingerprints)) continue;
|
||||
if (hasCompletionEvent) continue;
|
||||
if (finalResponseText) {
|
||||
rows.push(traceFinalResponseRow(event, finalResponseText, displayOptions));
|
||||
@@ -73,7 +84,9 @@ export function traceDisplayRows(trace: Record<string, unknown> = {}, events: Tr
|
||||
continue;
|
||||
}
|
||||
if (isAssistantTraceEvent(event)) {
|
||||
if (finalResponseFingerprint && traceAssistantEventFingerprint(effectiveTrace, event) === finalResponseFingerprint) continue;
|
||||
const fingerprint = traceAssistantEventFingerprint(effectiveTrace, event);
|
||||
if (isOuterFinalResponseTraceEvent(effectiveTrace, event, displayOptions, terminalAssistantFingerprints)) continue;
|
||||
if (finalResponseFingerprint && fingerprint === finalResponseFingerprint) continue;
|
||||
rows.push(traceAssistantMessageRow(effectiveTrace, event, { terminal: false }, displayOptions));
|
||||
continue;
|
||||
}
|
||||
@@ -88,7 +101,23 @@ export function traceDisplayRows(trace: Record<string, unknown> = {}, events: Tr
|
||||
if (progressRow) rows.push(progressRow);
|
||||
if (rows.length > 0) return rows;
|
||||
if (orderedEvents.length === 0) return [];
|
||||
return orderedEvents.filter((event) => !isSuppressedTraceEvent(event)).map((event) => traceDisplayRow(effectiveTrace, event, displayOptions));
|
||||
return orderedEvents
|
||||
.filter((event) => !isSuppressedTraceEvent(event))
|
||||
.filter((event) => !isOuterFinalResponseTraceEvent(effectiveTrace, event, displayOptions, terminalAssistantFingerprints))
|
||||
.map((event) => traceDisplayRow(effectiveTrace, event, displayOptions));
|
||||
}
|
||||
|
||||
function isOuterFinalResponseTraceEvent(
|
||||
trace: Record<string, unknown>,
|
||||
event: TraceEvent,
|
||||
options: ResolvedTraceDisplayRowsOptions,
|
||||
terminalAssistantFingerprints: Set<string>
|
||||
): boolean {
|
||||
if (options.finalResponsePlacement !== "outer") return false;
|
||||
if (isTerminalAssistantTraceEvent(event)) return true;
|
||||
if (!isAssistantTraceEvent(event)) return false;
|
||||
const fingerprint = traceAssistantEventFingerprint(trace, event);
|
||||
return Boolean(fingerprint && terminalAssistantFingerprints.has(fingerprint));
|
||||
}
|
||||
|
||||
function traceSequenceAuthority(trace: Record<string, unknown>): TraceSequenceAuthority {
|
||||
@@ -177,42 +206,71 @@ function stripTraceToolOutputFromSummary(value: string): string {
|
||||
|
||||
function traceToolCallRow(trace: Record<string, unknown>, event: TraceEvent, options: ResolvedTraceDisplayRowsOptions): TraceEventRow {
|
||||
const command = cleanShellCommand(event.command);
|
||||
const outputPreview = traceEventText(event, ["outputSummary", "stdoutSummary", "stderrSummary", "output", "outputText"]);
|
||||
const status = traceStatusToken(event);
|
||||
const toolName = traceToolName(event, command);
|
||||
const body = traceToolCallBody(event, command);
|
||||
const eventKey = traceEventIdentityToken(event, options.sequenceAuthority);
|
||||
const duration = numberOrNull(event.durationMs);
|
||||
const exitCode = numberOrNull(event.exitCode);
|
||||
return {
|
||||
rowId: `tool:${event.itemId ?? eventKey ?? `${event.label ?? event.type ?? "tool"}:${event.createdAt ?? "unknown"}`}`,
|
||||
seq: traceEventDisplaySeq(event, options.sequenceAuthority),
|
||||
sequenceAuthority: options.sequenceAuthority,
|
||||
tone: traceEventTone(event),
|
||||
header: `${traceEventClock(event, options)} total=${formatTraceDuration(traceRelativeMs(trace, event))} ${status} ${toolName}`.trim(),
|
||||
header: [
|
||||
traceEventClock(event, options),
|
||||
`total=${formatTraceDuration(traceRelativeMs(trace, event))}`,
|
||||
status,
|
||||
toolName,
|
||||
duration !== null ? `duration=${formatTraceDuration(duration)}` : null,
|
||||
exitCode !== null ? `exit=${exitCode}` : null
|
||||
].filter(Boolean).join(" "),
|
||||
body,
|
||||
preview: command ? compactTraceOneLine(command, 2000) : null,
|
||||
preview: traceToolPreview(command, outputPreview),
|
||||
bodyFormat: "text"
|
||||
};
|
||||
}
|
||||
|
||||
function traceToolPreview(command: string | null, output: string | null): string | null {
|
||||
const parts = [
|
||||
command ? compactTraceOneLine(command, 92) : null,
|
||||
output ? `output: ${compactTraceOneLine(output, 92)}` : null
|
||||
].filter((value): value is string => Boolean(value));
|
||||
return parts.length > 0 ? parts.join(" | ") : null;
|
||||
}
|
||||
|
||||
function traceToolName(event: TraceEvent, command: string | null): string {
|
||||
const direct = nonEmptyString(event.toolName);
|
||||
if (direct && !isProtocolToolLifecycleName(direct)) return direct;
|
||||
if (direct && !isProtocolToolLifecycleName(direct) && !isGenericToolName(direct)) return direct;
|
||||
const label = readableTraceLabel(event).replace(/^agentrun:tool:/u, "").replace(/^tool:/u, "").replace(/^item\//u, "");
|
||||
if (label && !isProtocolToolLifecycleName(label)) return label;
|
||||
if (label && !isProtocolToolLifecycleName(label) && !isGenericToolName(label)) return label;
|
||||
const commandName = traceToolNameFromCommand(command);
|
||||
if (commandName) return commandName;
|
||||
if (direct && !isProtocolToolLifecycleName(direct)) return direct;
|
||||
const previewName = traceToolNameFromPreview(event);
|
||||
if (previewName) return previewName;
|
||||
return "tool call";
|
||||
}
|
||||
|
||||
function isGenericToolName(value: unknown): boolean {
|
||||
return /^(commandExecution|tool|shell|command|unknown)$/iu.test(String(value ?? "").trim());
|
||||
}
|
||||
|
||||
function isProtocolToolLifecycleName(value: unknown): boolean {
|
||||
return /^(?:item\/)?(?:started|completed)$/iu.test(String(value ?? "").trim()) || /^item\/(?:started|completed)$/iu.test(String(value ?? "").trim());
|
||||
}
|
||||
|
||||
function traceToolNameFromCommand(command: string | null): string | null {
|
||||
const text = cleanTraceText(command);
|
||||
const match = text.match(/^([A-Za-z][A-Za-z0-9_.:-]{1,80})(?:\s|$)/u);
|
||||
return match?.[1] ?? null;
|
||||
const segments = text.split(/\s*(?:&&|\|\||;)\s*/u).map((value) => value.trim()).filter(Boolean);
|
||||
const candidate = segments.find((value) => !/^(?:cd|export|set|source|umask)\b/u.test(value)) ?? segments[0] ?? text;
|
||||
const normalized = candidate
|
||||
.replace(/^(?:[A-Za-z_][A-Za-z0-9_]*=[^\s]+\s+)+/u, "")
|
||||
.replace(/^(?:sudo\s+|env\s+|command\s+)+/u, "")
|
||||
.replace(/^timeout\s+\S+\s+/u, "");
|
||||
const match = normalized.match(/^([A-Za-z./][A-Za-z0-9_./:-]{0,80})(?:\s|$)/u);
|
||||
return match?.[1]?.split("/").filter(Boolean).at(-1) ?? null;
|
||||
}
|
||||
|
||||
function traceToolNameFromPreview(event: TraceEvent): string | null {
|
||||
@@ -234,11 +292,19 @@ function jsonPreviewStringField(text: string | null, key: string): string | null
|
||||
}
|
||||
|
||||
function traceToolCallBody(event: TraceEvent, command: string | null): string | null {
|
||||
const stdout = traceEventText(event, ["stdout", "stdoutText", "stdoutSummary"]);
|
||||
const stderr = traceEventText(event, ["stderr", "stderrText", "stderrSummary"]);
|
||||
const output = traceEventText(event, ["output", "outputText", "outputSummary"]);
|
||||
const distinctOutput = output && output !== stdout && output !== stderr ? output : null;
|
||||
const outputBytes = numberOrNull(event.outputBytes);
|
||||
const lines = [
|
||||
command,
|
||||
traceDetailBlock("stdout", traceEventText(event, ["stdout", "stdoutText", "stdoutSummary"])),
|
||||
traceDetailBlock("stderr", traceEventText(event, ["stderr", "stderrText", "stderrSummary"])),
|
||||
command ? null : traceEventText(event, ["output", "outputText", "outputSummary"]),
|
||||
traceDetailBlock("stdout", stdout),
|
||||
traceDetailBlock("stderr", stderr),
|
||||
traceDetailBlock("output", distinctOutput),
|
||||
event.commandTruncated === true ? "commandTruncated=true" : null,
|
||||
outputBytes !== null ? `outputBytes=${outputBytes}` : null,
|
||||
event.outputTruncated === true ? "outputTruncated=true" : null,
|
||||
event.exitCode !== undefined ? `exitCode=${event.exitCode}` : null
|
||||
].map(cleanTraceDetailText).filter(Boolean);
|
||||
return lines.length > 0 ? lines.join("\n") : null;
|
||||
@@ -634,6 +700,7 @@ function compactTraceOneLine(value: unknown, limit = 220): string {
|
||||
}
|
||||
|
||||
function numberOrNull(value: unknown): number | null {
|
||||
if (value === null || value === undefined || value === "") return null;
|
||||
const number = Number(value);
|
||||
return Number.isInteger(number) ? number : null;
|
||||
}
|
||||
|
||||
@@ -7,6 +7,7 @@ import { createCoalescedEventQueue } from "@/utils/scheduler/coalesced-event-que
|
||||
import { composeWorkbenchScopedKey, firstScopePart } from "@/utils/workbench-key";
|
||||
import { recordWorkbenchRuntimeDiagnostic, recordWorkbenchSseLifecycle } from "@/utils/workbench-performance";
|
||||
import type { WorkbenchRealtimeCapabilities } from "@/config/runtime";
|
||||
import { decodeWorkbenchRealtimeEventFrame } from "./workbench-realtime-codec";
|
||||
|
||||
export interface WorkbenchRealtimeTraceSnapshot {
|
||||
traceId?: string | null;
|
||||
@@ -120,7 +121,7 @@ export interface WorkbenchEventStream {
|
||||
export interface WorkbenchSseIngressFrame {
|
||||
eventName: string;
|
||||
rawText: string;
|
||||
decodeStatus: "accepted" | "invalid-json" | "not-object";
|
||||
decodeStatus: import("./workbench-realtime-codec").WorkbenchRealtimeDecodeStatus;
|
||||
}
|
||||
|
||||
interface QueuedRealtimeEvent {
|
||||
@@ -251,20 +252,9 @@ function appendNumberParam(params: URLSearchParams, key: string, value: number |
|
||||
|
||||
function enqueueRealtimeFrame(raw: unknown, eventName: string, options: WorkbenchEventStreamOptions, queue: ReturnType<typeof createCoalescedEventQueue<QueuedRealtimeEvent>>): void {
|
||||
const rawText = typeof raw === "string" ? raw : String(raw ?? "");
|
||||
const decoded = decodeRealtimeEvent(rawText);
|
||||
const decoded = decodeWorkbenchRealtimeEventFrame(rawText);
|
||||
options.onIngress?.({ eventName, rawText, decodeStatus: decoded.status });
|
||||
if (decoded.payload) queue.push({ payload: decoded.payload, eventName });
|
||||
}
|
||||
|
||||
function decodeRealtimeEvent(raw: string): { payload: WorkbenchRealtimeEvent | null; status: WorkbenchSseIngressFrame["decodeStatus"] } {
|
||||
try {
|
||||
const value = JSON.parse(raw);
|
||||
return value && typeof value === "object" && !Array.isArray(value)
|
||||
? { payload: value as WorkbenchRealtimeEvent, status: "accepted" }
|
||||
: { payload: null, status: "not-object" };
|
||||
} catch {
|
||||
return { payload: null, status: "invalid-json" };
|
||||
}
|
||||
if (decoded.payload) queue.push({ payload: decoded.payload as WorkbenchRealtimeEvent, eventName });
|
||||
}
|
||||
|
||||
function scheduleRealtimeFlushYield(flush: () => void, yieldMs: number | null | undefined): () => void {
|
||||
|
||||
@@ -0,0 +1,17 @@
|
||||
export type WorkbenchRealtimeDecodeStatus = "accepted" | "invalid-json" | "not-object";
|
||||
|
||||
export interface WorkbenchRealtimeDecodedFrame {
|
||||
payload: Record<string, unknown> | null;
|
||||
status: WorkbenchRealtimeDecodeStatus;
|
||||
}
|
||||
|
||||
export function decodeWorkbenchRealtimeEventFrame(raw: string): WorkbenchRealtimeDecodedFrame {
|
||||
try {
|
||||
const value = JSON.parse(raw);
|
||||
return value && typeof value === "object" && !Array.isArray(value)
|
||||
? { payload: value as Record<string, unknown>, status: "accepted" }
|
||||
: { payload: null, status: "not-object" };
|
||||
} catch {
|
||||
return { payload: null, status: "invalid-json" };
|
||||
}
|
||||
}
|
||||
@@ -6,9 +6,16 @@ import MessageMarkdown from "@/components/workbench/MessageMarkdown.vue";
|
||||
import { useWorkbenchBottomFollowScroll } from "@/composables/useWorkbenchScrollRuntime";
|
||||
import { formatDisplayClock } from "@/config/runtime";
|
||||
import { readWorkbenchString, writeWorkbenchString } from "@/utils/workbench-storage-runtime";
|
||||
import { traceDisplayRows, type TraceEventRow } from "../../../../../tools/src/hwlab-cli/trace-renderer.ts";
|
||||
import { traceDisplayRows, type TraceEventRow, type TraceFinalResponsePlacement } from "../../../../../tools/src/hwlab-cli/trace-renderer.ts";
|
||||
|
||||
const props = defineProps<{ trace?: RunnerTrace | null; defaultExpanded?: boolean; storageKey?: string; autoExpanded?: boolean | null }>();
|
||||
const props = defineProps<{
|
||||
trace?: RunnerTrace | null;
|
||||
defaultExpanded?: boolean;
|
||||
storageKey?: string;
|
||||
autoExpanded?: boolean | null;
|
||||
finalResponsePlacement?: TraceFinalResponsePlacement;
|
||||
outerFinalResponseText?: string | null;
|
||||
}>();
|
||||
|
||||
const TRACE_RENDER_ROW_WINDOW = 120;
|
||||
const listRef = ref<HTMLOListElement | null>(null);
|
||||
@@ -17,7 +24,11 @@ const { following, keepBottomAfterUpdate, onScroll, scrollToBottom } = useWorkbe
|
||||
|
||||
const events = computed(() => Array.isArray(props.trace?.events) ? props.trace.events : []);
|
||||
const eventCount = computed(() => props.trace?.eventCount ?? events.value.length);
|
||||
const rawReadableRows = computed(() => expanded.value ? traceDisplayRows(traceRecord(props.trace), events.value as Record<string, unknown>[], { formatClock: formatDisplayClock }) : []);
|
||||
const rawReadableRows = computed(() => expanded.value ? traceDisplayRows(traceRecord(props.trace), events.value as Record<string, unknown>[], {
|
||||
formatClock: formatDisplayClock,
|
||||
finalResponsePlacement: props.finalResponsePlacement,
|
||||
outerFinalResponseText: props.outerFinalResponseText
|
||||
}) : []);
|
||||
const readableRows = computed(() => rawReadableRows.value);
|
||||
const hiddenRowCount = computed(() => Math.max(0, readableRows.value.length - TRACE_RENDER_ROW_WINDOW));
|
||||
const renderedRows = computed(() => hiddenRowCount.value > 0 ? readableRows.value.slice(-TRACE_RENDER_ROW_WINDOW) : readableRows.value);
|
||||
|
||||
@@ -16,7 +16,11 @@ const { following, keepBottomAfterUpdate, onScroll, scrollToBottom, toggleFollow
|
||||
const trace = computed(() => props.message.runnerTrace ?? null);
|
||||
const events = computed(() => Array.isArray(trace.value?.events) ? trace.value.events : []);
|
||||
const eventCount = computed(() => trace.value?.eventCount ?? events.value.length);
|
||||
const readableRows = computed(() => traceDisplayRows(traceRecord(trace.value), events.value as Record<string, unknown>[], { formatClock: formatDisplayClock }));
|
||||
const readableRows = computed(() => traceDisplayRows(traceRecord(trace.value), events.value as Record<string, unknown>[], {
|
||||
formatClock: formatDisplayClock,
|
||||
finalResponsePlacement: "outer",
|
||||
outerFinalResponseText: firstNonEmptyString(props.message.text)
|
||||
}));
|
||||
const noiseCount = computed(() => traceNoiseEventCount(events.value as Record<string, unknown>[]));
|
||||
const lastEvent = computed(() => events.value.at(-1) ?? null);
|
||||
const lastReadableRow = computed(() => readableRows.value.at(-1) ?? null);
|
||||
|
||||
@@ -145,7 +145,7 @@ function traceStorageKey(message: ChatMessage): string {
|
||||
<button v-if="detailEnabled && message.role === 'agent'" class="message-detail-button" type="button" aria-label="运行详情" title="运行详情" @click="emit('details', message)">!</button>
|
||||
</div>
|
||||
</header>
|
||||
<TraceTimeline v-if="message.role === 'agent' && message.runnerTrace" :trace="message.runnerTrace" :auto-expanded="autoExpanded" :storage-key="traceStorageKey(message)" />
|
||||
<TraceTimeline v-if="message.role === 'agent' && message.runnerTrace" :trace="message.runnerTrace" :auto-expanded="autoExpanded" :storage-key="traceStorageKey(message)" final-response-placement="outer" :outer-final-response-text="visibleText" />
|
||||
<LoadingState v-if="isAwaitingAgentBody(message)" class="message-loading" label="思考中..." compact />
|
||||
<MessageMarkdown v-if="visibleText" class="message-text" :source="visibleText" />
|
||||
<ApiErrorDiagnostic v-if="diagnostic.visible" class="message-diagnostic projection-diagnostic" :error="diagnostic.text" :api-error="diagnostic.apiError" :diagnostic="diagnostic.diagnostic" compact />
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
// SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-18-p0-unique-projection; PJ2026-010401 Web工作台 draft-2026-06-18-r1.
|
||||
// Responsibility: Trace snapshot helpers for Workbench realtime authority reducers.
|
||||
|
||||
import type { AgentChatResultResponse, AgentRunProvenance, ChatMessage, ProjectionDiagnostic, TraceEvent, WorkbenchTurnTimingProjection } from "@/types";
|
||||
import { firstNonEmptyString } from "@/utils";
|
||||
import type { AgentChatResultResponse, AgentRunProvenance, ChatMessage, ProjectionDiagnostic, TraceEvent, WorkbenchTurnTimingProjection } from "../types";
|
||||
import { firstNonEmptyString } from "../utils";
|
||||
|
||||
export interface TraceSnapshot {
|
||||
traceId?: string;
|
||||
|
||||
@@ -3,8 +3,8 @@
|
||||
// Mechanical source: /root/opencode/packages/app/src/context/global-sync/event-reducer.ts:21-48 global event classification, :93-270 session/message reducer action boundary.
|
||||
// Mechanical source: /root/opencode/packages/opencode/src/cli/cmd/run/session-data.ts:1-17 reducer side-effect boundary and replay notes.
|
||||
|
||||
import type { WorkbenchRealtimeEvent } from "@/api/workbench-events";
|
||||
import { firstNonEmptyString } from "@/utils";
|
||||
import type { WorkbenchRealtimeEvent } from "../api/workbench-events";
|
||||
import { firstNonEmptyString } from "../utils";
|
||||
import { workbenchRealtimePrimaryAuthorityDecision } from "./workbench-realtime-authority";
|
||||
|
||||
export type WorkbenchRealtimeAction =
|
||||
|
||||
@@ -3,13 +3,11 @@
|
||||
|
||||
import type { WorkbenchKafkaSseDebugEvent } from "@/api/workbench-debug";
|
||||
import type { WorkbenchRealtimeEvent } from "@/api/workbench-events";
|
||||
import { mergeRunnerTrace, snapshotToRunnerTrace } from "@/composables/workbench-trace-snapshot";
|
||||
import type { ChatMessage, TraceEvent } from "@/types";
|
||||
import type { ChatMessage } from "@/types";
|
||||
import { firstNonEmptyString } from "@/utils";
|
||||
import { reduceWorkbenchRealtimeEvent } from "./workbench-event-reducer";
|
||||
import { reduceWorkbenchLiveKafkaMessageState, workbenchLiveKafkaMessageId } from "./workbench-live-kafka-event";
|
||||
import { projectWorkbenchLiveKafkaMessage } from "./workbench-live-kafka-event";
|
||||
import { planWorkbenchRealtimeApply } from "./workbench-realtime-plan";
|
||||
import { realtimeSnapshotToTraceSnapshot } from "./workbench-trace-detail";
|
||||
|
||||
export interface WorkbenchIsolatedKafkaDebugLog {
|
||||
id: string;
|
||||
@@ -57,7 +55,14 @@ export function applyWorkbenchIsolatedKafkaDebugEvent(
|
||||
if (!traceStep || traceStep.type !== "apply-trace-event" || !traceStep.event) {
|
||||
return withLog(base, event, traceId, reduced.action.type, plan.steps.map((step) => step.type), false, reduced.action.type === "ignore" ? reduced.action.reason : "debug-envelope-not-a-trace-event");
|
||||
}
|
||||
const message = projectDebugMessage(state.message, traceId, event, traceStep.event);
|
||||
const message = projectWorkbenchLiveKafkaMessage({
|
||||
previous: state.message,
|
||||
traceId,
|
||||
sessionId: firstNonEmptyString(event.hwlabSessionId, event.sessionId, traceStep.event.sessionId, state.message?.sessionId) ?? "ses_workbench_isolated_debug",
|
||||
event: traceStep.event,
|
||||
receivedAt: new Date().toISOString(),
|
||||
title: "Code Agent · 隔离调试"
|
||||
});
|
||||
return withLog({ ...base, message, appliedCount: state.appliedCount + 1, error: null }, event, traceId, reduced.action.type, plan.steps.map((step) => step.type), true, null);
|
||||
}
|
||||
|
||||
@@ -72,54 +77,6 @@ export function workbenchCurrentDebugTraceId(messages: ChatMessage[], sessionLas
|
||||
) ?? null;
|
||||
}
|
||||
|
||||
function projectDebugMessage(previous: ChatMessage | null, traceId: string, realtimeEvent: WorkbenchRealtimeEvent, event: TraceEvent): ChatMessage {
|
||||
const liveState = reduceWorkbenchLiveKafkaMessageState({
|
||||
text: previous?.text ?? "",
|
||||
status: previous?.status ?? "running",
|
||||
terminal: false
|
||||
}, event);
|
||||
const sessionId = firstNonEmptyString(realtimeEvent.hwlabSessionId, realtimeEvent.sessionId, event.sessionId, previous?.sessionId) ?? "ses_workbench_isolated_debug";
|
||||
const updatedAt = firstNonEmptyString(event.createdAt, event.updatedAt, realtimeEvent.eventCreatedAt, realtimeEvent.serverSentAt) ?? new Date().toISOString();
|
||||
const startedAt = firstNonEmptyString(previous?.timing?.startedAt, updatedAt) ?? updatedAt;
|
||||
const snapshot = realtimeSnapshotToTraceSnapshot(traceId, {
|
||||
traceId,
|
||||
sessionId,
|
||||
status: liveState.status,
|
||||
events: [event],
|
||||
eventCount: 1,
|
||||
startedAt,
|
||||
lastEventAt: updatedAt,
|
||||
finishedAt: liveState.terminal ? updatedAt : null,
|
||||
durationMs: finiteNumber(event.elapsedMs)
|
||||
}, [event], { eventSource: "hwlab-kafka-sse" });
|
||||
const runnerTrace = mergeRunnerTrace(previous?.runnerTrace ?? null, snapshotToRunnerTrace(snapshot));
|
||||
const messageId = workbenchLiveKafkaMessageId(traceId);
|
||||
return {
|
||||
...(previous ?? {}),
|
||||
id: previous?.id ?? messageId,
|
||||
messageId: previous?.messageId ?? messageId,
|
||||
role: "agent",
|
||||
title: "Code Agent · 隔离调试",
|
||||
text: liveState.text,
|
||||
status: liveState.status as ChatMessage["status"],
|
||||
traceId,
|
||||
turnId: traceId,
|
||||
sessionId,
|
||||
createdAt: previous?.createdAt ?? startedAt,
|
||||
updatedAt,
|
||||
runnerTrace,
|
||||
traceAutoLifecycle: liveState.terminal ? "terminal" : "running",
|
||||
timing: {
|
||||
startedAt,
|
||||
lastEventAt: updatedAt,
|
||||
finishedAt: liveState.terminal ? updatedAt : null,
|
||||
durationMs: liveState.terminal ? finiteNumber(event.elapsedMs) : null,
|
||||
valuesRedacted: true
|
||||
},
|
||||
...(liveState.terminal && liveState.text ? { finalResponse: { text: liveState.text, status: liveState.status, traceId } } : {})
|
||||
} as ChatMessage;
|
||||
}
|
||||
|
||||
function rejected(state: WorkbenchIsolatedKafkaDebugState, event: WorkbenchRealtimeEvent, traceId: string | null, reason: string): WorkbenchIsolatedKafkaDebugState {
|
||||
return withLog(state, event, traceId, "ignore", [], false, reason);
|
||||
}
|
||||
@@ -154,8 +111,3 @@ function realtimeTraceId(event: WorkbenchRealtimeEvent): string | null {
|
||||
function recordValue(value: unknown): Record<string, unknown> {
|
||||
return value && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : {};
|
||||
}
|
||||
|
||||
function finiteNumber(value: unknown): number | null {
|
||||
const number = Number(value);
|
||||
return Number.isFinite(number) && number >= 0 ? Math.trunc(number) : null;
|
||||
}
|
||||
|
||||
@@ -0,0 +1,91 @@
|
||||
import assert from "node:assert/strict";
|
||||
import { test } from "bun:test";
|
||||
|
||||
import type { ChatMessage, TraceEvent } from "../types";
|
||||
import { projectWorkbenchLiveKafkaMessage } from "./workbench-live-kafka-event";
|
||||
|
||||
test("live Kafka projection refreshes lastEventAt from each ingress receipt", () => {
|
||||
const traceId = "trc_live_timing";
|
||||
const sessionId = "ses_live_timing";
|
||||
const firstReceivedAt = "2026-07-10T10:00:05.000Z";
|
||||
const secondReceivedAt = "2026-07-10T10:00:09.000Z";
|
||||
const first = projectWorkbenchLiveKafkaMessage({
|
||||
previous: null,
|
||||
traceId,
|
||||
sessionId,
|
||||
receivedAt: firstReceivedAt,
|
||||
event: traceEvent(1, "tool")
|
||||
});
|
||||
const second = projectWorkbenchLiveKafkaMessage({
|
||||
previous: first,
|
||||
traceId,
|
||||
sessionId,
|
||||
receivedAt: secondReceivedAt,
|
||||
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(second.timing?.lastEventAgeMs, 0);
|
||||
assert.equal(second.runnerTrace?.lastEventAt, secondReceivedAt);
|
||||
assert.equal(second.runnerTrace?.timing?.lastEventAt, secondReceivedAt);
|
||||
assert.equal(second.runnerTrace?.eventCount, 2);
|
||||
});
|
||||
|
||||
test("live Kafka projection clears stale terminal timing when a running event is accepted", () => {
|
||||
const traceId = "trc_live_reopened";
|
||||
const sessionId = "ses_live_reopened";
|
||||
const terminalAt = "2026-07-10T10:00:10.000Z";
|
||||
const terminal = projectWorkbenchLiveKafkaMessage({
|
||||
previous: seededMessage(traceId, sessionId),
|
||||
traceId,
|
||||
sessionId,
|
||||
receivedAt: terminalAt,
|
||||
event: traceEvent(2, "result", { terminal: true, status: "completed" })
|
||||
});
|
||||
const acceptedAt = "2026-07-10T10:00:12.000Z";
|
||||
const running = projectWorkbenchLiveKafkaMessage({
|
||||
previous: terminal,
|
||||
traceId,
|
||||
sessionId,
|
||||
receivedAt: acceptedAt,
|
||||
event: traceEvent(3, "tool")
|
||||
});
|
||||
|
||||
assert.equal(terminal.finishedAt, terminalAt);
|
||||
const terminalFinalResponse = terminal.finalResponse as { text?: string } | null | undefined;
|
||||
assert.equal(terminalFinalResponse?.text, "final answer");
|
||||
assert.equal(running.status, "running");
|
||||
assert.equal(running.lastEventAt, acceptedAt);
|
||||
assert.equal(running.finishedAt, null);
|
||||
assert.equal(running.durationMs, null);
|
||||
assert.equal(running.finalResponse, null);
|
||||
assert.equal(running.runnerTrace?.finishedAt, null);
|
||||
assert.equal(running.runnerTrace?.durationMs, null);
|
||||
});
|
||||
|
||||
function seededMessage(traceId: string, sessionId: string): ChatMessage {
|
||||
return projectWorkbenchLiveKafkaMessage({
|
||||
previous: null,
|
||||
traceId,
|
||||
sessionId,
|
||||
receivedAt: "2026-07-10T10:00:01.000Z",
|
||||
event: traceEvent(1, "assistant", { assistantText: "final answer" })
|
||||
});
|
||||
}
|
||||
|
||||
function traceEvent(sourceSeq: number, type: string, extra: Record<string, unknown> = {}): TraceEvent {
|
||||
return {
|
||||
traceId: "trc_live_timing",
|
||||
sessionId: "ses_live_timing",
|
||||
source: "agentrun.kafka",
|
||||
sourceEventId: `evt_live_${sourceSeq}`,
|
||||
sourceSeq,
|
||||
type,
|
||||
eventType: type,
|
||||
status: "running",
|
||||
createdAt: `2026-07-10T10:00:0${sourceSeq}.000Z`,
|
||||
...extra
|
||||
};
|
||||
}
|
||||
@@ -1,8 +1,9 @@
|
||||
// SPEC: PJ2026-0104010803 Workbench live Kafka SSE.
|
||||
// Responsibility: project one transparent hwlab.event.v1 business event into visible turn state without replay/finalizer/polling.
|
||||
|
||||
import type { TraceEvent } from "@/types";
|
||||
import { firstNonEmptyString } from "@/utils";
|
||||
import { mergeRunnerTrace } from "../composables/workbench-trace-snapshot";
|
||||
import type { ChatMessage, TraceEvent, WorkbenchTurnTimingProjection } from "../types";
|
||||
import { firstNonEmptyString } from "../utils";
|
||||
|
||||
export interface WorkbenchLiveMessageState {
|
||||
text: string;
|
||||
@@ -10,6 +11,15 @@ export interface WorkbenchLiveMessageState {
|
||||
terminal: boolean;
|
||||
}
|
||||
|
||||
export interface WorkbenchLiveKafkaProjectionInput {
|
||||
previous: ChatMessage | null;
|
||||
traceId: string;
|
||||
sessionId: string;
|
||||
event: TraceEvent;
|
||||
receivedAt: string;
|
||||
title?: string;
|
||||
}
|
||||
|
||||
export function reduceWorkbenchLiveKafkaMessageState(previous: WorkbenchLiveMessageState, event: TraceEvent): WorkbenchLiveMessageState {
|
||||
const terminal = workbenchLiveKafkaEventIsTerminal(event);
|
||||
const status = terminal ? firstNonEmptyString(event.status, "completed") ?? "completed" : "running";
|
||||
@@ -45,6 +55,87 @@ export function workbenchLiveKafkaTerminalText(event: TraceEvent, previousText:
|
||||
return text ?? `AgentRun ${status}`;
|
||||
}
|
||||
|
||||
export function projectWorkbenchLiveKafkaMessage(input: WorkbenchLiveKafkaProjectionInput): ChatMessage {
|
||||
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 liveState = reduceWorkbenchLiveKafkaMessageState({
|
||||
text: previous?.text ?? "",
|
||||
status: previous?.status ?? "running",
|
||||
terminal: false
|
||||
}, input.event);
|
||||
const startedAt = timestamp(previous?.timing?.startedAt)
|
||||
?? timestamp(previous?.startedAt)
|
||||
?? timestamp(input.event.createdAt)
|
||||
?? receivedAt;
|
||||
const durationMs = liveState.terminal ? Math.max(0, Date.parse(receivedAt) - Date.parse(startedAt)) : null;
|
||||
const timing: WorkbenchTurnTimingProjection = {
|
||||
startedAt,
|
||||
lastEventAt: receivedAt,
|
||||
finishedAt: liveState.terminal ? receivedAt : null,
|
||||
durationMs,
|
||||
observedAt: receivedAt,
|
||||
lastEventAgeMs: 0,
|
||||
valuesRedacted: true
|
||||
};
|
||||
const mergedRunnerTrace = mergeRunnerTrace(previous?.runnerTrace ?? null, {
|
||||
traceId: input.traceId,
|
||||
sessionId: input.sessionId,
|
||||
status: liveState.status,
|
||||
traceStatus: liveState.status,
|
||||
events: [input.event],
|
||||
eventCount: 1,
|
||||
eventSource: "hwlab-kafka-sse",
|
||||
timing,
|
||||
startedAt,
|
||||
lastEventAt: receivedAt,
|
||||
finishedAt: liveState.terminal ? receivedAt : null,
|
||||
durationMs,
|
||||
updatedAt: receivedAt
|
||||
});
|
||||
const runnerTrace = {
|
||||
...mergedRunnerTrace,
|
||||
status: liveState.status,
|
||||
traceStatus: liveState.status,
|
||||
timing,
|
||||
startedAt,
|
||||
lastEventAt: receivedAt,
|
||||
finishedAt: liveState.terminal ? receivedAt : null,
|
||||
durationMs,
|
||||
updatedAt: receivedAt
|
||||
};
|
||||
const messageId = previous?.messageId ?? previous?.id ?? workbenchLiveKafkaMessageId(input.traceId);
|
||||
return {
|
||||
...(previous ?? {}),
|
||||
id: previous?.id ?? messageId,
|
||||
messageId,
|
||||
role: "agent",
|
||||
title: input.title ?? previous?.title ?? "Code Agent",
|
||||
text: liveState.text,
|
||||
status: liveState.status as ChatMessage["status"],
|
||||
traceId: input.traceId,
|
||||
turnId: previous?.turnId ?? input.traceId,
|
||||
sessionId: input.sessionId,
|
||||
createdAt: previous?.createdAt ?? startedAt,
|
||||
updatedAt: receivedAt,
|
||||
startedAt,
|
||||
lastEventAt: receivedAt,
|
||||
finishedAt: liveState.terminal ? receivedAt : null,
|
||||
durationMs,
|
||||
runnerTrace,
|
||||
traceAutoLifecycle: liveState.terminal ? "terminal" : "running",
|
||||
timing,
|
||||
finalResponse: liveState.terminal && liveState.text
|
||||
? { text: liveState.text, status: liveState.status, traceId: input.traceId }
|
||||
: null
|
||||
} as ChatMessage;
|
||||
}
|
||||
|
||||
function timestamp(value: unknown): string | null {
|
||||
const text = firstNonEmptyString(value);
|
||||
return text && Number.isFinite(Date.parse(text)) ? text : null;
|
||||
}
|
||||
|
||||
function recordValue(value: unknown): Record<string, unknown> | null {
|
||||
return value && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : null;
|
||||
}
|
||||
|
||||
@@ -3,9 +3,9 @@
|
||||
// Mechanical source: /root/opencode/packages/app/src/context/global-sync/event-reducer.ts:93-270 session/message reducer action boundary.
|
||||
// Mechanical source: /root/opencode/packages/opencode/src/cli/cmd/run/session-data.ts:1-17 reducer side-effect boundary and replay notes.
|
||||
|
||||
import type { WorkbenchRealtimeEvent } from "@/api/workbench-events";
|
||||
import type { WorkbenchStreamTransportRecovery } from "@/utils/workbench-realtime-runtime";
|
||||
import { firstNonEmptyString, normalizeWorkbenchSessionId } from "@/utils";
|
||||
import type { WorkbenchRealtimeEvent } from "../api/workbench-events";
|
||||
import { firstNonEmptyString, normalizeWorkbenchSessionId } from "../utils";
|
||||
import type { WorkbenchStreamTransportRecovery } from "../utils/workbench-realtime-runtime";
|
||||
import type { WorkbenchRealtimeAction } from "./workbench-event-reducer";
|
||||
|
||||
export type WorkbenchRealtimeApplyStep =
|
||||
|
||||
@@ -22,7 +22,7 @@ import { initialWorkbenchSessionIdFromLocation } from "./workbench-projection";
|
||||
import { cleanupWorkbenchServerStateSessions, selectActiveMessages, selectActiveSession, selectSessionList, selectSessionStatusAuthority, selectTraceAuthorityById, selectTurnStatusAuthority, type WorkbenchServerAction } from "./workbench-server-state";
|
||||
import { cleanupDroppedWorkbenchSessionCaches, trimWorkbenchSessionCache } from "./workbench-session-cache";
|
||||
import { reduceWorkbenchRealtimeEvent, workbenchRealtimeEventIsBusinessActivity, type WorkbenchRealtimeAction } from "./workbench-event-reducer";
|
||||
import { reduceWorkbenchLiveKafkaMessageState, workbenchLiveKafkaEventIsTerminal, workbenchLiveKafkaMessageId } from "./workbench-live-kafka-event";
|
||||
import { projectWorkbenchLiveKafkaMessage } from "./workbench-live-kafka-event";
|
||||
import { messageHasSealedTerminalResult, messageIsSealedTerminal, traceAuthorityIsSealed } from "./workbench-terminal-authority";
|
||||
import { boundedProjectionMessageLimit, mergeBoundedProjectionMessages, selectProjectionMessageWindow, traceProjectionIsTerminalSealed } from "./workbench-message-projection-budget";
|
||||
import {
|
||||
@@ -1360,56 +1360,52 @@ export const useWorkbenchStore = defineStore("workbench", () => {
|
||||
}
|
||||
const events = event ? [event] : Array.isArray(snapshot?.events) ? snapshot.events : [];
|
||||
markWorkbenchTraceEventsReceived({ traceId: id, events, transport: "sse", serverSentAt: realtimeEvent?.serverSentAt, eventCreatedAt: realtimeEvent?.eventCreatedAt, traceSeq: realtimeEvent?.traceSeq ?? realtimeEvent?.cursor?.traceSeq });
|
||||
const liveTerminal = liveKafkaEnvelope && workbenchLiveKafkaEventIsTerminal(event);
|
||||
const eventSnapshot = snapshot ?? { traceId: id, status: liveKafkaEnvelope && !liveTerminal ? "running" : event?.status, events };
|
||||
if (liveKafkaEnvelope && event) applyLiveKafkaBusinessEvent(id, sessionId, event);
|
||||
applyTraceSnapshot(id, realtimeSnapshotToTraceSnapshot(id, eventSnapshot, events, liveKafkaEnvelope ? { eventSource: "hwlab-kafka-sse" } : undefined), liveKafkaEnvelope);
|
||||
if (liveKafkaEnvelope && event) {
|
||||
applyLiveKafkaBusinessEvent(id, sessionId, event, new Date().toISOString());
|
||||
return;
|
||||
}
|
||||
const eventSnapshot = snapshot ?? { traceId: id, status: event?.status, events };
|
||||
applyTraceSnapshot(id, realtimeSnapshotToTraceSnapshot(id, eventSnapshot, events));
|
||||
}
|
||||
|
||||
function applyLiveKafkaBusinessEvent(traceId: string, authoritySessionId: string | null, event: TraceEvent): void {
|
||||
function applyLiveKafkaBusinessEvent(traceId: string, authoritySessionId: string | null, event: TraceEvent, receivedAt: string): void {
|
||||
const eventSessionId = firstNonEmptyString(event.sessionId);
|
||||
const ownerSessionId = traceOwnerSessionId(traceId, authoritySessionId ?? eventSessionId, true);
|
||||
if (!ownerSessionId) return;
|
||||
const previousMessage = (serverState.value.messagesBySessionId[ownerSessionId] ?? []).find((message) => messageMatchesTraceAuthority(message, traceId, authoritySessionId ?? eventSessionId, ownerSessionId, true)) ?? null;
|
||||
const liveState = reduceWorkbenchLiveKafkaMessageState({ text: previousMessage?.text ?? "", status: previousMessage?.status ?? "running", terminal: false }, event);
|
||||
const terminal = liveState.terminal;
|
||||
const status = liveState.status;
|
||||
const assistantText = liveState.text;
|
||||
const updatedAt = firstNonEmptyString(event.createdAt, event.updatedAt) ?? new Date().toISOString();
|
||||
const messageId = workbenchLiveKafkaMessageId(traceId);
|
||||
const message = previousMessage ?? makeMessage("agent", "", "running", {
|
||||
id: messageId,
|
||||
messageId,
|
||||
const message = projectWorkbenchLiveKafkaMessage({
|
||||
previous: previousMessage,
|
||||
traceId,
|
||||
sessionId: ownerSessionId,
|
||||
title: "Code Agent",
|
||||
traceAutoLifecycle: "running",
|
||||
createdAt: updatedAt,
|
||||
updatedAt
|
||||
event,
|
||||
receivedAt
|
||||
});
|
||||
reduceServerState({
|
||||
type: "message.upsert",
|
||||
sessionId: ownerSessionId,
|
||||
message: {
|
||||
...message,
|
||||
...(assistantText ? { text: assistantText } : {}),
|
||||
status: status as ChatMessage["status"],
|
||||
traceAutoLifecycle: terminal ? "terminal" : "running",
|
||||
...(terminal && assistantText ? { finalResponse: { text: assistantText, status, traceId } } : {}),
|
||||
updatedAt
|
||||
}
|
||||
message
|
||||
});
|
||||
if (message.runnerTrace) rememberTraceAuthority(message.runnerTrace);
|
||||
markWorkbenchTraceProjected(traceId);
|
||||
const terminal = message.traceAutoLifecycle === "terminal";
|
||||
const status = firstNonEmptyString(message.status) ?? "running";
|
||||
const finalResponse = terminal && message.text ? { text: message.text, status, traceId } : undefined;
|
||||
rememberTurnStatus(traceId, {
|
||||
traceId,
|
||||
sessionId: ownerSessionId,
|
||||
status,
|
||||
running: !terminal,
|
||||
terminal,
|
||||
finalResponse: terminal && assistantText ? { text: assistantText, status, traceId } : undefined,
|
||||
updatedAt
|
||||
finalResponse,
|
||||
timing: message.timing,
|
||||
startedAt: message.timing?.startedAt,
|
||||
lastEventAt: message.timing?.lastEventAt,
|
||||
finishedAt: message.timing?.finishedAt,
|
||||
durationMs: message.timing?.durationMs,
|
||||
updatedAt: receivedAt
|
||||
} as AgentChatResultResponse);
|
||||
const existing = sessions.value.find((session) => session.sessionId === ownerSessionId) ?? null;
|
||||
if (existing) rememberSessionList(mergeSessionIntoList(sessions.value, { ...existing, status, lastTraceId: traceId, updatedAt: firstNonEmptyString(event.createdAt, event.updatedAt) ?? new Date().toISOString() }));
|
||||
if (existing) rememberSessionList(mergeSessionIntoList(sessions.value, { ...existing, status, lastTraceId: traceId, updatedAt: receivedAt }));
|
||||
if (terminal && currentRequest.value?.traceId === traceId) {
|
||||
chatPending.value = false;
|
||||
currentRequest.value = null;
|
||||
|
||||
Reference in New Issue
Block a user