Merge pull request #1792 from pikasTech/fix/1791-webprobe-optional-tool-events

fix: 允许实时探针验收无工具调用的合法终态

.agents/skills/unidesk-webdev/SKILL.md

@@ -42,6 +42,9 @@ description: UniDesk Web 开发与浏览器验证技能。用户处理 UniDesk/H
 - Workbench Kafka 实时 fanout 的重复验收：
   - 使用 `observe command --type validateRealtimeFanout --profile <yaml-profile> --provider <provider> --text <first> --second-text <second>`；
   - connected、refresh/reconnect 和 capability 期望来自 owning YAML，可独立组合，禁止在命令中硬编码 live-only 或 refresh mode；
+  - 每轮只把 `user`、`assistant` 和 `terminal` 对应的 AgentRun/HWLAB 事件族作为必需核心事件；
+  - `tool_call`/`command_output` 与 HWLAB `tool`/`status` 是条件事件对，仅在实际出现时校验数量和 lineage；
+  - 直接答复不得为了通过探针而合成工具事件，超时证据必须披露已经观测和仍然缺失的事件族；
   - refresh 启用时必须证明 retention 到 live handoff、正式 `user` event、`hwlab.event.v1` 同 envelope，以及主页面单一 EventSource 和同一 UI reducer 终态；
   - observer 后台维持两个独立 product subscriber 和三条 debug SSE；
   - 命令提交后按 `status <observer> --command-id <command>` 轮询，不得回退为长连接 `web-probe script`。

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

scripts/src/hwlab-node-lanes.ts

@@ -206,6 +206,10 @@ export interface HwlabRuntimeWebProbeRealtimeFanoutProfileSpec {
     agentrun: readonly string[];
     hwlab: readonly string[];
   }>;
+  readonly conditionalEventTypePairs: Readonly<Record<string, Readonly<{
+    agentrun: string;
+    hwlab: string;
+  }>>>;
 }
 
 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,
   };
 }
 

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<string, any>, traceId: string, profile: Record<string, any>,
+  ) => Record<string, any>;
+  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<string, string>) => row.eventType !== "terminal_status"),
+      hwlab: snapshot.records.hwlab.filter((row: Record<string, string>) => 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<string, any>,
+  ) => 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 () => {

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 || "")

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", () => {

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", () => {