M3E.3: fail-closed SOURCE readiness on STOP, TX faults and boot ambiguity

docs/PTP_M3E_BENCH_EVIDENCE_GATE.md

@@ -111,4 +111,22 @@ and canonical PCAP continuity. All hardware qualification fields remain
 
 All M3E.2 changes are Python host-tool regression/doc additions. The proven
 M3D.2 SOURCE/RECEIVER firmware binaries and native ARStack Studio are not
-rebuilt or modified by this checkpoint.
+rebuilt or modified by this checkpoint.
+## M3E.3 — SOURCE readiness freshness / boot provenance regression (2026-10-17)
+
+A previous host gate accepted **any** historical `PTP status=RUNNING`
+sample with positive Announce/Sync/FollowUp as SOURCE readiness. This could
+return `operationalLifecycle.status=OBSERVED` even when a later STOP, rejected
+START, unavailable status, or TX fault was already present in the same raw log.
+
+The M3E.3 host-only gate now requires the **latest** SOURCE status to be
+RUNNING with complete native firmware counters and `TXfail=0`. It revokes
+readiness on `PTP stop requested`, `PTP start rejected`, STOPPED, truncated
+SHOW, or unavailable status. Any nonzero reported PTP TX failure in the same
+boot is conservatively latched as a veto. Every IDENTIFY line must also carry
+the same 16-hex-digit boot ID; an identity without a boot ID cannot inherit
+a preceding measurement. Receiver logic and wire decoding are unchanged.
+
+This is regression protection for **firmware-reported** software evidence,
+not proof of wire transmission, EMAC phase alignment, IED lock, or public
+release readiness. No firmware images or Studio workflow have changed.

scripts/ptp_m3e_bench_gate.py

@@ -27,9 +27,11 @@
 SCENARIOS = ("auto", "receiver-acquisition", "receiver-recovery",
              "receiver-gm-handoff", "source-ready")
 BOOT_ID = re.compile(r"\bboot_id=([a-fA-F0-9]{16})\b")
-SOURCE_READY = re.compile(
-    r"\bPTP status=RUNNING Announce=([0-9]+) Sync=([0-9]+) "
-    r"FollowUp=([0-9]+)\b")
+SOURCE_STATUS = re.compile(
+    r"\bPTP status=(RUNNING|STOPPED) Announce=([0-9]+) Sync=([0-9]+) "
+    r"FollowUp=([0-9]+) PdelayFrames=([0-9]+) TXfail=([0-9]+)\b")
+SOURCE_INVALIDATION = re.compile(
+    r"\bPTP (?:stop requested|start rejected|status unavailable)\b")
 STATUS_FIELD = re.compile(r"\b([A-Za-z][A-Za-z0-9]*)=([^\s]+)")
 
 class EvidenceError(ValueError):
@@ -203,24 +205,49 @@ def operational_evidence(data: bytes, role: str, scenario: str) -> dict:
     require((role == "source") == (scenario == "source-ready"),
             "scenario incompatible with firmware role")
     lines = data.decode("utf-8", errors="replace").splitlines()
-    boot_ids = [m.group(1).lower() for line in lines
-                if IDENTITY.search(line) and (m := BOOT_ID.search(line))]
-    if len(set(boot_ids)) != 1 or not boot_ids:
+    identity_lines = [line for line in lines if IDENTITY.search(line)]
+    boot_ids = [m.group(1).lower() for line in identity_lines
+                if (m := BOOT_ID.search(line))]
+    # Every board identity must belong to the same observed boot. A later
+    # identity missing boot_id must not silently inherit the previous lease.
+    if (not boot_ids or len(boot_ids) != len(identity_lines) or
+            len(set(boot_ids)) != 1):
         return {"status": "INCOMPLETE", "scenario": scenario,
                 "reason": "one unambiguous 16-digit boot_id required",
                 "reportedLockSamples": 0, "reportedRecovery": False,
                 "reportedGmHandoff": False}
 
     if role == "source":
-        ready = [tuple(int(x) for x in m.groups())
-                 for line in lines if (m := SOURCE_READY.search(line))]
-        usable = any(all(x > 0 for x in values) for values in ready)
+        # A historical RUNNING line is not current readiness. The actual
+        # firmware SHOW includes TXfail and PdelayFrames; partial output is
+        # never acceptable. STOP, rejected START and unavailable status
+        # invalidate all prior ready observations.
+        last_ready = False
+        tx_fault_observed = False
+        ready_samples = 0
+        for line in lines:
+            if SOURCE_INVALIDATION.search(line):
+                last_ready = False
+            if "PTP status=" not in line:
+                continue
+            match = SOURCE_STATUS.search(line)
+            if match is None:
+                last_ready = False
+                continue
+            state, *reported = match.groups()
+            announce, sync, followup, _pdelay, txfail = map(int, reported)
+            tx_fault_observed = tx_fault_observed or txfail != 0
+            last_ready = (state == "RUNNING" and
+                          announce > 0 and sync > 0 and followup > 0 and
+                          txfail == 0)
+            ready_samples += int(last_ready)
+        usable = last_ready and not tx_fault_observed
         return {"status": "OBSERVED" if usable else "INCOMPLETE",
                 "scenario": scenario,
-                "reason": "source Announce+Sync+FollowUp TX readiness reported"
-                          if usable else "verified source TX readiness evidence missing",
-                "reportedSourceReadySamples": sum(all(x > 0 for x in v)
-                                                   for v in ready),
+                "reason": "latest SOURCE status RUNNING with successful PTP TX"
+                          if usable else "latest SOURCE readiness absent, stopped or TX faulted",
+                "reportedSourceReadySamples": ready_samples,
+                "sourceTxFaultObserved": tx_fault_observed,
                 "reportedLockSamples": 0,
                 "reportedRecovery": False, "reportedGmHandoff": False}
 

tests/test_ptp_m3e_bench_gate.py

@@ -319,6 +319,43 @@ def test_source_requires_actual_sent_ptp_triplet(self):
             "source-ready", "source")
         self.assertEqual(complete["status"], "OBSERVED")
 
+    def test_source_readiness_is_latest_not_any_historical_running(self):
+        prefix = (f"ARSTACK identity build={HEAD[:16]} boot_id={BOOT}\n"
+                  "PTP2 role=SOURCE discipline=UNLOCKED source=NONE offset=NA\n")
+        running = "PTP status=RUNNING Announce=4 Sync=4 FollowUp=4 PdelayFrames=0 TXfail=0\n"
+        for suffix in ("PTP stop requested\n",
+                       "PTP status=STOPPED Announce=4 Sync=4 FollowUp=4 PdelayFrames=0 TXfail=0\n",
+                       "PTP status unavailable\n",
+                       "PTP start rejected: link down\n",
+                       "PTP status=RUNNING Announce=4 Sync=4 FollowUp=4 PdelayFrames=0\n"):
+            with self.subTest(suffix=suffix):
+                actual = self.run_gate((prefix + running + suffix).encode(),
+                                       "source-ready", "source")
+                self.assertEqual(actual["status"], "INCOMPLETE")
+        # A later clean RUNNING status is evidence of a newly reported state.
+        clean_restart = prefix + running + "PTP stop requested\n" + running
+        self.assertEqual(self.run_gate(clean_restart.encode(), "source-ready",
+                                       "source")["status"], "OBSERVED")
+
+    def test_source_any_reported_tx_failure_vetoes_same_boot(self):
+        prefix = (f"ARSTACK identity build={HEAD[:16]} boot_id={BOOT}\n"
+                  "PTP2 role=SOURCE discipline=UNLOCKED source=NONE offset=NA\n")
+        bad = "PTP status=RUNNING Announce=4 Sync=4 FollowUp=4 PdelayFrames=0 TXfail=1\n"
+        good = "PTP status=RUNNING Announce=8 Sync=8 FollowUp=8 PdelayFrames=0 TXfail=0\n"
+        actual = self.run_gate((prefix + bad + good).encode(), "source-ready", "source")
+        self.assertEqual(actual["status"], "INCOMPLETE")
+        self.assertTrue(actual["sourceTxFaultObserved"])
+        self.assertEqual(self.run_gate((prefix + bad).encode(), "source-ready",
+                                       "source")["status"], "INCOMPLETE")
+
+    def test_missing_boot_id_on_later_identity_fails_closed(self):
+        data = (f"ARSTACK identity build={HEAD[:16]} boot_id={BOOT}\n"
+                "PTP2 role=SOURCE discipline=UNLOCKED source=NONE offset=NA\n"
+                f"ARSTACK identity build={HEAD[:16]}\n"
+                "PTP status=RUNNING Announce=4 Sync=4 FollowUp=4 PdelayFrames=0 TXfail=0\n")
+        self.assertEqual(self.run_gate(data.encode(), "source-ready",
+                                       "source")["status"], "INCOMPLETE")
+
     def test_incompatible_role_scenario_is_rejected(self):
         with self.assertRaisesRegex(module.EvidenceError, "incompatible"):
             self.run_gate(bench_log("LOCKED"), "source-ready", "receiver")