diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_stream_ingress_probe.py b/experiments/perception/worker/streaming_profile_stage1/pilot_stream_ingress_probe.py index ddd7fdb..fae8b2b 100644 --- a/experiments/perception/worker/streaming_profile_stage1/pilot_stream_ingress_probe.py +++ b/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)