From 98e4cde51f0dbbb4b6ece8505eedf5be61804d7b Mon Sep 17 00:00:00 2001 From: Codex Date: Sat, 11 Jul 2026 20:56:01 +0200 Subject: [PATCH] =?UTF-8?q?fix:=20=E6=A0=A1=E6=AD=A3=E5=AE=9E=E6=97=B6?= =?UTF-8?q?=E6=8E=A2=E9=92=88=E5=8F=AF=E9=80=89=E5=B7=A5=E5=85=B7=E4=BA=8B?= =?UTF-8?q?=E4=BB=B6=E5=90=88=E5=90=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- config/hwlab-node-lanes.yaml | 11 ++- scripts/src/hwlab-node-lanes.ts | 23 +++++ ...web-observe-runner-realtime-source.test.ts | 86 +++++++++++++++++++ ...node-web-observe-runner-realtime-source.ts | 73 ++++++++++++++-- .../web-probe-observe-actions.test.ts | 8 ++ .../web-probe-observe-options.test.ts | 10 ++- 6 files changed, 196 insertions(+), 15 deletions(-) diff --git a/config/hwlab-node-lanes.yaml b/config/hwlab-node-lanes.yaml index c474c40f..7f3aea15 100644 --- a/config/hwlab-node-lanes.yaml +++ b/config/hwlab-node-lanes.yaml @@ -602,16 +602,19 @@ templates: requiredEventTypes: agentrun: - user_message - - tool_call - - command_output - assistant_message - terminal_status hwlab: - user - - tool - - status - assistant - terminal + conditionalEventTypePairs: + tool-call: + agentrun: tool_call + hwlab: tool + command-output: + agentrun: command_output + hwlab: status workbenchKafkaDebugReplay: enabled: true topic: hwlab.event.debug.v1 diff --git a/scripts/src/hwlab-node-lanes.ts b/scripts/src/hwlab-node-lanes.ts index b67d9713..8f02280b 100644 --- a/scripts/src/hwlab-node-lanes.ts +++ b/scripts/src/hwlab-node-lanes.ts @@ -206,6 +206,10 @@ export interface HwlabRuntimeWebProbeRealtimeFanoutProfileSpec { agentrun: readonly string[]; hwlab: readonly string[]; }>; + readonly conditionalEventTypePairs: Readonly>>; } export interface HwlabRuntimeWebProbeAuthLoginSpec { @@ -1514,6 +1518,24 @@ function webProbeRealtimeFanoutProfileConfig(value: unknown, path: string): Hwla const hwlabEventTypes = nonEmptyStringArrayField(requiredEventTypes, "hwlab", `${path}.requiredEventTypes`); if (new Set(agentrunEventTypes).size !== agentrunEventTypes.length) throw new Error(`${path}.requiredEventTypes.agentrun must not contain duplicates`); if (new Set(hwlabEventTypes).size !== hwlabEventTypes.length) throw new Error(`${path}.requiredEventTypes.hwlab must not contain duplicates`); + const conditionalEventTypePairsRaw = asRecord(raw.conditionalEventTypePairs, `${path}.conditionalEventTypePairs`); + if (Object.keys(conditionalEventTypePairsRaw).length === 0) throw new Error(`${path}.conditionalEventTypePairs must contain at least one pair`); + const conditionalEventTypePairs = Object.fromEntries(Object.entries(conditionalEventTypePairsRaw).map(([id, value]) => { + if (!/^[a-z][a-z0-9-]*$/u.test(id)) throw new Error(`${path}.conditionalEventTypePairs.${id} must use a stable lowercase id`); + const pairPath = `${path}.conditionalEventTypePairs.${id}`; + const pair = asRecord(value, pairPath); + const unknownFields = Object.keys(pair).filter((field) => field !== "agentrun" && field !== "hwlab"); + if (unknownFields.length > 0) throw new Error(`${pairPath}.${unknownFields[0]} is not allowed`); + const agentrun = stringField(pair, "agentrun", pairPath); + const hwlab = stringField(pair, "hwlab", pairPath); + if (agentrunEventTypes.includes(agentrun)) throw new Error(`${pairPath}.agentrun duplicates a required event type`); + if (hwlabEventTypes.includes(hwlab)) throw new Error(`${pairPath}.hwlab duplicates a required event type`); + return [id, { agentrun, hwlab }]; + })); + const conditionalAgentrunTypes = Object.values(conditionalEventTypePairs).map((pair) => pair.agentrun); + const conditionalHwlabTypes = Object.values(conditionalEventTypePairs).map((pair) => pair.hwlab); + if (new Set(conditionalAgentrunTypes).size !== conditionalAgentrunTypes.length) throw new Error(`${path}.conditionalEventTypePairs agentrun types must not contain duplicates`); + if (new Set(conditionalHwlabTypes).size !== conditionalHwlabTypes.length) throw new Error(`${path}.conditionalEventTypePairs hwlab types must not contain duplicates`); return { subscriberCount: 2, debugStreams: debugStreams as ("stdio" | "agentrun" | "hwlab")[], @@ -1535,6 +1557,7 @@ function webProbeRealtimeFanoutProfileConfig(value: unknown, path: string): Hwla agentrun: agentrunEventTypes, hwlab: hwlabEventTypes, }, + conditionalEventTypePairs, }; } diff --git a/scripts/src/hwlab-node-web-observe-runner-realtime-source.test.ts b/scripts/src/hwlab-node-web-observe-runner-realtime-source.test.ts index f1082eff..cd22896b 100644 --- a/scripts/src/hwlab-node-web-observe-runner-realtime-source.test.ts +++ b/scripts/src/hwlab-node-web-observe-runner-realtime-source.test.ts @@ -36,6 +36,7 @@ test("realtime fanout keeps the YAML terminal timeout when durationMs is absent" }, forbiddenRequestPaths: [], requiredEventTypes: { agentrun: ["terminal_status"], hwlab: ["terminal"] }, + conditionalEventTypePairs: { "tool-call": { agentrun: "tool_call", hwlab: "tool" } }, expectedKafka: { topics: { stdio: "codex-stdio.raw.v1", agentrun: "agentrun.event.v1", hwlab: "hwlab.event.v1" }, groups: { directPublish: "direct", transactionalProjector: "projector", liveSse: "live" }, @@ -46,6 +47,91 @@ test("realtime fanout keeps the YAML terminal timeout when durationMs is absent" assert.equal(effectiveProfile(profile, null).terminalTimeoutMs, 480_000); assert.equal(effectiveProfile(profile, undefined).terminalTimeoutMs, 480_000); assert.equal(effectiveProfile(profile, 5_000).terminalTimeoutMs, 5_000); + assert.deepEqual(effectiveProfile(profile, null).conditionalEventTypePairs, profile.conditionalEventTypePairs); +}); + +test("realtime fanout terminal completion does not require absent optional tool families", () => { + const source = nodeWebObserveRunnerRealtimeSource(); + const build = new Function(`${source}\nreturn realtimeDebugTurnCompletion;`) as () => ( + snapshot: Record, traceId: string, profile: Record, + ) => Record; + const completion = build(); + const traceId = "trc_direct_answer"; + const profile = { + debugStreams: ["stdio", "agentrun", "hwlab"], + requiredEventTypes: { + agentrun: ["user_message", "assistant_message", "terminal_status"], + hwlab: ["user", "assistant", "terminal"], + }, + conditionalEventTypePairs: { + "tool-call": { agentrun: "tool_call", hwlab: "tool" }, + "command-output": { agentrun: "command_output", hwlab: "status" }, + }, + }; + const snapshot = { + records: { + stdio: [{ traceId, eventType: "codex.stdio.stdout" }], + agentrun: ["user_message", "backend_status", "assistant_message", "terminal_status"].map((eventType) => ({ traceId, eventType })), + hwlab: ["user", "backend", "assistant", "terminal"].map((eventType) => ({ traceId, eventType })), + }, + }; + + const directAnswer = completion(snapshot, traceId, profile); + assert.equal(directAnswer.complete, true); + assert.deepEqual(directAnswer.missingEventTypes, { agentrun: [], hwlab: [] }); + + const missingTerminal = completion({ + records: { + ...snapshot.records, + agentrun: snapshot.records.agentrun.filter((row: Record) => row.eventType !== "terminal_status"), + hwlab: snapshot.records.hwlab.filter((row: Record) => row.eventType !== "terminal"), + }, + }, traceId, profile); + assert.equal(missingTerminal.complete, false); + assert.deepEqual(missingTerminal.missingEventTypes, { agentrun: ["terminal_status"], hwlab: ["terminal"] }); + assert.ok(missingTerminal.observedEventTypes.agentrun.includes("assistant_message")); +}); + +test("realtime fanout validates optional tool events only when they occur", () => { + const source = nodeWebObserveRunnerRealtimeSource(); + const build = new Function(`${source}\nreturn realtimeValidateConditionalEventTypePairs;`) as () => ( + input: Record, + ) => void; + const validate = build(); + const traceId = "trc_optional_tool"; + const base = { + turn: 1, + traceId, + profile: { + conditionalEventTypePairs: { + "tool-call": { agentrun: "tool_call", hwlab: "tool" }, + "command-output": { agentrun: "command_output", hwlab: "status" }, + }, + }, + }; + + assert.doesNotThrow(() => validate({ ...base, debug: { records: { agentrun: [], hwlab: [] } } })); + assert.doesNotThrow(() => validate({ + ...base, + debug: { + records: { + agentrun: [{ traceId, eventType: "tool_call", eventId: "evt_tool" }], + hwlab: [{ traceId, eventType: "tool", sourceEventId: "evt_tool" }], + }, + }, + })); + assert.throws( + () => validate({ + ...base, + debug: { + records: { + agentrun: [{ traceId, eventType: "command_output", eventId: "evt_output" }], + hwlab: [], + }, + }, + }), + /conditional event pair command-output count differs/u, + ); }); test("realtime fanout retries a YAML-bounded transient health navigation", async () => { diff --git a/scripts/src/hwlab-node-web-observe-runner-realtime-source.ts b/scripts/src/hwlab-node-web-observe-runner-realtime-source.ts index ffd82a98..cfd7b8c0 100644 --- a/scripts/src/hwlab-node-web-observe-runner-realtime-source.ts +++ b/scripts/src/hwlab-node-web-observe-runner-realtime-source.ts @@ -747,6 +747,7 @@ function realtimeFanoutEffectiveProfile(profile, durationMs) { reconnectQuietMs: Number(profile.reconnectQuietMs), forbiddenRequestPaths: Array.isArray(profile.forbiddenRequestPaths) ? profile.forbiddenRequestPaths.map(String) : [], requiredEventTypes: profile.requiredEventTypes && typeof profile.requiredEventTypes === "object" ? profile.requiredEventTypes : {}, + conditionalEventTypePairs: profile.conditionalEventTypePairs && typeof profile.conditionalEventTypePairs === "object" ? sanitize(profile.conditionalEventTypePairs) : {}, expectedProductSse: sanitize(expectedProductSse), expectedKafka: sanitize(expectedKafka), }; @@ -1043,15 +1044,44 @@ async function realtimeWaitTurnIds(debug, key, traceId, timeoutMs) { } async function realtimeWaitDebugTurnComplete(debug, key, traceId, profile) { - return realtimePoll(async () => { - const snapshot = await realtimeDebugSnapshot(debug, key); - if (!profile.debugStreams.every((stream) => (snapshot.records[stream] || []).some((item) => item.traceId === traceId))) return null; - const agentrunTypes = new Set((snapshot.records.agentrun || []).filter((item) => item.traceId === traceId).map((item) => item.eventType)); - const hwlabTypes = new Set((snapshot.records.hwlab || []).filter((item) => item.traceId === traceId).map((item) => item.eventType)); - if (!(profile.requiredEventTypes.agentrun || []).every((type) => agentrunTypes.has(type))) return null; - if (!(profile.requiredEventTypes.hwlab || []).every((type) => hwlabTypes.has(type))) return null; - return snapshot; - }, profile.terminalTimeoutMs, "realtime debug terminal event families"); + let latest = null; + try { + return await realtimePoll(async () => { + const snapshot = await realtimeDebugSnapshot(debug, key); + latest = realtimeDebugTurnCompletion(snapshot, traceId, profile); + return latest.complete ? snapshot : null; + }, profile.terminalTimeoutMs, "realtime debug terminal event families"); + } catch (error) { + const wrapped = error instanceof Error ? error : new Error(String(error)); + const missing = latest?.missingEventTypes || { agentrun: [], hwlab: [] }; + wrapped.message = "realtime debug terminal event families missing agentrun=[" + missing.agentrun.join(",") + "] hwlab=[" + missing.hwlab.join(",") + "] after " + profile.terminalTimeoutMs + "ms"; + wrapped.details = { ...(wrapped.details || {}), traceId, completion: latest, valuesRedacted: true }; + throw wrapped; + } +} + +function realtimeDebugTurnCompletion(snapshot, traceId, profile) { + const observedEventTypes = {}; + const recordCounts = {}; + for (const stream of profile.debugStreams || []) { + const rows = (snapshot.records?.[stream] || []).filter((item) => item.traceId === traceId); + observedEventTypes[stream] = [...new Set(rows.map((item) => item.eventType).filter(Boolean))]; + recordCounts[stream] = rows.length; + } + const missingEventTypes = { + agentrun: (profile.requiredEventTypes?.agentrun || []).filter((type) => !observedEventTypes.agentrun?.includes(type)), + hwlab: (profile.requiredEventTypes?.hwlab || []).filter((type) => !observedEventTypes.hwlab?.includes(type)), + }; + const missingStreams = (profile.debugStreams || []).filter((stream) => !recordCounts[stream]); + return { + complete: missingStreams.length === 0 && missingEventTypes.agentrun.length === 0 && missingEventTypes.hwlab.length === 0, + traceId, + observedEventTypes, + missingEventTypes, + missingStreams, + recordCounts, + valuesRedacted: true, + }; } async function realtimeWaitTurnStable(debug, key, traceId, subscribers, profile) { @@ -1241,6 +1271,7 @@ function realtimeValidateTurnEvidence(input) { const hwlabTypes = new Set(input.debug.records.hwlab.filter((item) => item.traceId === input.traceId).map((item) => item.eventType)); for (const type of profile.requiredEventTypes.agentrun || []) if (!agentrunTypes.has(type)) throw new Error("turn " + input.turn + " AgentRun lacks " + type); for (const type of profile.requiredEventTypes.hwlab || []) if (!hwlabTypes.has(type)) throw new Error("turn " + input.turn + " HWLAB lacks " + type); + realtimeValidateConditionalEventTypePairs(input); const identityRows = Object.values(input.debug.records).flat().filter((item) => item.traceId === input.traceId); if (!input.runId || new Set(identityRows.map((item) => item.runId)).size !== 1) throw new Error("turn " + input.turn + " runId missing or mismatched across streams"); const commandIds = new Set(identityRows.map((item) => item.commandId).filter(Boolean)); @@ -1267,6 +1298,30 @@ function realtimeValidateTurnEvidence(input) { } } +function realtimeValidateConditionalEventTypePairs(input) { + const pairs = input.profile.conditionalEventTypePairs || {}; + const agentrunRows = (input.debug.records.agentrun || []).filter((item) => item.traceId === input.traceId); + const hwlabRows = (input.debug.records.hwlab || []).filter((item) => item.traceId === input.traceId); + for (const [id, pair] of Object.entries(pairs)) { + const sourceRows = agentrunRows.filter((item) => item.eventType === pair.agentrun); + const projectedRows = hwlabRows.filter((item) => item.eventType === pair.hwlab); + if (sourceRows.length === 0 && projectedRows.length === 0) continue; + if (sourceRows.length !== projectedRows.length) { + throw new Error("turn " + input.turn + " conditional event pair " + id + " count differs AgentRun=" + sourceRows.length + " HWLAB=" + projectedRows.length); + } + for (const source of sourceRows) { + if (!source.eventId || !projectedRows.some((projected) => projected.sourceEventId === source.eventId)) { + throw new Error("turn " + input.turn + " conditional event pair " + id + " lacks HWLAB sourceEventId lineage"); + } + } + for (const projected of projectedRows) { + if (!projected.sourceEventId || !sourceRows.some((source) => source.eventId === projected.sourceEventId)) { + throw new Error("turn " + input.turn + " conditional event pair " + id + " has an orphan HWLAB event"); + } + } + } +} + function realtimeExpectedStreamSessionId(stream, hwlabSessionId) { if (stream !== "stdio") return hwlabSessionId; const base = String(hwlabSessionId || "") diff --git a/scripts/src/hwlab-node/web-probe-observe-actions.test.ts b/scripts/src/hwlab-node/web-probe-observe-actions.test.ts index 466b9cf1..149facfc 100644 --- a/scripts/src/hwlab-node/web-probe-observe-actions.test.ts +++ b/scripts/src/hwlab-node/web-probe-observe-actions.test.ts @@ -28,6 +28,14 @@ test("realtime fanout runner contract derives topics, groups, and independent ca projectionOutboxRelay: false, projectionRealtime: false, }); + assert.deepEqual(profiles["pure-kafka-live"]?.requiredEventTypes, { + agentrun: ["user_message", "assistant_message", "terminal_status"], + hwlab: ["user", "assistant", "terminal"], + }); + assert.deepEqual(profiles["pure-kafka-live"]?.conditionalEventTypePairs, { + "tool-call": { agentrun: "tool_call", hwlab: "tool" }, + "command-output": { agentrun: "command_output", hwlab: "status" }, + }); }); test("Workbench Kafka debug replay runner contract is derived from owning YAML", () => { diff --git a/scripts/src/hwlab-node/web-probe-observe-options.test.ts b/scripts/src/hwlab-node/web-probe-observe-options.test.ts index 18d9f737..c7d11dcb 100644 --- a/scripts/src/hwlab-node/web-probe-observe-options.test.ts +++ b/scripts/src/hwlab-node/web-probe-observe-options.test.ts @@ -41,8 +41,14 @@ test("validateRealtimeFanout parses the YAML profile and two independent prompts refreshHandoff: true, replayPriorTraceOnReconnect: true, }); - assert.ok(spec.webProbe?.realtimeFanoutProfiles?.["pure-kafka-live"]?.requiredEventTypes.agentrun.includes("user_message")); - assert.ok(spec.webProbe?.realtimeFanoutProfiles?.["pure-kafka-live"]?.requiredEventTypes.hwlab.includes("user")); + assert.deepEqual(spec.webProbe?.realtimeFanoutProfiles?.["pure-kafka-live"]?.requiredEventTypes, { + agentrun: ["user_message", "assistant_message", "terminal_status"], + hwlab: ["user", "assistant", "terminal"], + }); + assert.deepEqual(spec.webProbe?.realtimeFanoutProfiles?.["pure-kafka-live"]?.conditionalEventTypePairs, { + "tool-call": { agentrun: "tool_call", hwlab: "tool" }, + "command-output": { agentrun: "command_output", hwlab: "status" }, + }); }); test("validateRealtimeFanout refuses body as a substitute for the second prompt", () => {