test(perception): seal source end marker before IPC shutdown

experiments/perception/worker/streaming_profile_stage1/pilot_stream_ingress_probe.py

@@ -145,6 +145,7 @@ def producer(args):
                     },
                 )
             end_ns = time.monotonic_ns()
+            (root / "source-end.json").write_text(json.dumps({"end_monotonic_ns": str(end_ns)}))
             sender.end()
         else:
             connection.sendall(wire.open_packet(identity, "recorded-acquisition", 1))
@@ -164,8 +165,8 @@ def producer(args):
             connection.sendall(payload)
             emit_json(journal, record(event))
             end_ns = time.monotonic_ns()
+            (root / "source-end.json").write_text(json.dumps({"end_monotonic_ns": str(end_ns)}))
             connection.close()  # Deliberate EOF between fragments; no End.
-    (root / "source-end.json").write_text(json.dumps({"end_monotonic_ns": str(end_ns)}))
     # The supervised source child remains alive until owned-group cleanup.
     # Its normal completion is not mistaken for an unexpected model death.
     threading.Event().wait(30)
@@ -184,11 +185,12 @@ def run_case(base, case, generation):
         "fragment_timeout_ms": 250,
         "idle_timeout_ms": 2000,
     }
+    experiment_id = json.loads(bounded_file(base / "metadata.json", 65536))["experiment_id"]
     identity = StreamStart(
-        run_id=f"binary-ingress-{case}",
+        run_id=f"{experiment_id}-{case}",
         source_id="RAVNOVES00-prefix" if case == "recorded" else "synthetic",
         worker_id="worker-006",
-        epoch_id=f"ipc-{generation}",
+        epoch_id=f"{experiment_id}-epoch-{generation}",
         lease_generation=generation,
         profile_sha256=hashlib.sha256(
             bounded_file(base / "candidate-profile.json", 65536)