Merge pull request #262 from pikasTech/fix/kafka-trace-context

补齐 AgentRun Kafka 事件 trace 上下文

src/mgr/kafka-shadow-producer.ts

@@ -1,6 +1,7 @@
 import type { AgentRunStore } from "./store.js";
 import type { CommandRecord, JsonRecord, JsonValue, RunEvent, RunRecord } from "../common/types.js";
 import { agentRunKafkaConfig, jsonBytes, publishAgentRunKafkaMessageBestEffort, safeText, sha256Hex, stableKafkaJson, type AgentRunKafkaConfig } from "../common/kafka-events.js";
+import { agentRunBusinessTraceId } from "../common/otel-trace.js";
 
 const TRUE_VALUES = new Set(["1", "true", "yes", "on"]);
 const warnedCodes = new Set<string>();
@@ -86,14 +87,18 @@ async function resolveRun(input: ShadowEventInput): Promise<RunRecord | null> {
   return await input.store.getRun(input.runId);
 }
 
-function buildAgentRunShadowMessage(input: ShadowEventInput & { run: RunRecord | null; config: AgentRunKafkaConfig }): { topic: string; key: string; value: JsonRecord; headers: Record<string, string> } {
+export function buildAgentRunShadowMessage(input: ShadowEventInput & { run: RunRecord | null; config: AgentRunKafkaConfig }): { topic: string; key: string; value: JsonRecord; headers: Record<string, string> } {
   const payload = input.eventPayload ?? eventPayload(input.event);
+  const traceId = agentRunBusinessTraceId(input.run, input.command);
+  const sessionId = safeText(input.run?.sessionRef?.sessionId);
   const event = {
     schema: "agentrun.event.v1",
     eventType: input.eventType,
     source: input.config.clientId,
     producedAt: new Date().toISOString(),
     mode: "shadow-produce-only",
+    traceId,
+    sessionId,
     run: runSummary(input.run),
     command: commandSummary(input.command),
     event: runEventSummary(input.event, payload),

src/selftest/cases/35-kafka-shadow-producer.ts

@@ -0,0 +1,71 @@
+import assert from "node:assert/strict";
+import { buildAgentRunShadowMessage } from "../../mgr/kafka-shadow-producer.js";
+import { MemoryAgentRunStore } from "../../mgr/store.js";
+import type { AgentRunKafkaConfig } from "../../common/kafka-events.js";
+import type { JsonRecord } from "../../common/types.js";
+import type { SelfTestCase } from "../harness.js";
+
+const config: AgentRunKafkaConfig = {
+  brokers: ["127.0.0.1:9092"],
+  clientId: "agentrun-selftest",
+  agentrunEventTopic: "agentrun.event.v1",
+  codexStdioTopic: "codex-stdio.raw.v1",
+  enabled: true,
+  valuesPrinted: false,
+};
+
+const selfTest: SelfTestCase = async () => {
+  const store = new MemoryAgentRunStore();
+  const runTraceId = "trc_run_context";
+  const commandTraceId = "trc_command_payload";
+  const agentRunSessionId = "ses_agentrun_selftest";
+  const run = store.createRun({
+    tenantId: "hwlab",
+    projectId: "pikasTech/HWLAB",
+    workspaceRef: { kind: "opaque", path: ".", hwlabTraceId: "trc_workspace_context" },
+    sessionRef: { sessionId: agentRunSessionId, metadata: { hwlabTraceId: "trc_session_context" } },
+    resourceBundleRef: null,
+    providerId: "G14",
+    backendProfile: "codex",
+    executionPolicy: {
+      sandbox: "workspace-write",
+      approval: "never",
+      timeoutMs: 1000,
+      network: "restricted",
+      secretScope: { providerCredentials: [], toolCredentials: [] },
+    },
+    traceSink: { kind: "hwlab", traceId: runTraceId },
+  });
+  const runMessage = buildAgentRunShadowMessage({ env: {}, eventType: "agentrun.run.created", run, config });
+  const runValue = runMessage.value as JsonRecord;
+  assert.equal(runValue.traceId, runTraceId);
+  assert.equal(runValue.sessionId, agentRunSessionId);
+  assert.equal((runValue.run as JsonRecord).sessionId, agentRunSessionId);
+
+  const commandOnlyTraceRun = store.createRun({
+    tenantId: "hwlab",
+    projectId: "pikasTech/HWLAB",
+    workspaceRef: { kind: "opaque", path: "." },
+    sessionRef: { sessionId: "ses_agentrun_command_trace" },
+    resourceBundleRef: null,
+    providerId: "G14",
+    backendProfile: "codex",
+    executionPolicy: {
+      sandbox: "workspace-write",
+      approval: "never",
+      timeoutMs: 1000,
+      network: "restricted",
+      secretScope: { providerCredentials: [], toolCredentials: [] },
+    },
+    traceSink: null,
+  });
+  const command = store.createCommand(commandOnlyTraceRun.id, { type: "turn", payload: { traceId: commandTraceId, prompt: "redacted" } });
+  const commandMessage = buildAgentRunShadowMessage({ env: {}, eventType: "agentrun.command.created", run: commandOnlyTraceRun, command, config });
+  const commandValue = commandMessage.value as JsonRecord;
+  assert.equal(commandValue.traceId, commandTraceId);
+  assert.equal(commandValue.sessionId, "ses_agentrun_command_trace");
+  assert.equal((commandValue.command as JsonRecord).commandId, command.id);
+  return { name: "kafka-shadow-producer", tests: ["kafka-shadow-context-trace-id"] };
+};
+
+export default selfTest;