From 8c585e3c2f9319a67decbef31a106f08f4c342b6 Mon Sep 17 00:00:00 2001
From: Codex <codex@local>
Date: Fri, 10 Jul 2026 13:12:58 +0200
Subject: [PATCH] fix(web-probe): validate stdio session lineage

---
 ...node-web-observe-runner-realtime-source.test.ts | 14 ++++++++++++++
 ...wlab-node-web-observe-runner-realtime-source.ts | 14 +++++++++++++-
 2 files changed, 27 insertions(+), 1 deletion(-)

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 ae9e1d2c..a2c79b70 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
@@ -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);
+});
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 0c6fd73f..56f31e0e 100644
--- a/scripts/src/hwlab-node-web-observe-runner-realtime-source.ts
+++ b/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