diff --git a/src/mgr/kafka-shadow-producer.ts b/src/mgr/kafka-shadow-producer.ts index 45cc541..11ddf1b 100644 --- a/src/mgr/kafka-shadow-producer.ts +++ b/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(); @@ -86,14 +87,18 @@ async function resolveRun(input: ShadowEventInput): Promise { return await input.store.getRun(input.runId); } -function buildAgentRunShadowMessage(input: ShadowEventInput & { run: RunRecord | null; config: AgentRunKafkaConfig }): { topic: string; key: string; value: JsonRecord; headers: Record } { +export function buildAgentRunShadowMessage(input: ShadowEventInput & { run: RunRecord | null; config: AgentRunKafkaConfig }): { topic: string; key: string; value: JsonRecord; headers: Record } { 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), diff --git a/src/selftest/cases/35-kafka-shadow-producer.ts b/src/selftest/cases/35-kafka-shadow-producer.ts new file mode 100644 index 0000000..2eb0bee --- /dev/null +++ b/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;