Merge pull request #1695 from pikasTech/fix/web-probe-stdio-session-lineage

fix: 按 AgentRun lineage 校验 stdio session

scripts/src/hwlab-node-web-observe-runner-realtime-source.test.ts

@@ -69,3 +69,17 @@ test("realtime fanout retries a YAML-bounded transient health navigation", async
   assert.equal(calls, 2);
   assert.equal((result.attempts as unknown[]).length, 2);
 });
+
+test("realtime fanout validates stdio against the scoped AgentRun session lineage", () => {
+  const source = nodeWebObserveRunnerRealtimeSource();
+  const build = new Function(`${source}\nreturn realtimeExpectedStreamSessionId;`) as () => (
+    stream: string,
+    hwlabSessionId: string,
+  ) => string;
+  const expectedStreamSessionId = build();
+  const hwlabSessionId = "ses_5ec4e141-6abc-466d-9afe-049f7c0ac105";
+
+  assert.equal(expectedStreamSessionId("stdio", hwlabSessionId), "ses_agentrun_5ec4e141_6abc_466d_9afe_049f7c0ac105");
+  assert.equal(expectedStreamSessionId("agentrun", hwlabSessionId), hwlabSessionId);
+  assert.equal(expectedStreamSessionId("hwlab", hwlabSessionId), hwlabSessionId);
+});

scripts/src/hwlab-node-web-observe-runner-realtime-source.ts

@@ -776,7 +776,10 @@ function realtimeValidateTurnEvidence(input) {
     const rows = (input.debug.records[stream] || []).filter((item) => item.traceId === input.traceId);
     if (rows.length === 0) throw new Error("turn " + input.turn + " " + stream + " has no trace records");
     if (!realtimeMonotonicOffsets(rows)) throw new Error("turn " + input.turn + " " + stream + " offsets are not monotonic");
-    if (rows.some((item) => item.hwlabSessionId !== input.sessionId)) throw new Error("turn " + input.turn + " " + stream + " session missing or mismatched");
+    const expectedSessionId = realtimeExpectedStreamSessionId(stream, input.sessionId);
+    if (rows.some((item) => item.hwlabSessionId !== expectedSessionId)) {
+      throw new Error("turn " + input.turn + " " + stream + " session lineage missing or mismatched");
+    }
     if (rows.some((item) => item.runId !== input.runId)) throw new Error("turn " + input.turn + " " + stream + " runId missing or mismatched");
     if (rows.some((item) => item.commandId && item.commandId !== input.commandId)) throw new Error("turn " + input.turn + " " + stream + " commandId mismatched");
     const commandBoundRows = stream === "stdio"
@@ -811,6 +814,15 @@ function realtimeValidateTurnEvidence(input) {
   }
 }
 
+function realtimeExpectedStreamSessionId(stream, hwlabSessionId) {
+  if (stream !== "stdio") return hwlabSessionId;
+  const base = String(hwlabSessionId || "")
+    .replace(/^ses_/u, "")
+    .replace(/[^A-Za-z0-9_]+/gu, "_")
+    .replace(/^_+|_+$/gu, "");
+  return base ? "ses_agentrun_" + base : null;
+}
+
 function realtimeAssertEnvelopePassthrough(kafkaRows, productRows, requireAllKafkaRows, label) {
   const sameEnvelope = (left, right) => left.envelopeFingerprint && left.envelopeFingerprint === right.envelopeFingerprint
     && left.eventId === right.eventId