diff --git a/AGENTS.md b/AGENTS.md index ec0db56c..78b25982 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -44,6 +44,7 @@ HWLAB 是硬件实验室运行面和控制面项目。本文件只作为低噪 - Code Agent 真实回复、provider blocker 与 readiness：[docs/reference/code-agent-chat-readiness.md](docs/reference/code-agent-chat-readiness.md) - OpenCode 独立 UI 接入、iframe 鉴权换票与 OTel 追踪：[docs/reference/opencode-integration.md](docs/reference/opencode-integration.md) - AgentRun 手动调度与 UniDesk SSH passthrough：[docs/reference/agentrun-code-agent-dispatch.md](docs/reference/agentrun-code-agent-dispatch.md) +- Kafka session 源码直连与独立 debug topic 单步调试：[docs/reference/kafka-source-direct-debug.md](docs/reference/kafka-source-direct-debug.md) - 文档治理与 docs-spec 本地权威：[docs/reference/documentation-governance.md](docs/reference/documentation-governance.md) - 中文优先规则：[docs/reference/chinese-first-documentation.md](docs/reference/chinese-first-documentation.md) - 用户反馈分流：[docs/reference/user-feedback-triage.md](docs/reference/user-feedback-triage.md) diff --git a/docs/reference/kafka-source-direct-debug.md b/docs/reference/kafka-source-direct-debug.md new file mode 100644 index 00000000..c2ee1a16 --- /dev/null +++ b/docs/reference/kafka-source-direct-debug.md @@ -0,0 +1,41 @@ +# Kafka 源码直连调试 + +- `hwlab-cli kafka regenerate hwlab` 用于把指定 AgentRun session 的事件直接映射为 HWLAB debug 事件： + - 调用生产路径同源的 AgentRun 解码与 HWLAB 事件映射函数； + - 不启动 Cloud API； + - 不访问数据库； + - 不依赖 transactional projector； + - 默认输出到 `hwlab.event.debug.v1`，并禁止写入产品 topic `hwlab.event.v1`。 + +- Kafka 输入适合验证真实 topic 中的 session 事件： + - 命令为 `hwlab-cli kafka regenerate hwlab --from kafka --session-id `； + - 默认输入 topic 为 `agentrun.event.debug.v1`； + - 产品 canonical 输入可显式指定 `--input-topic agentrun.event.v1`； + - output topic 优先读取 owning YAML 注入的 `HWLAB_KAFKA_HWLAB_DEBUG_EVENT_TOPIC`； + - consumer group prefix 优先读取 owning YAML 注入的 `HWLAB_KAFKA_HWLAB_DEBUG_GROUP_PREFIX`； + - 未注入时使用离线合同值 `hwlab.event.debug.v1` 与 `hwlab-v03-workbench-isolated-debug`； + - Workbench 消费 debug topic 由 `HWLAB_WORKBENCH_KAFKA_DEBUG_REPLAY_ENABLED` 独立控制； + - CLI 的离线映射不依赖该运行面开关； + - Kafka 输入默认发布到独立 HWLAB debug topic； + - `--no-publish` 可只生成本地证据。 + +- JSONL 输入适合完全离线的单步测试： + - 命令为 `hwlab-cli kafka regenerate hwlab --from jsonl --session-id --jsonl-file --no-publish`； + - 每行可直接放 envelope，也可使用 `{topic,partition,offset,key,value,headers}` transport wrapper； + - JSONL 输入默认不连接 Kafka； + - 完整映射结果写入 `.state/hwlab-cli/kafka-debug/` 下的 JSONL artifact； + - `--output-jsonl` 可显式选择 artifact 路径。 + +- 输入合同按来源隔离： + - `agentrun.event.v1` 必须通过严格 canonical decoder； + - `agentrun.event.reconstruction.v1` 只允许进入 debug decoder； + - stdio reconstruction 的 `eventId`、`outboxSeq`、`originalEventId` 和 `originalOutboxSeq` 必须保持 `null`； + - reconstruction 必须声明 `stdio-derived partial reconstruction`； + - reconstruction 必须保留 `identityDerivation`，禁止静默猜测 HWLAB session identity。 + +- 输出验证采用一对一和原序映射： + - `--expect-count ` 在映射和发布前锁定输入数量； + - 命令验证 source sequence、trace、raw AgentRun session、派生 HWLAB session 和源消息哈希； + - 默认输出为 compact text； + - 只有显式 `--json` 才输出结构化 JSON； + - 成功和失败都返回可执行的 `next` hint。 diff --git a/internal/cloud/kafka-event-bridge.ts b/internal/cloud/kafka-event-bridge.ts index f61412ca..e8be13e6 100644 --- a/internal/cloud/kafka-event-bridge.ts +++ b/internal/cloud/kafka-event-bridge.ts @@ -15,6 +15,10 @@ const TRUE_VALUES = new Set(["1", "true", "yes", "on"]); const DEFAULT_AGENTRUN_EVENT_TOPIC = "agentrun.event.v1"; 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"; +export const AGENTRUN_STDIO_RECONSTRUCTION_SCHEMA = "agentrun.event.reconstruction.v1"; +export const AGENTRUN_STDIO_RECONSTRUCTION_KIND = "stdio-derived partial reconstruction"; const DEFAULT_CLIENT_ID = "hwlab-v03-cloud-api"; const DEFAULT_QUERY_TIMEOUT_MS = 5000; const DEFAULT_QUERY_LIMIT = 50; @@ -427,10 +431,7 @@ export async function publishAgentRunKafkaMessageLive(kafkaMessage = {}, { produ sourceKey: transport.sourceKey, inputSha256: transport.inputSha256 }); - const publishMetadata = await producer.send({ - topic: config.hwlabTopic, - messages: [{ key: kafkaMessageKey(projected), value: JSON.stringify(projected), headers: kafkaHeaders(projected) }] - }); + const publishMetadata = await producer.send({ topic: config.hwlabTopic, messages: [buildHwlabKafkaProducerMessage(projected)] }); const publishedRecord = Array.isArray(publishMetadata) ? publishMetadata[0] : null; emitLiveKafkaOtelSpan("hwlab.kafka.live.direct_publish", projected, { topic: config.hwlabTopic, @@ -666,6 +667,184 @@ export function decodeCanonicalAgentRunKafkaMessage(kafkaMessage = {}) { } } +export function decodeAgentRunDebugProjectionMessage(kafkaMessage = {}) { + const valueText = kafkaMessage?.message?.value ? Buffer.from(kafkaMessage.message.value).toString("utf8") : ""; + const envelope = parseJson(valueText); + if (envelope?.schema === "agentrun.event.v1") { + const decoded = decodeCanonicalAgentRunKafkaMessage(kafkaMessage); + return decoded.ok + ? { ...decoded, inputKind: "canonical-durable", reconstruction: null } + : { ...decoded, inputKind: "canonical-invalid", reconstruction: null }; + } + try { + assertAgentRunStdioReconstruction(envelope); + return { + ok: true, + event: envelope, + sourceEventId: null, + inputSha256: sha256(valueText), + inputKind: AGENTRUN_STDIO_RECONSTRUCTION_KIND, + reconstruction: envelope.reconstruction, + valuesPrinted: false + }; + } catch (error) { + return { + ok: false, + envelope, + sourceEventId: null, + inputSha256: sha256(valueText), + inputKind: "reconstruction-invalid", + reconstruction: envelope?.reconstruction ?? null, + error: { + code: error?.code ?? "workbench_kafka_debug_message_invalid", + message: errorMessage(error), + valuesRedacted: true + }, + valuesPrinted: false + }; + } +} + +export function projectAgentRunKafkaMessageToHwlabDebugEvent(kafkaMessage = {}, options = {}) { + const decoded = decodeAgentRunDebugProjectionMessage(kafkaMessage); + if (!decoded.ok) return decoded; + const transport = kafkaTransport(kafkaMessage); + const projected = projectAgentRunKafkaEventToHwlabEvent(decoded.event, { + source: stringValue(options.source) || "hwlab-cli-kafka-debug", + producedAt: options.producedAt, + sourceTopic: transport.sourceTopic, + sourcePartition: transport.sourcePartition, + sourceOffset: transport.sourceOffset, + sourceKey: transport.sourceKey, + inputSha256: transport.inputSha256 + }); + if (!stringValue(projected?.traceId) || (!decoded.reconstruction && !stringValue(projected?.sessionId))) { + return { + ok: false, + inputKind: decoded.inputKind, + error: { + code: "hwlab_kafka_debug_correlation_missing", + message: decoded.reconstruction + ? "AgentRun reconstruction debug projection requires traceId lineage." + : "Canonical AgentRun debug projection requires traceId and sessionId lineage.", + valuesRedacted: true + }, + valuesPrinted: false + }; + } + const debugSessionId = decoded.reconstruction ? stringValue(decoded.event.hwlabSessionId) : stringValue(projected.sessionId); + const debugEnvelope = { + ...projected, + schema: DEFAULT_HWLAB_DEBUG_EVENT_TOPIC, + eventType: "hwlab.trace.event.debug-projected", + sessionId: debugSessionId, + hwlabSessionId: debugSessionId, + event: decoded.reconstruction ? { ...projected.event, sessionId: debugSessionId } : projected.event, + outboxSeq: null, + ...(decoded.reconstruction ? { eventId: null, sourceEventId: null } : {}), + diagnostics: { + projectedBy: "hwlab-kafka-event-bridge", + projectionMode: "source-direct-debug", + inputKind: decoded.inputKind, + valuesPrinted: false + }, + debugLineage: { + inputSchema: firstText(decoded.event.schema), + inputEventType: firstText(decoded.event.eventType), + inputEventId: firstText(decoded.event.eventId), + inputOutboxSeq: integerValue(decoded.event.outboxSeq), + inputTraceId: firstText(decoded.event.traceId, decoded.event.event?.payload?.traceId), + inputSessionId: firstText(decoded.event.sessionId, decoded.event.run?.sessionId), + inputHwlabSessionId: firstText(decoded.event.hwlabSessionId, decoded.event.run?.hwlabSessionId), + inputRunId: firstText(decoded.event.runId, decoded.event.run?.runId), + inputCommandId: firstText(decoded.event.commandId, decoded.event.command?.commandId), + sourceSeq: integerValue(decoded.event.sourceSeq ?? decoded.event.event?.seq), + valuesPrinted: false + }, + ...(decoded.reconstruction ? { + reconstruction: { + ...decoded.reconstruction, + observedAt: timestampValue(decoded.event.observedAt), + inputSchema: decoded.event.schema, + inputEventType: decoded.event.eventType, + valuesPrinted: false + } + } : {}), + valuesPrinted: false + }; + return { + ok: true, + inputKind: decoded.inputKind, + envelope: debugEnvelope, + sourceSeq: integerValue(decoded.event.sourceSeq ?? decoded.event.event?.seq), + traceId: debugEnvelope.traceId, + sessionId: debugEnvelope.sessionId, + valuesPrinted: false + }; +} + +function assertAgentRunStdioReconstruction(value) { + if (!value || typeof value !== "object" || Array.isArray(value)) throw contractError("hwlab_kafka_debug_json_invalid", "AgentRun reconstruction value must be a JSON object."); + if (value.schema !== AGENTRUN_STDIO_RECONSTRUCTION_SCHEMA || value.eventType !== "agentrun.event.reconstructed") { + throw contractError("hwlab_kafka_debug_schema_invalid", "AgentRun debug input must be a canonical committed event or a stdio reconstruction envelope."); + } + if (value.eventId !== null || value.outboxSeq !== null) { + throw contractError("hwlab_kafka_debug_identity_invalid", "Stdio reconstruction must keep eventId and outboxSeq null."); + } + for (const [name, field] of [["traceId", value.traceId], ["sessionId", value.sessionId], ["runId", value.runId]]) { + if (!stringValue(field)) throw contractError("hwlab_kafka_debug_lineage_invalid", `Stdio reconstruction ${name} is required.`); + } + if (!Number.isInteger(Number(value.sourceSeq)) || Number(value.sourceSeq) <= 0) { + throw contractError("hwlab_kafka_debug_sequence_invalid", "Stdio reconstruction sourceSeq must be a positive integer."); + } + if (!value.event || typeof value.event !== "object" || Array.isArray(value.event) || value.event.id !== null || Number(value.event.seq) !== Number(value.sourceSeq)) { + throw contractError("hwlab_kafka_debug_identity_invalid", "Stdio reconstruction event identity must remain null and its seq must match sourceSeq."); + } + if (!value.event.payload || typeof value.event.payload !== "object" || Array.isArray(value.event.payload)) { + throw contractError("hwlab_kafka_debug_schema_invalid", "Stdio reconstruction event payload must be an object."); + } + if (value.event.payload.reconstructed !== true || value.event.payload.reconstructionKind !== AGENTRUN_STDIO_RECONSTRUCTION_KIND) { + throw contractError("hwlab_kafka_debug_lineage_invalid", "Stdio reconstruction payload must disclose its partial reconstruction kind."); + } + const reconstruction = value.reconstruction; + if (!reconstruction || typeof reconstruction !== "object" || Array.isArray(reconstruction) + || reconstruction.kind !== AGENTRUN_STDIO_RECONSTRUCTION_KIND + || reconstruction.completeLifecycle !== false + || reconstruction.originalEventId !== null + || reconstruction.originalOutboxSeq !== null) { + throw contractError("hwlab_kafka_debug_lineage_invalid", "Stdio reconstruction lineage is incomplete or claims durable identity."); + } + assertReconstructionIdentityDerivation(value, reconstruction.identityDerivation); + if (value.valuesPrinted !== false || containsSecretLikeKey(value)) { + throw contractError("hwlab_kafka_debug_redaction_invalid", "Stdio reconstruction must be redacted and declare valuesPrinted=false."); + } + return value; +} + +function assertReconstructionIdentityDerivation(value, derivation) { + if (!derivation || typeof derivation !== "object" || Array.isArray(derivation) || derivation.valuesPrinted !== false + || derivation.sourceField !== "sessionId" || derivation.targetField !== "hwlabSessionId") { + throw contractError("hwlab_kafka_debug_identity_derivation_invalid", "Stdio reconstruction must disclose the session identity derivation contract."); + } + const rawSessionId = stringValue(value.sessionId); + const match = rawSessionId?.match(/^ses_agentrun_([0-9a-fA-F]{8})_([0-9a-fA-F]{4})_([0-9a-fA-F]{4})_([0-9a-fA-F]{4})_([0-9a-fA-F]{12})$/u); + if (match) { + const expectedHwlabSessionId = `ses_${match.slice(1).join("-").toLowerCase()}`; + if (value.hwlabSessionId !== expectedHwlabSessionId || value.run?.hwlabSessionId !== expectedHwlabSessionId + || derivation.kind !== "agentrun-session-uuid-to-hwlab-session" || derivation.reversible !== true + || derivation.sourcePrefix !== "ses_agentrun_" || derivation.targetPrefix !== "ses_" + || derivation.separatorTransform !== "underscore-to-hyphen") { + throw contractError("hwlab_kafka_debug_identity_derivation_invalid", "Derived hwlabSessionId does not match the disclosed reversible AgentRun session identity transform."); + } + return; + } + if (value.hwlabSessionId !== null || value.run?.hwlabSessionId !== null + || derivation.kind !== "none" || derivation.reversible !== false + || derivation.reason !== "source sessionId is not ses_agentrun_") { + throw contractError("hwlab_kafka_debug_identity_derivation_invalid", "Non-reversible AgentRun session identity must keep hwlabSessionId null and disclose why."); + } +} + function assertCanonicalAgentRunEvent(value) { if (!value || typeof value !== "object" || Array.isArray(value)) throw contractError("workbench_kafka_json_invalid", "Kafka value must be a JSON object."); assertClosedObject(value, ["schema", "eventType", "source", "eventId", "outboxSeq", "sourceSeq", "committedAt", "traceId", "sessionId", "hwlabSessionId", "runId", "commandId", "run", "command", "event", "valuesPrinted"], "AgentRun envelope"); @@ -718,7 +897,7 @@ function kafkaTransport(kafkaMessage = {}) { sourcePartition: integerValue(kafkaMessage.partition), sourceOffset: stringValue(kafkaMessage.message?.offset), sourceKey: kafkaMessage.message?.key ? Buffer.from(kafkaMessage.message.key).toString("utf8") : null, - inputSha256: sha256(valueText), + inputSha256: stringValue(kafkaMessage.inputSha256) || sha256(valueText), valuesPrinted: false }; } @@ -734,7 +913,7 @@ function requireKafkaProjectorStore(runtimeStore, capabilities = {}) { if (missing.length > 0) throw contractError("hwlab_kafka_projector_store_invalid", `Kafka durable capabilities require a runtime store: ${missing.join(", ")}`); } -export async function queryKafkaEventStream({ env = process.env, stream = "hwlab", topic = null, traceId = null, sessionId = null, runId = null, commandId = null, limit = DEFAULT_QUERY_LIMIT, timeoutMs = DEFAULT_QUERY_TIMEOUT_MS, fromBeginning = true, kafkaFactory = defaultKafkaFactory } = {}) { +export async function queryKafkaEventStream({ env = process.env, stream = "hwlab", topic = null, traceId = null, sessionId = null, runId = null, commandId = null, limit = DEFAULT_QUERY_LIMIT, timeoutMs = DEFAULT_QUERY_TIMEOUT_MS, fromBeginning = true, groupIdPrefix = null, kafkaFactory = defaultKafkaFactory } = {}) { const brokers = csv(env.HWLAB_KAFKA_BOOTSTRAP_SERVERS); if (brokers.length === 0) throw new Error("HWLAB_KAFKA_BOOTSTRAP_SERVERS is required for Kafka event queries."); const clientId = stringValue(env.HWLAB_KAFKA_CLIENT_ID) || DEFAULT_CLIENT_ID; @@ -742,7 +921,9 @@ export async function queryKafkaEventStream({ env = process.env, stream = "hwlab const maxEvents = Math.max(1, integerValue(limit) || DEFAULT_QUERY_LIMIT); const budgetMs = Math.max(250, integerValue(timeoutMs) || DEFAULT_QUERY_TIMEOUT_MS); const kafka = kafkaFactory({ brokers, clientId: `${clientId}-query` }); - const consumer = kafka.consumer({ groupId: `${clientId}-query-${Date.now()}-${randomUUID().slice(0, 8)}`, allowAutoTopicCreation: false }); + const resolvedGroupIdPrefix = stringValue(groupIdPrefix) || `${clientId}-query`; + const groupId = `${resolvedGroupIdPrefix}-${Date.now()}-${randomUUID().slice(0, 8)}`; + const consumer = kafka.consumer({ groupId, allowAutoTopicCreation: false }); const events = []; let timer = null; let running = false; @@ -770,6 +951,7 @@ export async function queryKafkaEventStream({ env = process.env, stream = "hwlab 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(); @@ -785,6 +967,7 @@ export async function queryKafkaEventStream({ env = process.env, stream = "hwlab ok: true, stream, topic: resolvedTopic, + groupId, count: events.length, limit: maxEvents, timeoutMs: budgetMs, @@ -976,6 +1159,14 @@ async function boundedDisconnect(consumer, timeoutMs = 2000) { } } +export function buildHwlabKafkaProducerMessage(projected) { + return { + key: kafkaMessageKey(projected), + value: JSON.stringify(projected), + headers: kafkaHeaders(projected) + }; +} + function kafkaMessageKey(projected) { return firstText(projected.sessionId, projected.traceId, projected.context?.commandId, projected.context?.runId) || "hwlab-event"; } @@ -1032,6 +1223,7 @@ function stringValue(value) { } function integerValue(value) { + if (value === null || value === undefined || (typeof value === "string" && value.trim() === "")) return null; const parsed = Number(value); return Number.isFinite(parsed) ? Math.floor(parsed) : null; } diff --git a/tools/hwlab-cli/bin/hwlab-cli.ts b/tools/hwlab-cli/bin/hwlab-cli.ts index b9aa0833..dfc91ad9 100644 --- a/tools/hwlab-cli/bin/hwlab-cli.ts +++ b/tools/hwlab-cli/bin/hwlab-cli.ts @@ -1,4 +1,8 @@ #!/usr/bin/env bun import { main } from "../../src/hwlab-cli-lib.ts"; -await main(); +const argv = process.argv.slice(2); +if (argv[0] === "kafka") { + const { mainKafkaCli } = await import("../../src/hwlab-cli/kafka-regenerate.ts"); + await mainKafkaCli(argv.slice(1)); +} else await main(argv); diff --git a/tools/hwlab-cli/kafka-regenerate.test.ts b/tools/hwlab-cli/kafka-regenerate.test.ts new file mode 100644 index 00000000..06a00f52 --- /dev/null +++ b/tools/hwlab-cli/kafka-regenerate.test.ts @@ -0,0 +1,330 @@ +import assert from "node:assert/strict"; +import { mkdtemp, readFile, writeFile } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; + +import { test } from "bun:test"; + +import { queryKafkaEventStream } from "../../internal/cloud/kafka-event-bridge.ts"; +import { mapAgentRunRecordsToHwlabDebugEvents, runKafkaCli } from "../src/hwlab-cli/kafka-regenerate.ts"; + +const SESSION_ID = "ses_agentrun_5ec4e141_6abc_466d_9afe_049f7c0ac105"; +const HWLAB_SESSION_ID = "ses_5ec4e141-6abc-466d-9afe-049f7c0ac105"; +const TRACE_ID = "trc_mretx18t3jl4tg"; +const RUN_ID = "run_a78cbcb05f2c4afaa748a3158db44998"; + +test("offline JSONL maps 35 stdio reconstructions to 35 ordered HWLAB debug events without runtime dependencies", async () => { + const cwd = await mkdtemp(path.join(os.tmpdir(), "hwlab-kafka-debug-jsonl-")); + const inputFile = path.join(cwd, "agentrun-events.jsonl"); + const outputFile = path.join(cwd, "hwlab-events.jsonl"); + const records = Array.from({ length: 35 }, (_, index) => reconstructionRecord(index + 1)); + await writeFile(inputFile, records.map((record) => JSON.stringify(record)).join("\n") + "\n", "utf8"); + + const result = await runKafkaCli([ + "regenerate", "hwlab", + "--from", "jsonl", + "--session-id", SESSION_ID, + "--trace-id", TRACE_ID, + "--jsonl-file", inputFile, + "--output-jsonl", outputFile, + "--expect-count", "35", + "--no-publish", + "--json" + ], { + cwd, + env: {}, + now: () => "2026-07-10T12:00:00.000Z", + async createProducer() { throw new Error("offline JSONL must not create a producer"); } + }); + + assert.equal(result.exitCode, 0); + assert.equal(result.payload.input.readCount, 35); + assert.equal(result.payload.output.count, 35); + assert.equal(result.payload.output.publishedCount, 0); + assert.deepEqual(result.payload.validation, { + expectedCount: 35, + oneToOne: true, + orderPreserved: true, + lineagePreserved: true, + firstSourceSeq: 1, + lastSourceSeq: 35, + inputKinds: ["stdio-derived partial reconstruction"] + }); + assert.deepEqual(result.payload.runtimeDependencies, { + cloudApi: false, + database: false, + transactionalProjector: false, + kafka: false + }); + + const output = (await readFile(outputFile, "utf8")).trim().split("\n").map(JSON.parse); + assert.equal(output.length, 35); + assert.deepEqual(output.map((row) => row.value.debugLineage.sourceSeq), Array.from({ length: 35 }, (_, index) => index + 1)); + assert.ok(output.every((row) => row.topic === "hwlab.event.debug.v1")); + assert.ok(output.every((row) => row.value.schema === "hwlab.event.debug.v1")); + assert.ok(output.every((row) => row.value.eventId === null && row.value.sourceEventId === null && row.value.outboxSeq === null)); + assert.ok(output.every((row) => row.value.debugLineage.inputEventId === null && row.value.debugLineage.inputOutboxSeq === null)); + assert.ok(output.every((row) => row.value.sourceEvent.partition === null && row.value.sourceEvent.offset === null)); + assert.ok(output.every((row) => row.value.traceId === TRACE_ID && row.value.sessionId === HWLAB_SESSION_ID)); + assert.ok(output.every((row) => row.value.debugLineage.inputSessionId === SESSION_ID)); + assert.ok(output.every((row) => row.value.debugLineage.inputHwlabSessionId === HWLAB_SESSION_ID)); + assert.ok(output.every((row) => row.value.reconstruction.kind === "stdio-derived partial reconstruction")); + assert.ok(output.every((row) => row.value.reconstruction.originalEventId === null && row.value.reconstruction.originalOutboxSeq === null)); + assert.ok(output.every((row) => row.value.reconstruction.identityDerivation.kind === "agentrun-session-uuid-to-hwlab-session")); + assert.ok(output.every((row) => row.value.reconstruction.identityDerivation.reversible === true)); +}); + +test("Kafka reader and producer are injected while canonical AgentRun events use strict decode and isolated output", async () => { + const cwd = await mkdtemp(path.join(os.tmpdir(), "hwlab-kafka-debug-canonical-")); + const records = Array.from({ length: 35 }, (_, index) => canonicalRecord(index + 1)); + const readerCalls: any[] = []; + const producerCalls: any[] = []; + let connected = 0; + let disconnected = 0; + const result = await runKafkaCli([ + "regenerate", "hwlab", + "--from", "kafka", + "--session-id", SESSION_ID, + "--trace-id", TRACE_ID, + "--input-topic", "agentrun.event.v1", + "--expect-count", "35", + "--output-jsonl", path.join(cwd, "canonical-output.jsonl"), + "--json" + ], { + cwd, + env: { + HWLAB_KAFKA_BOOTSTRAP_SERVERS: "kafka.test:9092", + HWLAB_KAFKA_HWLAB_DEBUG_EVENT_TOPIC: "hwlab.event.debug.v1", + HWLAB_KAFKA_HWLAB_DEBUG_GROUP_PREFIX: "hwlab-v03-workbench-isolated-debug" + }, + now: () => "2026-07-10T12:00:00.000Z", + async readKafka(input) { + readerCalls.push(input); + return { ok: true, groupId: "hwlab-v03-agentrun-event-debug-regenerate-test", events: records }; + }, + async createProducer() { + return { + async connect() { connected += 1; }, + async send(input: any) { producerCalls.push(input); return [{ topicName: input.topic, partition: 0, baseOffset: "100" }]; }, + async disconnect() { disconnected += 1; } + }; + } + }); + + assert.equal(result.exitCode, 0); + assert.equal(readerCalls.length, 1); + assert.equal(readerCalls[0].groupIdPrefix, "hwlab-v03-workbench-isolated-debug"); + assert.equal(readerCalls[0].sessionId, SESSION_ID); + assert.equal(readerCalls[0].limit, 36, "exact count verification must leave room to detect one extra event"); + assert.equal(connected, 1); + assert.equal(disconnected, 1); + assert.equal(producerCalls.length, 1); + assert.equal(producerCalls[0].topic, "hwlab.event.debug.v1"); + assert.equal(producerCalls[0].messages.length, 35); + const envelopes = producerCalls[0].messages.map((message: any) => JSON.parse(message.value)); + assert.deepEqual(envelopes.map((event: any) => event.debugLineage.sourceSeq), Array.from({ length: 35 }, (_, index) => index + 1)); + assert.deepEqual(envelopes.map((event: any) => event.eventId), Array.from({ length: 35 }, (_, index) => `hwlab:evt_${index + 1}`)); + assert.deepEqual(envelopes.map((event: any) => event.debugLineage.inputOutboxSeq), Array.from({ length: 35 }, (_, index) => index + 1)); + assert.ok(envelopes.every((event: any) => event.diagnostics.inputKind === "canonical-durable")); + assert.ok(envelopes.every((event: any) => event.schema === "hwlab.event.debug.v1")); + assert.equal(result.payload.output.publishedCount, 35); + assert.deepEqual(result.payload.config.outputTopic, { value: "hwlab.event.debug.v1", source: "env:HWLAB_KAFKA_HWLAB_DEBUG_EVENT_TOPIC" }); + assert.deepEqual(result.payload.config.groupPrefix, { value: "hwlab-v03-workbench-isolated-debug", source: "env:HWLAB_KAFKA_HWLAB_DEBUG_GROUP_PREFIX" }); + assert.equal(result.payload.validation.orderPreserved, true); + assert.equal(result.payload.validation.lineagePreserved, true); +}); + +test("reconstruction identity fabrication and product output topic fail closed", async () => { + const cwd = await mkdtemp(path.join(os.tmpdir(), "hwlab-kafka-debug-invalid-")); + const forged = reconstructionRecord(1); + forged.value.eventId = "evt_forged"; + const inputFile = path.join(cwd, "forged.jsonl"); + await writeFile(inputFile, `${JSON.stringify(forged)}\n`, "utf8"); + + const forgedResult = await runKafkaCli([ + "regenerate", "hwlab", "--from", "jsonl", "--session-id", SESSION_ID, + "--jsonl-file", inputFile, "--no-publish" + ], { cwd, env: {}, now: () => "2026-07-10T12:00:00.000Z" }); + assert.equal(forgedResult.exitCode, 1); + assert.equal(forgedResult.payload.error.code, "hwlab_kafka_debug_identity_invalid"); + + const productResult = await runKafkaCli([ + "regenerate", "hwlab", "--from", "jsonl", "--session-id", SESSION_ID, + "--jsonl-file", inputFile, "--output-topic", "hwlab.event.v1", "--no-publish" + ], { cwd, env: {}, now: () => "2026-07-10T12:00:00.000Z" }); + assert.equal(productResult.exitCode, 1); + assert.equal(productResult.payload.error.code, "product_topic_forbidden"); +}); + +test("Kafka query exposes the isolated generated group and original value hash", async () => { + const valueText = JSON.stringify(canonicalRecord(1).value); + const consumerConfigs: any[] = []; + const consumer = { + async connect() {}, + async subscribe() {}, + async run({ eachMessage }: any) { + await eachMessage({ + topic: "agentrun.event.v1", + partition: 0, + message: { offset: "1601", key: Buffer.from(SESSION_ID), value: Buffer.from(valueText), timestamp: "1783681200000" } + }); + }, + 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: 1, + groupIdPrefix: "hwlab-v03-workbench-isolated-debug", + kafkaFactory() { + return { consumer(config: any) { consumerConfigs.push(config); return consumer; } }; + } + }); + 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)); +}); + +test("non-reversible AgentRun session identity stays only in debug lineage", () => { + const rawSessionId = "ses_agentrun_not_uuid"; + const record = reconstructionRecord(1); + record.key = rawSessionId; + record.value.sessionId = rawSessionId; + record.value.hwlabSessionId = null; + record.value.run.sessionId = rawSessionId; + record.value.run.hwlabSessionId = null; + record.value.event.payload.sessionId = rawSessionId; + record.value.reconstruction.identityDerivation = { + kind: "none", + sourceField: "sessionId", + targetField: "hwlabSessionId", + reversible: false, + reason: "source sessionId is not ses_agentrun_", + valuesPrinted: false + }; + const mapped = mapAgentRunRecordsToHwlabDebugEvents([record], { + sessionId: rawSessionId, + traceId: TRACE_ID, + expectedCount: 1, + producedAt: "2026-07-10T12:00:00.000Z" + }); + assert.equal(mapped.events[0].sessionId, null); + assert.equal(mapped.events[0].hwlabSessionId, null); + assert.equal(mapped.events[0].event.sessionId, null); + assert.equal(mapped.events[0].debugLineage.inputSessionId, rawSessionId); + assert.equal(mapped.events[0].reconstruction.identityDerivation.kind, "none"); +}); + +function reconstructionRecord(seq: number): any { + const value = { + schema: "agentrun.event.reconstruction.v1", + eventType: "agentrun.event.reconstructed", + source: "agentrun-cli-stdio-reconstruction", + eventId: null, + outboxSeq: null, + sourceSeq: seq, + observedAt: `2026-07-10T11:00:${String(seq).padStart(2, "0")}.000Z`, + traceId: TRACE_ID, + sessionId: SESSION_ID, + hwlabSessionId: HWLAB_SESSION_ID, + runId: RUN_ID, + commandId: "cmd_9fff03fc124b4203b065d04bbb7c4f1e", + run: { runId: RUN_ID, sessionId: SESSION_ID, hwlabSessionId: HWLAB_SESSION_ID, status: "running", valuesPrinted: false }, + command: { commandId: "cmd_9fff03fc124b4203b065d04bbb7c4f1e", runId: RUN_ID, seq: 1, type: "run", state: "running", valuesPrinted: false }, + event: { + id: null, + runId: RUN_ID, + seq, + type: "backend_status", + payload: { + traceId: TRACE_ID, + sessionId: SESSION_ID, + phase: `reconstructed-${seq}`, + reconstructed: true, + reconstructionKind: "stdio-derived partial reconstruction" + }, + createdAt: `2026-07-10T11:00:${String(seq).padStart(2, "0")}.000Z` + }, + reconstruction: { + kind: "stdio-derived partial reconstruction", + completeLifecycle: false, + originalEventId: null, + originalOutboxSeq: null, + sourceTopic: "codex-stdio.raw.v1", + sourcePartition: 0, + sourceOffset: String(3500 + seq), + sourceFrameSeq: seq, + sourceValueSha256: `sha256:${String(seq).padStart(64, "0")}`, + identityDerivation: { + kind: "agentrun-session-uuid-to-hwlab-session", + sourceField: "sessionId", + targetField: "hwlabSessionId", + reversible: true, + sourcePrefix: "ses_agentrun_", + targetPrefix: "ses_", + separatorTransform: "underscore-to-hyphen", + valuesPrinted: false + } + }, + valuesPrinted: false + }; + return { topic: "agentrun.event.debug.v1", partition: null, offset: null, key: SESSION_ID, value, headers: { "x-trace-id": TRACE_ID } }; +} + +function canonicalRecord(seq: number): any { + const eventId = `evt_${seq}`; + const value = { + schema: "agentrun.event.v1", + eventType: "agentrun.event.committed", + source: "agentrun-manager", + eventId, + outboxSeq: seq, + sourceSeq: seq, + committedAt: `2026-07-10T11:00:${String(seq).padStart(2, "0")}.000Z`, + traceId: TRACE_ID, + sessionId: SESSION_ID, + hwlabSessionId: HWLAB_SESSION_ID, + runId: RUN_ID, + commandId: "cmd_9fff03fc124b4203b065d04bbb7c4f1e", + run: { + runId: RUN_ID, + status: "running", + terminalStatus: null, + failureKind: null, + tenantId: "tenant-test", + projectId: "project-test", + providerId: "gpt.pika", + backendProfile: "gpt.pika", + sessionId: SESSION_ID, + hwlabSessionId: HWLAB_SESSION_ID, + conversationId: null, + threadId: null, + valuesPrinted: false + }, + command: { + commandId: "cmd_9fff03fc124b4203b065d04bbb7c4f1e", + runId: RUN_ID, + seq: 1, + type: "run", + state: "running", + payloadHash: "hash", + valuesPrinted: false + }, + event: { + id: eventId, + runId: RUN_ID, + seq, + type: "backend_status", + payload: { traceId: TRACE_ID, sessionId: SESSION_ID, phase: `canonical-${seq}` }, + createdAt: `2026-07-10T11:00:${String(seq).padStart(2, "0")}.000Z` + }, + valuesPrinted: false + }; + return { topic: "agentrun.event.v1", partition: 0, offset: String(1600 + seq), key: SESSION_ID, value }; +} + +function createSha256(value: string) { + return new Bun.CryptoHasher("sha256").update(value).digest("hex"); +} diff --git a/tools/src/hwlab-cli/kafka-regenerate.ts b/tools/src/hwlab-cli/kafka-regenerate.ts new file mode 100644 index 00000000..bc5b9db4 --- /dev/null +++ b/tools/src/hwlab-cli/kafka-regenerate.ts @@ -0,0 +1,507 @@ +// Responsibility: regenerate isolated HWLAB debug Kafka events from AgentRun Kafka or JSONL input without Cloud API, DB, or projector dependencies. + +import { createHash, randomUUID } from "node:crypto"; +import { mkdir, readFile, writeFile } from "node:fs/promises"; +import path from "node:path"; + +import { Kafka, logLevel } from "kafkajs"; + +import { + buildHwlabKafkaProducerMessage, + DEFAULT_AGENTRUN_DEBUG_EVENT_TOPIC, + DEFAULT_HWLAB_DEBUG_EVENT_TOPIC, + projectAgentRunKafkaMessageToHwlabDebugEvent, + queryKafkaEventStream +} from "../../../internal/cloud/kafka-event-bridge.ts"; + +const CLI_NAME = "hwlab-cli"; +const VERSION = "0.3.0-kafka-debug"; +const DEFAULT_GROUP_PREFIX = "hwlab-v03-workbench-isolated-debug"; +const DEFAULT_LIMIT = 500; +const DEFAULT_TIMEOUT_MS = 5000; + +type JsonRecord = Record; +type EnvLike = Record; +type ParsedArgs = Record & { _: string[] }; + +type KafkaCliOptions = { + cwd?: string; + env?: EnvLike; + now?: () => string; + readKafka?: (input: Record) => Promise>; + createProducer?: (input: { env: EnvLike; clientId: string }) => Promise | any; +}; + +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`); + process.exitCode = result.exitCode; +} + +export async function runKafkaCli(argv: string[], options: KafkaCliOptions = {}) { + const now = options.now ?? (() => new Date().toISOString()); + try { + const parsed = parseOptions(argv); + const command = parsed._[0] || "help"; + const resource = parsed._[1] || ""; + 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 }); + } + const payload = await regenerateHwlabDebugEvents(parsed, { + cwd: options.cwd ?? process.cwd(), + env: options.env ?? process.env, + now, + readKafka: options.readKafka ?? defaultKafkaReader, + createProducer: options.createProducer ?? defaultProducerFactory + }); + return result(0, payload, now); + } catch (error) { + return result(1, failure(error), now); + } +} + +export async function regenerateHwlabDebugEvents(parsed: ParsedArgs, dependencies: Required) { + const sessionId = requiredSessionId(parsed.sessionId); + const traceId = optionalTraceId(parsed.traceId); + const sourceMode = enumValue(parsed.from ?? "kafka", "from", ["kafka", "jsonl"]); + const inputTopicResolution = resolveConfig(parsed.inputTopic, undefined, DEFAULT_AGENTRUN_DEBUG_EVENT_TOPIC, "--input-topic", "contract-default"); + const outputTopicResolution = resolveConfig(parsed.outputTopic, dependencies.env.HWLAB_KAFKA_HWLAB_DEBUG_EVENT_TOPIC, DEFAULT_HWLAB_DEBUG_EVENT_TOPIC, "--output-topic", "env:HWLAB_KAFKA_HWLAB_DEBUG_EVENT_TOPIC"); + const groupPrefixResolution = resolveConfig(parsed.groupPrefix, dependencies.env.HWLAB_KAFKA_HWLAB_DEBUG_GROUP_PREFIX, DEFAULT_GROUP_PREFIX, "--group-prefix", "env:HWLAB_KAFKA_HWLAB_DEBUG_GROUP_PREFIX"); + const inputTopic = inputTopicResolution.value; + const outputTopic = outputTopicResolution.value; + assertDebugTopic(outputTopic); + 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 = expectedCount === undefined ? limit : Math.min(5000, expectedCount + 1); + const groupPrefix = groupPrefixResolution.value; + assertDebugGroup(groupPrefix); + const shouldPublish = parsed.noPublish === true ? false : parsed.publish === true ? true : sourceMode === "kafka"; + + const readResult = sourceMode === "jsonl" + ? await readJsonlInput(requiredText(parsed.jsonlFile, "jsonlFile"), dependencies.cwd, inputTopic) + : await dependencies.readKafka({ + env: dependencies.env, + stream: "agentrun", + topic: inputTopic, + sessionId, + traceId: traceId || null, + limit: readLimit, + timeoutMs, + fromBeginning: parsed.fromEnd !== true, + groupIdPrefix: groupPrefix + }); + const records = Array.isArray(readResult.events) ? readResult.events : []; + const mapped = mapAgentRunRecordsToHwlabDebugEvents(records, { + sessionId, + traceId, + expectedCount, + producedAt: dependencies.now() + }); + + const stateId = `hwlab-debug-${Date.now().toString(36)}-${randomUUID().slice(0, 8)}`; + const artifactPath = path.resolve( + dependencies.cwd, + text(parsed.outputJsonl) || path.join(".state", "hwlab-cli", "kafka-debug", stateId, "hwlab-events.jsonl") + ); + const artifactRows = mapped.events.map((envelope) => { + const message = buildHwlabKafkaProducerMessage(envelope); + return { + topic: outputTopic, + partition: null, + offset: null, + key: message.key, + value: envelope, + headers: message.headers + }; + }); + const artifactText = artifactRows.map((row) => JSON.stringify(row)).join("\n") + "\n"; + await mkdir(path.dirname(artifactPath), { recursive: true }); + await writeFile(artifactPath, artifactText, "utf8"); + + let publishedCount = 0; + let publishMetadata: unknown[] = []; + if (shouldPublish) { + const producer = await dependencies.createProducer({ env: dependencies.env, clientId: `${groupPrefix}-producer` }); + try { + await producer.connect?.(); + publishMetadata = await producer.send({ + topic: outputTopic, + messages: mapped.events.map(buildHwlabKafkaProducerMessage) + }); + publishedCount = mapped.events.length; + } catch (error) { + throw cliError("hwlab_debug_publish_failed", "failed to publish regenerated HWLAB events to the isolated debug topic", { + topic: outputTopic, + mappedCount: mapped.events.length, + artifactPath, + causeCode: (error as any)?.code ?? null, + causeMessage: error instanceof Error ? error.message : String(error) + }); + } finally { + if (typeof producer.disconnect === "function") await producer.disconnect().catch(() => undefined); + } + } + + const first = mapped.events[0]; + const last = mapped.events.at(-1); + const sourceFile = text(readResult.sourceFile); + const traceArgument = traceId ? ` --trace-id ${traceId}` : ""; + const baseCommand = sourceMode === "jsonl" + ? `hwlab-cli kafka regenerate hwlab --from jsonl --session-id ${sessionId}${traceArgument} --jsonl-file ${shellArg(sourceFile || String(parsed.jsonlFile))}` + : `hwlab-cli kafka regenerate hwlab --from kafka --session-id ${sessionId}${traceArgument} --input-topic ${inputTopic}`; + const nextCommand = shouldPublish + ? `${baseCommand} --no-publish --output-topic ${outputTopic} --expect-count ${mapped.events.length} --json` + : `${baseCommand} --publish --output-topic ${outputTopic} --expect-count ${mapped.events.length} --json`; + return { + ok: true, + action: "kafka.regenerate.hwlab", + status: "succeeded", + sourceMode, + config: { + inputTopic: inputTopicResolution, + outputTopic: outputTopicResolution, + groupPrefix: groupPrefixResolution + }, + input: { + topic: inputTopic, + groupId: readResult.groupId ?? null, + readCount: records.length, + matchedCount: mapped.events.length, + sessionId, + traceId: traceId || null + }, + output: { + schema: DEFAULT_HWLAB_DEBUG_EVENT_TOPIC, + topic: outputTopic, + count: mapped.events.length, + published: shouldPublish, + publishedCount, + metadataCount: Array.isArray(publishMetadata) ? publishMetadata.length : 0, + artifact: { + path: artifactPath, + bytes: Buffer.byteLength(artifactText), + sha256: `sha256:${sha256(artifactText)}` + } + }, + validation: { + expectedCount: expectedCount ?? null, + oneToOne: mapped.inputCount === mapped.events.length, + orderPreserved: mapped.orderPreserved, + lineagePreserved: mapped.lineagePreserved, + firstSourceSeq: first?.debugLineage?.sourceSeq ?? null, + lastSourceSeq: last?.debugLineage?.sourceSeq ?? null, + inputKinds: [...new Set(mapped.inputKinds)] + }, + runtimeDependencies: { + cloudApi: false, + database: false, + transactionalProjector: false, + kafka: sourceMode === "kafka" || shouldPublish + }, + next: { + command: nextCommand, + reason: shouldPublish + ? "用同一源码映射做一次无 Kafka 写入复验；随后 Workbench 调试模式按 traceId 消费独立 debug topic。" + : "离线映射已固化为 JSONL；确认内容后可显式 --publish 写入独立 debug topic。" + }, + valuesPrinted: false + }; +} + +export function mapAgentRunRecordsToHwlabDebugEvents(records: JsonRecord[], options: { sessionId: string; traceId?: string; expectedCount?: number; producedAt?: string }) { + const matchingRecords = records.filter((record) => recordMatches(record, options.sessionId, options.traceId)); + if (matchingRecords.length === 0) { + throw cliError("agentrun_debug_events_missing", "no AgentRun events matched the requested sessionId/traceId", { + sessionId: options.sessionId, + traceId: options.traceId || null, + readCount: records.length + }); + } + if (options.expectedCount !== undefined && matchingRecords.length !== options.expectedCount) { + throw cliError("agentrun_debug_event_count_mismatch", "matched AgentRun event count does not equal --expect-count", { + expectedCount: options.expectedCount, + actualCount: matchingRecords.length + }); + } + const events = []; + const inputKinds = []; + let orderPreserved = true; + let lineagePreserved = true; + for (const [index, record] of matchingRecords.entries()) { + const kafkaMessage = kafkaMessageFromRecord(record, index); + const projected = projectAgentRunKafkaMessageToHwlabDebugEvent(kafkaMessage, { producedAt: options.producedAt }); + if (!projected.ok) { + throw cliError(projected.error?.code || "hwlab_debug_projection_failed", projected.error?.message || "AgentRun debug projection failed", { + index, + topic: kafkaMessage.topic, + partition: kafkaMessage.partition, + offset: kafkaMessage.message.offset + }); + } + const input = valueFromRecord(record); + const sourceSeq = Number(input.sourceSeq ?? input.event?.seq); + orderPreserved &&= projected.envelope.debugLineage?.sourceSeq === sourceSeq; + lineagePreserved &&= projected.envelope.debugLineage?.inputTraceId === text(input.traceId ?? input.event?.payload?.traceId) + && sessionCandidates(input).includes(options.sessionId) + && projected.envelope.sourceEvent?.sha256 === kafkaMessage.inputSha256; + events.push(projected.envelope); + inputKinds.push(projected.inputKind); + } + if (events.length !== matchingRecords.length || !orderPreserved || !lineagePreserved) { + throw cliError("hwlab_debug_mapping_invariant_failed", "AgentRun -> HWLAB debug mapping did not preserve one-to-one order and lineage", { + inputCount: matchingRecords.length, + outputCount: events.length, + orderPreserved, + lineagePreserved + }); + } + return { events, inputKinds, inputCount: matchingRecords.length, orderPreserved, lineagePreserved }; +} + +function kafkaHelp() { + return { + ok: true, + action: "kafka.help", + status: "succeeded", + commands: [ + "hwlab-cli kafka regenerate hwlab --from kafka --session-id ses_... [--trace-id trc_...] [--input-topic agentrun.event.debug.v1] [--expect-count 35] [--json]", + "hwlab-cli kafka regenerate hwlab --from jsonl --session-id ses_... --jsonl-file agentrun-events.jsonl --no-publish [--output-jsonl hwlab-events.jsonl] [--json]" + ], + defaults: { + inputTopic: DEFAULT_AGENTRUN_DEBUG_EVENT_TOPIC, + outputTopic: DEFAULT_HWLAB_DEBUG_EVENT_TOPIC, + groupPrefix: DEFAULT_GROUP_PREFIX, + kafkaInputPublishes: true, + jsonlInputPublishes: false + }, + boundaries: { + productTopicAllowed: false, + cloudApi: false, + database: false, + transactionalProjector: false, + reconstructionIdentityFabricated: false + }, + next: { + command: "hwlab-cli kafka regenerate hwlab --from jsonl --session-id ses_... --jsonl-file agentrun-events.jsonl --no-publish --json", + reason: "先用完全离线 JSONL 验证一对一顺序映射，再决定是否发布到独立 debug topic。" + }, + valuesPrinted: false + }; +} + +async function defaultKafkaReader(input: Record) { + return queryKafkaEventStream({ + ...input, + kafkaFactory(config: { brokers: string[]; clientId: string }) { return boundedKafka(config); } + }); +} + +async function defaultProducerFactory({ env, clientId }: { env: EnvLike; clientId: string }) { + const brokers = csv(env.HWLAB_KAFKA_BOOTSTRAP_SERVERS); + if (brokers.length === 0) throw cliError("kafka_brokers_missing", "HWLAB_KAFKA_BOOTSTRAP_SERVERS is required for Kafka publish", { field: "HWLAB_KAFKA_BOOTSTRAP_SERVERS" }); + return boundedKafka({ clientId, brokers }).producer({ allowAutoTopicCreation: false }); +} + +function boundedKafka({ clientId, brokers }: { clientId: string; brokers: string[] }) { + return new Kafka({ + clientId, + brokers, + logLevel: logLevel.NOTHING, + connectionTimeout: 5000, + requestTimeout: 10000, + retry: { retries: 1, maxRetryTime: 10000 } + }); +} + +async function readJsonlInput(fileValue: string, cwd: string, defaultTopic: string) { + const file = path.resolve(cwd, fileValue); + const content = await readFile(file, "utf8"); + const events = []; + for (const [index, line] of content.split(/\r?\n/u).entries()) { + if (!line.trim()) continue; + let value; + try { value = JSON.parse(line); } catch (error) { + throw cliError("jsonl_line_invalid", "input JSONL contains invalid JSON", { file, line: index + 1, message: error instanceof Error ? error.message : String(error) }); + } + events.push(normalizeRecord(value, defaultTopic)); + } + return { ok: true, topic: defaultTopic, groupId: null, events, sourceFile: file }; +} + +function normalizeRecord(record: any, defaultTopic: string) { + if (!record || typeof record !== "object" || Array.isArray(record)) throw cliError("jsonl_record_invalid", "each JSONL row must be an object"); + if (Object.hasOwn(record, "value")) return { ...record, topic: text(record.topic) || defaultTopic }; + return { topic: defaultTopic, partition: null, offset: null, key: null, value: record }; +} + +function kafkaMessageFromRecord(record: JsonRecord, index: number) { + const value = valueFromRecord(record); + const valueText = typeof record.value === "string" ? record.value : JSON.stringify(value); + return { + topic: text(record.topic) || DEFAULT_AGENTRUN_DEBUG_EVENT_TOPIC, + partition: nullableInteger(record.partition), + inputSha256: text(record.valueSha256) || sha256(valueText), + message: { + offset: record.offset === null || record.offset === undefined ? null : String(record.offset), + key: record.key === null || record.key === undefined ? null : Buffer.from(String(record.key)), + value: Buffer.from(valueText), + timestamp: record.timestamp === null || record.timestamp === undefined ? null : String(record.timestamp), + headers: record.headers ?? undefined, + debugIndex: index + } + }; +} + +function valueFromRecord(record: JsonRecord) { + const candidate = Object.hasOwn(record, "value") ? record.value : record; + if (typeof candidate === "string") { + try { return JSON.parse(candidate); } catch { throw cliError("jsonl_value_invalid", "JSONL transport value must contain a JSON object"); } + } + if (!candidate || typeof candidate !== "object" || Array.isArray(candidate)) throw cliError("jsonl_value_invalid", "JSONL transport value must be a JSON object"); + return candidate; +} + +function recordMatches(record: JsonRecord, sessionId: string, traceId?: string) { + const value = valueFromRecord(record); + if (!sessionCandidates(value).includes(sessionId)) return false; + return !traceId || traceCandidates(value).includes(traceId); +} + +function sessionCandidates(value: any) { + return uniqueText([ + value.sessionId, + value.hwlabSessionId, + value.run?.sessionId, + value.run?.hwlabSessionId, + value.event?.payload?.sessionId, + value.event?.payload?.hwlabSessionId + ]); +} + +function traceCandidates(value: any) { + return uniqueText([value.traceId, value.run?.traceId, value.command?.traceId, value.event?.payload?.traceId, value.event?.payload?.hwlabTraceId]); +} + +function parseOptions(argv: string[]): ParsedArgs { + const out: ParsedArgs = { _: [] }; + for (let index = 0; index < argv.length; index += 1) { + const item = argv[index] ?? ""; + if (!item.startsWith("--")) { out._.push(item); continue; } + const eq = item.indexOf("="); + const rawKey = eq >= 0 ? item.slice(2, eq) : item.slice(2); + const key = rawKey.replace(/-([a-z])/gu, (_, character) => String(character).toUpperCase()); + if (eq >= 0) { out[key] = item.slice(eq + 1); continue; } + const next = argv[index + 1]; + if (next && !next.startsWith("--")) { out[key] = next; index += 1; } + else out[key] = true; + } + return out; +} + +function renderKafkaCliText(payload: Record) { + if (payload.ok === false) return `failed action=${payload.action} code=${payload.error?.code} message=${JSON.stringify(payload.error?.message)} next=${JSON.stringify(payload.next?.command)}`; + const validation = payload.validation ?? {}; + return [ + "ok", + `action=${payload.action}`, + payload.input ? `input=${payload.input.matchedCount}/${payload.input.readCount}` : null, + payload.output ? `output=${payload.output.count}` : null, + payload.output ? `published=${payload.output.publishedCount}` : null, + validation.orderPreserved !== undefined ? `order=${validation.orderPreserved ? "preserved" : "failed"}` : null, + payload.output?.topic ? `topic=${payload.output.topic}` : null, + payload.config?.outputTopic?.source ? `topicSource=${payload.config.outputTopic.source}` : null, + payload.config?.groupPrefix?.source ? `groupSource=${payload.config.groupPrefix.source}` : null, + payload.output?.artifact?.path ? `artifact=${shellArg(payload.output.artifact.path)}` : null, + payload.next?.command ? `next=${JSON.stringify(payload.next.command)}` : null + ].filter(Boolean).join(" "); +} + +function result(exitCode: number, payload: Record, now: () => string) { + return { exitCode, payload: { generatedAt: now(), cli: CLI_NAME, version: VERSION, ...payload } }; +} + +function failure(error: any) { + return { + ok: false, + action: "kafka.regenerate.hwlab", + 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 注入后重试。" }, + valuesPrinted: false + }; +} + +function assertDebugTopic(topic: string) { + if (topic === "hwlab.event.v1" || !/(?:^|[.-])debug(?:[.-]|$)/u.test(topic)) { + throw cliError("product_topic_forbidden", "output topic must be an isolated debug topic; hwlab.event.v1 is forbidden", { topic }); + } +} + +function assertDebugGroup(groupPrefix: string) { + if (!/(?:^|[.-])debug(?:[.-]|$)/u.test(groupPrefix)) throw cliError("debug_group_required", "Kafka consumer group prefix must identify an isolated debug group", { groupPrefix }); +} + +function requiredSessionId(value: unknown) { + const result = requiredText(value, "sessionId"); + if (!/^ses_[A-Za-z0-9_.:-]+$/u.test(result)) throw cliError("invalid_session_id", "sessionId must start with ses_", { sessionId: result }); + 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 }); + return result; +} + +function enumValue(value: unknown, field: string, allowed: string[]) { + const result = text(value); + if (!allowed.includes(result)) throw cliError("invalid_option", `${field} must be one of: ${allowed.join(", ")}`, { field, value: result }); + return result; +} + +function resolveConfig(cliValue: unknown, envValue: unknown, fallback: string, cliSource: string, envSource: string) { + const explicit = text(cliValue); + if (explicit) return { value: explicit, source: cliSource }; + const injected = text(envValue); + if (injected) return { value: injected, source: envSource }; + return { value: fallback, source: "contract-default" }; +} + +function optionalPositiveInteger(value: unknown, field: string) { + if (value === undefined) return undefined; + return boundedInteger(value, 0, 1, 5000, field); +} + +function boundedInteger(value: unknown, fallback: number, minimum: number, maximum: number, field: string) { + const parsed = value === undefined ? fallback : Number(value); + if (!Number.isInteger(parsed) || parsed < minimum || parsed > maximum) throw cliError("invalid_integer", `${field} must be an integer between ${minimum} and ${maximum}`, { field, value }); + return parsed; +} + +function nullableInteger(value: unknown) { + if (value === null || value === undefined || value === "") return null; + const parsed = Number(value); + if (!Number.isInteger(parsed) || parsed < 0) throw cliError("invalid_partition", "partition must be a non-negative integer or null", { value }); + return parsed; +} + +function requiredText(value: unknown, field: string) { + const result = text(value); + if (!result) throw cliError("missing_required_value", `${field} is required`, { field }); + return result; +} + +function cliError(code: string, message: string, details: Record = {}) { + return Object.assign(new Error(message), { code, details }); +} + +function uniqueText(values: unknown[]) { return [...new Set(values.map(text).filter(Boolean))]; } +function text(value: unknown) { return String(value ?? "").trim(); } +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, `'"'"'`)}'`; }