feat: 实现纯 Kafka 页面刷新重放交接

2026-07-11 05:15:35 +02:00
parent 0714304b20
commit 2cb3d3aaa6
34 changed files with 2547 additions and 97 deletions
@@ -28,12 +28,13 @@ test("offline HWLAB JSONL uses the shared Web pipeline and renders final respons
  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" })
+    hwlabRecord(1, { type: "user", eventType: "user", status: "sent", label: "agentrun:user:message", userMessageId: "msg_render_trace_user", messageId: "msg_render_trace_user", targetTraceId: "trc_render_trace_target", text: "hi" }),
+    hwlabRecord(2, { 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(3, { 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(4, { 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(5, { type: "assistant", eventType: "assistant", status: "running", label: "agentrun:assistant:message", itemId: "assistant_final", message: finalText, assistantText: finalText, assistantSource: "completed-agent-message" }),
+    hwlabRecord(6, { 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(7, { 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");

@@ -50,12 +51,16 @@ test("offline HWLAB JSONL uses the shared Web pipeline and renders final respons
  ], { 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.input.scannedCount, 7);
+  assert.equal(result.payload.input.matchedCount, 7);
+  assert.equal(result.payload.projection.decodedCount, 7);
+  assert.equal(result.payload.projection.appliedCount, 7);
+  assert.equal(result.payload.projection.eventCount, 6, "formal user input must not create an agent trace event");
+  assert.equal(result.payload.projection.userMessageCount, 1);
+  assert.deepEqual(result.payload.projection.userMessageIds, ["msg_render_trace_user"]);
+  assert.deepEqual(result.payload.projection.userTargetTraceIds, ["trc_render_trace_target"]);
  assert.equal(result.payload.projection.terminalObserved, true);
-  assert.equal(result.payload.projection.lastEventAt, "2026-07-10T12:00:00.005Z");
+  assert.equal(result.payload.projection.lastEventAt, "2026-07-10T12:00:00.006Z");
  assert.deepEqual(result.payload.identity.observedRunIds, [RUN_ID]);
  assert.deepEqual(result.payload.identity.observedCommandIds, ["cmd_render_trace"]);
  assert.equal(result.payload.identity.sourceEventIdentityDuplicates, 0);
@@ -73,6 +78,9 @@ test("offline HWLAB JSONL uses the shared Web pipeline and renders final respons
  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.equal((markdown.match(/^hi$/gmu) ?? []).length, 1);
+  assert.match(markdown, /## 用户输入/u);
+  assert.match(markdown, /target trace: trc_render_trace_target/u);
  assert.match(markdown, /## 运行记录/u);
  assert.match(markdown, /## 最终回复/u);
  assert.equal(result.markdownOutput?.trim(), markdown.trim());
@@ -215,14 +223,13 @@ test("command-scoped trace keeps attributable commandless final and terminal lif
    now: () => "2026-07-10T12:00:00.000Z",
    async readKafka(input) {
      readerInput = input;
+      const events = [exactTool, exactProgress, unscopedFinal, unscopedTerminal, runnerShutdown, unscopedBackend];
      return {
-        events: [exactTool, exactProgress, unscopedFinal, unscopedTerminal, runnerShutdown, unscopedBackend],
+        ...completeKafkaRead(events),
        scannedCount: 6,
        parsedCount: 6,
        matchedCount: 6,
-        filterRejectedCount: 0,
-        completionReason: "end-offset",
-        reachedEndOffsets: true
+        filterRejectedCount: 0
      };
    }
  });
@@ -232,6 +239,10 @@ test("command-scoped trace keeps attributable commandless final and terminal lif
  assert.equal(readerInput.runId, RUN_ID);
  assert.equal(result.payload.input.queryMatchedCount, 6);
  assert.equal(result.payload.input.matchedCount, 4);
+  assert.equal(result.payload.input.refreshHandoff.status, "succeeded");
+  assert.equal(result.payload.input.refreshHandoff.liveSubscriptionInstalled, true);
+  assert.equal(result.payload.input.refreshHandoff.counts.replayed, 6);
+  assert.deepEqual(result.payload.input.refreshHandoff.barrier, [{ partition: 0, endOffset: "1907" }]);
  assert.deepEqual(result.payload.input.commandScope.commandIdCounts, [
    { commandId: "cmd_render_trace", count: 2 },
    { commandId: "runner-shutdown", count: 1 }
@@ -243,6 +254,7 @@ test("command-scoped trace keeps attributable commandless final and terminal lif
  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.projection.pipeline[0], "createWorkbenchKafkaRefreshHandoff");
  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);
@@ -288,14 +300,13 @@ test("missing commandId cannot be satisfied by commandless lifecycle events from
    env: {},
    now: () => "2026-07-10T12:00:00.000Z",
    async readKafka() {
+      const events = [otherCommand, commandlessFinal, commandlessTerminal];
      return {
-        events: [otherCommand, commandlessFinal, commandlessTerminal],
+        ...completeKafkaRead(events),
        scannedCount: 3,
        parsedCount: 3,
        matchedCount: 3,
-        filterRejectedCount: 0,
-        completionReason: "end-offset",
-        reachedEndOffsets: true
+        filterRejectedCount: 0
      };
    }
  });
@@ -340,6 +351,7 @@ test("Kafka Trace render returns a typed partial result when the captured end of
        reachedEndOffsets: false,
        endOffsetsAvailable: true,
        endOffsets: [{ partition: 0, startOffset: "0", endOffset: "2" }],
+        completion: { reason: "timeout", complete: false, barrierReached: false, retentionStartVerified: true },
        lastScannedOffsets: [{ partition: 0, offset: "0" }]
      };
    }
@@ -351,6 +363,8 @@ test("Kafka Trace render returns a typed partial result when the captured end of
  assert.equal(result.payload.error.code, "source_scan_incomplete");
  assert.equal(result.payload.input.scanComplete, false);
  assert.equal(result.payload.input.completionReason, "timeout");
+  assert.equal(result.payload.input.refreshHandoff.status, "failed");
+  assert.equal(result.payload.input.refreshHandoff.error.code, "workbench_kafka_refresh_timeout");
  assert.equal(result.payload.validation.sourceScanComplete, false);
  assert.equal(result.payload.validation.commandLifecyclePreserved, false);
  assert.match(result.markdownOutput ?? "", /source scan: partial \(timeout\)/u);
@@ -358,6 +372,50 @@ test("Kafka Trace render returns a typed partial result when the captured end of
  assert.match(renderKafkaCliText(result.payload), /^partial action=kafka\.render\.trace code=source_scan_incomplete completion=timeout reachedEnd=false/u);
 });

+test("incomplete zero-match Kafka scan stays partial and never claims the trace is missing", async () => {
+  const cwd = await mkdtemp(path.join(os.tmpdir(), "hwlab-kafka-trace-empty-partial-"));
+  const result = await runKafkaCli([
+    "render", "trace",
+    "--from", "kafka",
+    "--trace-id", TRACE_ID,
+    "--group-prefix", "hwlab-v03-workbench-isolated-debug",
+    "--format", "markdown"
+  ], {
+    cwd,
+    env: {},
+    now: () => "2026-07-10T12:00:00.000Z",
+    async readKafka() {
+      return {
+        events: [],
+        scannedCount: 0,
+        parsedCount: 0,
+        matchedCount: 0,
+        filterRejectedCount: 0,
+        completionReason: "timeout",
+        reachedEndOffsets: false,
+        endOffsetsAvailable: true,
+        endOffsets: [{ partition: 0, startOffset: "0", endOffset: "12" }],
+        completion: { reason: "timeout", complete: false, barrierReached: false, retentionStartVerified: true },
+        lastScannedOffsets: []
+      };
+    }
+  });
+
+  assert.equal(result.exitCode, 2);
+  assert.equal(result.payload.status, "partial");
+  assert.equal(result.payload.error.code, "source_scan_incomplete");
+  assert.notEqual(result.payload.error.code, "source_trace_missing");
+  assert.equal(result.payload.input.scannedCount, 0);
+  assert.equal(result.payload.input.matchedCount, 0);
+  assert.equal(result.payload.input.scanComplete, false);
+  assert.equal(result.payload.input.refreshHandoff.status, "failed");
+  assert.equal(result.payload.input.refreshHandoff.error.code, "workbench_kafka_refresh_timeout");
+  assert.equal(result.payload.validation.refreshHandoffComplete, false);
+  assert.match(result.markdownOutput ?? "", /source scan: partial \(timeout\)/u);
+  assert.equal(typeof result.payload.output.artifact.path, "string");
+  assert.match(renderKafkaCliText(result.payload), /^partial action=kafka\.render\.trace code=source_scan_incomplete completion=timeout reachedEnd=false scanned=0 matched=0 applied=0/u);
+});
+
 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");
@@ -751,9 +809,46 @@ test("Kafka query reports invalid JSON, filter rejection, and bounded end-offset
  assert.equal(queried.completionReason, "end-offset");
  assert.equal(queried.reachedEndOffsets, true);
  assert.deepEqual(queried.endOffsets, [{ partition: 0, startOffset: "0", endOffset: "3" }]);
+  assert.deepEqual(queried.firstScannedOffsets, [{ partition: 0, offset: "0" }]);
  assert.deepEqual(queried.lastScannedOffsets, [{ partition: 0, offset: "2" }]);
 });

+test("Kafka query fails closed when retention advances between barrier capture and consumer assignment", async () => {
+  const matching = canonicalRecord(1).value;
+  const queried = await queryKafkaEventStream({
+    env: { HWLAB_KAFKA_BOOTSTRAP_SERVERS: "kafka.test:9092", HWLAB_KAFKA_CLIENT_ID: "hwlab-test" },
+    topic: "hwlab.event.v1",
+    sessionId: SESSION_ID,
+    limit: 10,
+    scanLimit: 10,
+    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 {
+            async connect() {},
+            async subscribe() {},
+            async run({ eachMessage }: any) {
+              await eachMessage({ topic: "hwlab.event.v1", partition: 0, message: { offset: "2", value: Buffer.from(JSON.stringify(matching)) } });
+            },
+            async stop() {},
+            async disconnect() {}
+          };
+        }
+      };
+    }
+  });
+
+  assert.equal(queried.completionReason, "retention-start-moved");
+  assert.equal(queried.completion.complete, false);
+  assert.equal(queried.completion.retentionStartVerified, false);
+  assert.equal(queried.completion.retentionStartMoved, true);
+  assert.equal(queried.reachedEndOffsets, false);
+  assert.equal(queried.scannedCount, 0);
+  assert.deepEqual(queried.firstScannedOffsets, [{ partition: 0, offset: "2" }]);
+});
+
 test("Kafka query prefers end-offset when the final record also reaches the event limit", async () => {
  const matching = canonicalRecord(1).value;
  const queried = await queryKafkaEventStream({
@@ -817,6 +912,107 @@ test("Kafka query treats a retained empty partition as a completed bounded scan"
  assert.deepEqual(queried.endOffsets, [{ partition: 0, startOffset: "7", endOffset: "7" }]);
 });

+test("Kafka query stops every partition at the captured exclusive barrier and excludes later records", 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 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: 10,
+    scanLimit: 10,
+    groupIdPrefix: "hwlab-v03-workbench-isolated-debug",
+    kafkaFactory() {
+      return {
+        admin() { return { async connect() {}, async fetchTopicOffsets() { return [{ partition: 0, low: "0", high: "2" }, { partition: 1, low: "0", high: "2" }]; }, async disconnect() {} }; },
+        consumer() {
+          return {
+            async connect() {},
+            async subscribe() {},
+            async run({ eachMessage }: any) {
+              await eachMessage({ topic: "agentrun.event.v1", partition: 0, message: { offset: "0", value: Buffer.from(JSON.stringify(matching)) } });
+              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)) } });
+              await eachMessage({ topic: "agentrun.event.v1", partition: 1, message: { offset: "0", value: Buffer.from(JSON.stringify(mismatch)) } });
+              await eachMessage({ topic: "agentrun.event.v1", partition: 1, message: { offset: "1", value: Buffer.from(JSON.stringify(matching)) } });
+              await eachMessage({ topic: "agentrun.event.v1", partition: 1, message: { offset: "3", value: Buffer.from(JSON.stringify(matching)) } });
+            },
+            async stop() {},
+            async disconnect() {}
+          };
+        }
+      };
+    }
+  });
+
+  assert.equal(queried.completionReason, "end-offset");
+  assert.equal(queried.completion.complete, true);
+  assert.equal(queried.completion.barrierReached, true);
+  assert.equal(queried.scannedCount, 4);
+  assert.equal(queried.excludedPostBarrierCount, 1);
+  assert.deepEqual(queried.events.map((event: any) => [event.partition, event.offset]), [[0, "0"], [1, "1"]]);
+  assert.deepEqual(queried.lastScannedOffsets, [{ partition: 0, offset: "1" }, { partition: 1, offset: "1" }]);
+});
+
+test("Kafka query reports scan-limit and abort as typed incomplete completion", async () => {
+  const matching = canonicalRecord(1).value;
+  const limited = await queryKafkaEventStream({
+    env: { HWLAB_KAFKA_BOOTSTRAP_SERVERS: "kafka.test:9092", HWLAB_KAFKA_CLIENT_ID: "hwlab-test" },
+    topic: "agentrun.event.v1",
+    sessionId: SESSION_ID,
+    limit: 10,
+    scanLimit: 1,
+    timeoutMs: 500,
+    groupIdPrefix: "hwlab-v03-workbench-isolated-debug",
+    kafkaFactory() {
+      return {
+        admin() { return { async connect() {}, async fetchTopicOffsets() { return [{ partition: 0, low: "0", high: "5" }]; }, async disconnect() {} }; },
+        consumer() { return { async connect() {}, async subscribe() {}, async run({ eachMessage }: any) { await eachMessage({ topic: "agentrun.event.v1", partition: 0, message: { offset: "0", value: Buffer.from(JSON.stringify(matching)) } }); }, async stop() {}, async disconnect() {} }; }
+      };
+    }
+  });
+  assert.deepEqual(limited.completion, {
+    reason: "scan-limit",
+    complete: false,
+    barrierReached: false,
+    retentionStartVerified: true,
+    retentionStartMoved: false,
+    timeout: false,
+    matchedEventLimitReached: false,
+    scanLimitReached: true,
+    aborted: false,
+    valuesRedacted: true
+  });
+
+  const controller = new AbortController();
+  let disconnected = false;
+  const abortedPromise = queryKafkaEventStream({
+    env: { HWLAB_KAFKA_BOOTSTRAP_SERVERS: "kafka.test:9092", HWLAB_KAFKA_CLIENT_ID: "hwlab-test" },
+    topic: "agentrun.event.v1",
+    sessionId: SESSION_ID,
+    limit: 10,
+    scanLimit: 10,
+    timeoutMs: 500,
+    signal: controller.signal,
+    groupIdPrefix: "hwlab-v03-workbench-isolated-debug",
+    kafkaFactory() {
+      return {
+        admin() { return { async connect() {}, async fetchTopicOffsets() { return [{ partition: 0, low: "0", high: "5" }]; }, async disconnect() {} }; },
+        consumer() { return { async connect() {}, async subscribe() {}, async run() { await new Promise(() => undefined); }, async stop() {}, async disconnect() { disconnected = true; } }; }
+      };
+    }
+  });
+  setTimeout(() => controller.abort(), 5);
+  const aborted = await abortedPromise;
+  assert.equal(aborted.completionReason, "aborted");
+  assert.equal(aborted.completion.aborted, true);
+  assert.equal(aborted.completion.complete, false);
+  assert.equal(disconnected, true);
+});
+
 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" },
@@ -932,6 +1128,31 @@ test("non-reversible AgentRun session identity stays only in debug lineage", ()
  assert.equal(mapped.events[0].reconstruction.identityDerivation.kind, "none");
 });

+function completeKafkaRead(events: any[]) {
+  const byPartition = new Map<number, bigint[]>();
+  for (const event of events) {
+    const partition = Number(event.partition);
+    const offset = BigInt(String(event.offset));
+    const offsets = byPartition.get(partition) ?? [];
+    offsets.push(offset);
+    byPartition.set(partition, offsets);
+  }
+  const endOffsets = [...byPartition.entries()].map(([partition, offsets]) => ({
+    partition,
+    startOffset: String(offsets.reduce((left, right) => left < right ? left : right)),
+    endOffset: String(offsets.reduce((left, right) => left > right ? left : right) + 1n)
+  })).sort((left, right) => left.partition - right.partition);
+  return {
+    topic: events[0]?.topic ?? "hwlab.event.v1",
+    events,
+    completionReason: "end-offset",
+    reachedEndOffsets: true,
+    endOffsetsAvailable: true,
+    endOffsets,
+    completion: { reason: "end-offset", complete: true, barrierReached: true, retentionStartVerified: true }
+  };
+}
+
 function reconstructionRecord(seq: number): any {
  const value = {
    schema: "agentrun.event.reconstruction.v1",