diff --git a/runs/action-aware-v0/analyze_pilot.py b/runs/action-aware-v0/analyze_pilot.py index 8c438ec..ba5511a 100644 --- a/runs/action-aware-v0/analyze_pilot.py +++ b/runs/action-aware-v0/analyze_pilot.py @@ -138,6 +138,37 @@ def binding_summary( } +def telemetry_coverage( + records: list[Mapping[str, Any]], *, start_ns: int, end_ns: int +) -> tuple[dict[str, float], bool]: + if not records: + raise ValueError("telemetry coverage requires records") + submit_gaps = [ + (int(right["submit_mono_ns"]) - int(left["submit_mono_ns"])) / 1e9 + for left, right in zip(records, records[1:], strict=False) + ] + uncovered_gaps = [ + max( + 0, + int(right["submit_mono_ns"]) - int(left["complete_mono_ns"]), + ) + / 1e9 + for left, right in zip(records, records[1:], strict=False) + ] + coverage = { + "start_gap_s": (int(records[0]["submit_mono_ns"]) - start_ns) / 1e9, + "end_gap_s": (end_ns - int(records[-1]["submit_mono_ns"])) / 1e9, + "max_internal_submit_gap_s": max(submit_gaps, default=0.0), + "max_uncovered_gap_s": max(uncovered_gaps, default=0.0), + } + covered = ( + 0.0 <= coverage["start_gap_s"] <= 1.0 + and 0.0 <= coverage["end_gap_s"] <= 1.0 + and 0.0 <= coverage["max_uncovered_gap_s"] <= 1.0 + ) + return coverage, covered + + def request_summary(path: Path, expected_count: int) -> dict[str, Any]: rows = load_jsonl(path) if len(rows) != expected_count: @@ -222,17 +253,8 @@ def analyze_run( ] if not full_records: raise ValueError(f"no telemetry records in measured window: {result_path}") - gaps = [ - (int(right["submit_mono_ns"]) - int(left["submit_mono_ns"])) / 1e9 - for left, right in zip(full_records, full_records[1:], strict=False) - ] - coverage = { - "start_gap_s": (int(full_records[0]["submit_mono_ns"]) - start_ns) / 1e9, - "end_gap_s": (arrival_end_ns - int(full_records[-1]["submit_mono_ns"])) / 1e9, - "max_internal_gap_s": max(gaps, default=0.0), - } - invariants["telemetry_coverage"] = all( - 0.0 <= value <= 1.0 for value in coverage.values() + coverage, invariants["telemetry_coverage"] = telemetry_coverage( + full_records, start_ns=start_ns, end_ns=arrival_end_ns ) binding = binding_summary( full_records, mns=int(config["mns"]), mbbt=int(config["mbbt"]) diff --git a/runs/action-aware-v0/test_pilot.py b/runs/action-aware-v0/test_pilot.py index c2867d9..afb1e29 100644 --- a/runs/action-aware-v0/test_pilot.py +++ b/runs/action-aware-v0/test_pilot.py @@ -81,6 +81,25 @@ def main() -> None: assert summary["waiting_unresolved_count"] == 1 assert summary["waiting_count"] == 4 + # A per-step stream may have a submit gap above one second when the + # preceding model execution itself spans that interval. Such a gap is + # covered telemetry, not a dropped-record interval. + asynchronous = [ + {"submit_mono_ns": 0, "complete_mono_ns": 1_200_000_000}, + {"submit_mono_ns": 1_100_000_000, "complete_mono_ns": 1_300_000_000}, + ] + coverage, covered = analysis.telemetry_coverage( + asynchronous, start_ns=0, end_ns=1_100_000_000 + ) + assert coverage["max_internal_submit_gap_s"] == 1.1 + assert coverage["max_uncovered_gap_s"] == 0.0 + assert covered + missing = copy.deepcopy(asynchronous) + missing[0]["complete_mono_ns"] = 0 + assert not analysis.telemetry_coverage( + missing, start_ns=0, end_ns=1_100_000_000 + )[1] + manifest = { "repetitions": {str(index): {} for index in (1, 2, 3)}, "regimes": {