diff --git a/scripts/compare-p10-brcb-sessions.py b/scripts/compare-p10-brcb-sessions.py index 4b8560ca6..fb40ab890 100644 --- a/scripts/compare-p10-brcb-sessions.py +++ b/scripts/compare-p10-brcb-sessions.py @@ -24,6 +24,7 @@ BUFOVFL = re.compile(r"(?:^|,)\s*BufOvfl=(\d+)(?:,|$)") SQ_GI = re.compile(r"\bsqWithGIreason=(\d+)\b") LIFECYCLE = re.compile(r"^\s*RCB lifecycle\s*:") +CHRONOLOGY = re.compile(r"^\s*BRCB chronology:\s*first=(\d+),\s*recent=(\d+),\s*omittedMiddle=(\d+)") def _field(line: str, key: str) -> str | None: match = re.search(rf"(?:^|,\s*){re.escape(key)}=([^,;\s]+)", line) @@ -45,6 +46,7 @@ class Stream: start_return_utc: str | None = None last_anomaly_utc: str | None = None chronology: tuple[tuple[str, ...], ...] = () + chronology_declared: tuple[int, int, int] | None = None recent_anomalies: tuple[tuple[str, ...], ...] = () def parse_report(report: str) -> dict: @@ -80,7 +82,8 @@ def parse_report(report: str) -> dict: anomalous_frames=0, discontinuities=0, after_last_finding=None, sq_with_gi_reason=None, gi_reason_frames=None, start_return_utc=None, - last_anomaly_utc=None, chronology=[], recent_anomalies=[]) + last_anomaly_utc=None, chronology=[], + chronology_declared=None, recent_anomalies=[]) continue if active is None or active not in streams: continue @@ -98,6 +101,9 @@ def parse_report(report: str) -> dict: streams[active]["gi_reason_frames"] = int(count) if count is not None and count.isdecimal() else None streams[active]["start_return_utc"] = _field(line, "startReturnUtc") streams[active]["last_anomaly_utc"] = _field(line, "lastAnomalyUtc") + m = CHRONOLOGY.match(line) + if m and streams[active]["kind"] == "BRCB": + streams[active]["chronology_declared"] = tuple(map(int, m.groups())) m = FRAME.match(line) if m: values = m.groups() @@ -128,6 +134,66 @@ def _elapsed_ms(start: str | None, end: str | None) -> float | None: except (TypeError, ValueError): return None +def _receive_order_gap_evidence(stream: Stream) -> dict: + """Only a fully retained decoder receive chronology authorizes temporal inferences. + + Neither callback timestamps nor SqNum/EntryID establish MMS packet order or SOE delivery. + An invalid, bounded, omitted or untimed chronology returns unknown, not zero. + """ + rows = stream.chronology + declared = stream.chronology_declared + reason = None + if declared is None: + reason = "MISSING_CHRONOLOGY_DECLARATION" + elif declared[2] != 0 or sum(declared[:2]) != stream.frames: + reason = "CHRONOLOGY_HAS_OMITTED_OR_UNACCOUNTED_FRAMES" + elif len(rows) != stream.frames or len(rows) != sum(declared[:2]): + reason = "RETAINED_CHRONOLOGY_INCOMPLETE" + elif [int(row[0]) for row in rows] != list(range(1, stream.frames + 1)): + reason = "RECEIVE_FRAME_INDICES_NOT_CONTIGUOUS" + times = [] + if reason is None: + for row in rows: + try: + moment = datetime.fromisoformat(row[1].replace("Z", "+00:00")) + if moment.tzinfo is None or moment.utcoffset() is None: + raise ValueError("no UTC offset") + except (TypeError, ValueError, OverflowError): + reason = "MISSING_OR_INVALID_RECEIVE_TIMESTAMPS" + break + times.append(moment) + if reason is None and any(b < a for a, b in zip(times, times[1:])): + reason = "NONMONOTONIC_DECODER_RECEIVE_TIMESTAMPS" + if reason is not None: + return dict(coverage="UNVERIFIABLE", reason=reason, + recorded_frames=len(rows), expected_frames=stream.frames, + longest_receive_gap_ms=None, + non_gi_backward_finding_after_gap=None) + gaps = [(b - a).total_seconds() * 1000 for a, b in zip(times, times[1:])] + candidates = [] + for index, gap in enumerate(gaps, start=1): + before, after = rows[index - 1], rows[index] + try: + prev_sq, curr_sq = int(before[2]), int(after[2]) + except ValueError: + continue + # A genuine decoded finding that recurs without GI or BufOvfl on this + # received frame; this is NOT evidence of harmless replay or lost SOE. + if (after[3] == "NO" and after[4] == "NO" and + int(after[6]) > 0 and curr_sq <= prev_sq): + candidates.append(dict(previous_frame=int(before[0]), frame=int(after[0]), + previous_sqnum=prev_sq, sqnum=curr_sq, + gap_ms=round(gap, 3), + elapsed_since_start_return_ms=_elapsed_ms( + stream.start_return_utc, after[1]), + received_utc=after[1], + entry_id_relation=after[5])) + strongest = max(candidates, key=lambda x: x["gap_ms"]) if candidates else None + return dict(coverage="COMPLETE_BOUNDED_DECODER_RECEIVE_ORDER", reason=None, + recorded_frames=len(rows), expected_frames=stream.frames, + longest_receive_gap_ms=round(max(gaps), 3) if gaps else None, + non_gi_backward_finding_after_gap=strongest) + def independent_triage(left: dict, right: dict) -> dict: """Evidence for *separate* sessions, never a paired BRCB reliability comparison.""" _require_matching_authority(left, right) @@ -155,14 +221,15 @@ def session_rows(streams: dict[str, Stream]) -> list[dict]: last_anomaly_ms_after_start_return=_elapsed_ms( stream.start_return_utc, stream.last_anomaly_utc), retained_chronology_frames=len(stream.chronology), - retained_anomaly_frames=len(stream.recent_anomalies))) + retained_anomaly_frames=len(stream.recent_anomalies), + receive_order_gap_evidence=_receive_order_gap_evidence(stream))) return rows return dict(status="INDEPENDENT_SESSIONS_OBSERVATION_ONLY", same_slot_dataset_scope=matched, paired_comparison_blocked_reason=None if matched else "different BRCB slots or DataSets", ied=left["ied"], fingerprint=left["fingerprint"], engine_commit=left["engine"], sessions=dict(open_scl=session_rows(la), discovery=session_rows(ra)), - qualification="NOT_PAIRED_RELIABILITY_COMPARISON; NOT_WIRE_ORDER_PROOF; NOT_SOE_LOSSLESS; NOT_R10_PHYSICAL_ACCEPTANCE") + qualification="NOT_PAIRED_RELIABILITY_COMPARISON; DECODER_RECEIVE_TIMESTAMPS_NOT_WIRE_ORDER; NOT_SOE_LOSSLESS; NOT_R10_PHYSICAL_ACCEPTANCE") def compare(left: dict, right: dict) -> dict: _require_matching_authority(left, right) diff --git a/scripts/test-compare-p10-brcb-sessions.py b/scripts/test-compare-p10-brcb-sessions.py index 847e8fef2..50661570e 100644 --- a/scripts/test-compare-p10-brcb-sessions.py +++ b/scripts/test-compare-p10-brcb-sessions.py @@ -139,6 +139,68 @@ def test_independent_triage_mismatched_engine_still_fails_closed(self): mod.independent_triage(mod.parse_report(report()), mod.parse_report(report(engine="f" * 40))) + def _full_chronology_report(self): + data = report(frames=4, overflow=1, with_chronology=False) + data = data.replace("BRCB chronology: first=2, recent=0, omittedMiddle=0", + "BRCB chronology: first=4, recent=0, omittedMiddle=0") + data = data.replace(" OptFlds : decoded=25/25", """ frame#1: receivedUtc=2026-10-11T06:25:51Z, SqNum=10, GIreason=NO, BufOvfl=YES, EntryIDrelation=unverified, findings=1 + frame#2: receivedUtc=2026-10-11T06:25:51.100Z, SqNum=11, GIreason=NO, BufOvfl=NO, EntryIDrelation=changed, findings=0 + frame#3: receivedUtc=2026-10-11T06:26:11Z, SqNum=1, GIreason=NO, BufOvfl=NO, EntryIDrelation=changed, findings=1 + frame#4: receivedUtc=2026-10-11T06:26:17Z, SqNum=2, GIreason=NO, BufOvfl=NO, EntryIDrelation=changed, findings=0 + OptFlds : decoded=25/25""") + return data + + def test_late_non_gi_recurrence_after_full_receive_gap(self): + sample = self._full_chronology_report() + triage = mod.independent_triage(mod.parse_report(sample), mod.parse_report(sample)) + evidence = triage["sessions"]["open_scl"][0]["receive_order_gap_evidence"] + self.assertEqual("COMPLETE_BOUNDED_DECODER_RECEIVE_ORDER", evidence["coverage"]) + self.assertEqual(19900.0, evidence["longest_receive_gap_ms"]) + anomaly = evidence["non_gi_backward_finding_after_gap"] + self.assertEqual(3, anomaly["frame"]) + self.assertEqual((11, 1), (anomaly["previous_sqnum"], anomaly["sqnum"])) + self.assertEqual(19900.0, anomaly["gap_ms"]) + self.assertEqual(21000.0, anomaly["elapsed_since_start_return_ms"]) + self.assertIn("NOT_SOE_LOSSLESS", triage["qualification"]) + + def test_truncated_chronology_never_invents_receive_gap(self): + sample = self._full_chronology_report() + sample = "\n".join(line for line in sample.splitlines() if "frame#2:" not in line) + evidence = mod.independent_triage(mod.parse_report(sample), mod.parse_report(sample))["sessions"]["open_scl"][0]["receive_order_gap_evidence"] + self.assertEqual("UNVERIFIABLE", evidence["coverage"]) + self.assertEqual("RETAINED_CHRONOLOGY_INCOMPLETE", evidence["reason"]) + self.assertIsNone(evidence["longest_receive_gap_ms"]) + self.assertIsNone(evidence["non_gi_backward_finding_after_gap"]) + + def test_omitted_middle_never_certifies_gaps(self): + sample = self._full_chronology_report() + sample = sample.replace("omittedMiddle=0", "omittedMiddle=1") + evidence = mod.independent_triage(mod.parse_report(sample), mod.parse_report(sample))["sessions"]["open_scl"][0]["receive_order_gap_evidence"] + self.assertEqual("CHRONOLOGY_HAS_OMITTED_OR_UNACCOUNTED_FRAMES", evidence["reason"]) + + def test_out_of_order_or_duplicate_frame_ids_do_not_certify_receive_order(self): + sample = self._full_chronology_report() + sample = sample.replace("frame#3:", "frame#2:") + evidence = mod.independent_triage(mod.parse_report(sample), mod.parse_report(sample))["sessions"]["open_scl"][0]["receive_order_gap_evidence"] + self.assertEqual("RECEIVE_FRAME_INDICES_NOT_CONTIGUOUS", evidence["reason"]) + + def test_nonmonotonic_timestamps_do_not_certify_receive_order(self): + sample = self._full_chronology_report() + sample = sample.replace("frame#3: receivedUtc=2026-10-11T06:26:11Z", + "frame#3: receivedUtc=2026-10-11T06:25:50Z") + evidence = mod.independent_triage(mod.parse_report(sample), mod.parse_report(sample))["sessions"]["open_scl"][0]["receive_order_gap_evidence"] + self.assertEqual("NONMONOTONIC_DECODER_RECEIVE_TIMESTAMPS", evidence["reason"]) + + def test_gi_reason_only_and_unverified_fixture_do_not_claim_non_gi_recurrence(self): + sample = self._full_chronology_report() + sample = sample.replace("frame#3: receivedUtc=2026-10-11T06:26:11Z, SqNum=1, GIreason=NO", + "frame#3: receivedUtc=2026-10-11T06:26:11Z, SqNum=1, GIreason=YES") + evidence = mod.independent_triage(mod.parse_report(sample), mod.parse_report(sample))["sessions"]["open_scl"][0]["receive_order_gap_evidence"] + self.assertEqual("COMPLETE_BOUNDED_DECODER_RECEIVE_ORDER", evidence["coverage"]) + self.assertIsNone(evidence["non_gi_backward_finding_after_gap"]) + old = mod.independent_triage(mod.parse_report(report()), mod.parse_report(report())) + self.assertEqual("UNVERIFIABLE", old["sessions"]["open_scl"][0]["receive_order_gap_evidence"]["coverage"]) + def test_empty_chronology_allowed_but_does_not_invent_events(self): result = mod.compare(mod.parse_report(report(with_chronology=False)), mod.parse_report(report(with_chronology=False)))