feat(gate): implement §4 liveness contract — fail-loud on silent degradation (proposer/f_θ/fallback)

Turns the methodology's §4 from doc into an executable defense:
- k3_report_gate.assert_liveness(): asserts proposer ran (blocks>0), f_θ ran when
  intended (f_theta_ran on every turn), and fallbacks_taken==[] — from RUNTIME
  signals, not from flags. Missing liveness field = violation. New codes:
  PROPOSER_NEVER_RAN, FTHETA_NOT_RUN, SILENT_FALLBACK, MISSING_LIVENESS.
- validate_report() dispatches liveness reports; validate_k3_reports.py (CI + the
  Mac-bridge on-device gate) now gates kind=mac_gemma4_kakeya_fused_chat.
- harness emits f_theta_intended + fallbacks_taken in the chat report.
- fused-chat presets set validate_reports=True → the Mac runner FAILS if the
  engine silently degraded to verifier-only.
- 100% coverage on k3_report_gate + manifest; walker verified to FAIL a degraded
  report (blocks=0 / f_θ bypassed) and pass a live one.

Co-authored-by: FluffyAIcode <FluffyAIcode@users.noreply.github.com>

Author: cursoragent

inference_engine/bench/k3_report_gate.py

@@ -89,6 +89,12 @@
 
 NATIVE_BASELINE_LABEL = "native_ar_baseline"
 
+# §4 liveness contract (docs/kakeya-autonomous-iteration-and-self-correction.md):
+# report kinds that carry per-turn component-liveness signals. The gate proves
+# the INTENDED components actually ran — the antidote to silent fallback /
+# simplification (proposer→AR, f_θ→bypass) that kept the "fused" label.
+LIVENESS_REPORT_KINDS = frozenset({"mac_gemma4_kakeya_fused_chat"})
+
 
 @dataclass(frozen=True)
 class GateViolation:
@@ -103,6 +109,82 @@ def is_gated_report(report: Any) -> bool:
     return isinstance(report, dict) and report.get("kind") == MAC_REPORT_KIND
 
 
+def is_liveness_report(report: Any) -> bool:
+    """True when ``report`` carries the §4 component-liveness contract."""
+    return isinstance(report, dict) and report.get("kind") in LIVENESS_REPORT_KINDS
+
+
+def assert_liveness(report: Dict[str, Any]) -> List[GateViolation]:
+    """§4 liveness contract — prove the intended components actually executed.
+
+    Asserts, from RUNTIME signals (never from flags passed in):
+      * proposer ran      — total proposer ``blocks`` across turns > 0,
+      * f_θ ran           — when ``f_theta_intended`` is true, every turn has
+                            ``f_theta_ran == true``,
+      * no silent fallback — ``fallbacks_taken`` (report- and turn-level) empty.
+    A missing liveness field is itself a violation (absence = "we don't know it
+    ran" = invalid), not a skip.
+    """
+    violations: List[GateViolation] = []
+    turns = report.get("turns")
+    if not isinstance(turns, list) or not turns:
+        return [GateViolation(
+            "MISSING_LIVENESS",
+            "liveness report has no 'turns'; component liveness cannot be "
+            "asserted (absence of evidence = invalid run)",
+        )]
+
+    # --- proposer liveness: blocks > 0 (else it silently fell back to AR) ---
+    total_blocks = 0
+    missing_blocks = False
+    for t in turns:
+        b = t.get("blocks") if isinstance(t, dict) else None
+        if isinstance(b, (int, float)) and not isinstance(b, bool):
+            total_blocks += int(b)
+        else:
+            missing_blocks = True
+    if missing_blocks:
+        violations.append(GateViolation(
+            "MISSING_LIVENESS",
+            "a turn lacks numeric 'blocks'; proposer liveness unknown",
+        ))
+    elif total_blocks <= 0:
+        violations.append(GateViolation(
+            "PROPOSER_NEVER_RAN",
+            "fused chat executed 0 proposer blocks across all turns — the "
+            "proposer silently fell back to native AR (verifier-only)",
+        ))
+
+    # --- f_θ liveness: if intended, it must run on every turn ---
+    if report.get("f_theta_intended") is True:
+        ran = [t.get("f_theta_ran") if isinstance(t, dict) else None for t in turns]
+        if any(r is None for r in ran):
+            violations.append(GateViolation(
+                "MISSING_LIVENESS",
+                "f_theta_intended=true but a turn lacks 'f_theta_ran'",
+            ))
+        elif not all(bool(r) for r in ran):
+            violations.append(GateViolation(
+                "FTHETA_NOT_RUN",
+                "f_theta_intended=true but f_theta_ran is false on >=1 turn — "
+                "f_θ restoration was silently bypassed",
+            ))
+
+    # --- no silent fallback: declared fallbacks (report- + turn-level) empty ---
+    fallbacks: List[str] = [str(x) for x in (report.get("fallbacks_taken") or [])]
+    for t in turns:
+        if isinstance(t, dict):
+            fallbacks += [str(x) for x in (t.get("fallbacks_taken") or [])]
+    if fallbacks:
+        violations.append(GateViolation(
+            "SILENT_FALLBACK",
+            f"fallbacks_taken is non-empty: {sorted(set(fallbacks))} — a "
+            "component degraded to a fallback; the system under test is not "
+            "the intended one",
+        ))
+    return violations
+
+
 def is_legacy_report(report: Dict[str, Any]) -> bool:
     """True when the report predates the evidence gate (schema < 2)."""
     try:
@@ -202,6 +284,8 @@ def validate_report(report: Dict[str, Any]) -> List[GateViolation]:
     one ``LEGACY_SCHEMA`` violation (the CI walker downgrades that one
     code to a warning — everything else fails the build).
     """
+    if is_liveness_report(report):
+        return assert_liveness(report)
     if not is_gated_report(report):
         return []
     if is_legacy_report(report):

inference_engine/bridge/manifest.py

@@ -711,7 +711,7 @@ def _harness_preset(
                 "max_new_tokens": ("int:max_new_tokens", "64"),
                 "block_size": ("int:block_size", "4"),
             },
-            validate_reports=False,
+            validate_reports=True,  # §4 liveness gate on-device (proposer/f_θ/fallback)
         ),
         Preset(
             name="mlx-kakeya-fused-chat-ftheta",
@@ -747,7 +747,7 @@ def _harness_preset(
                 "max_new_tokens": ("int:max_new_tokens", "32"),
                 "block_size": ("int:block_size", "4"),
             },
-            validate_reports=False,
+            validate_reports=True,  # §4 liveness gate: asserts f_theta_ran on-device
         ),
         Preset(
             name="mlx-kakeya-launcher-smoke",
@@ -769,7 +769,7 @@ def _harness_preset(
             ),
             timeout_minutes=45,
             params={"max_new_tokens": ("int:max_new_tokens", "64")},
-            validate_reports=False,
+            validate_reports=True,  # §4 liveness gate on-device
         ),
     )
 }

scripts/research/k3_integrated_niah_eval_mac.py

@@ -862,6 +862,11 @@ def _gen_turn(pid: List[int]) -> Dict[str, Any]:
                     "f_theta_dir": args.f_theta_dir, "sink": args.sink_size,
                     "window": args.window_size, "block_size": args.block_size,
                     "exact_layers": full_attn_idx, "chat_eos": sorted(chat_eos),
+                    # §4 liveness contract: f_θ is INTENDED on the torch path
+                    # (not --all-mlx-drafter); the evidence gate asserts
+                    # f_theta_ran on every turn when this is true.
+                    "f_theta_intended": mlx_drafter is None,
+                    "fallbacks_taken": [],
                     "turns": transcript}
                 if args.output:
                     op = Path(args.output)

scripts/validate_k3_reports.py

@@ -31,6 +31,7 @@
 from inference_engine.bench.k3_report_gate import (
     is_gated_report,
     is_legacy_report,
+    is_liveness_report,
     summarize_violations,
     validate_report,
 )
@@ -47,9 +48,13 @@ def main(argv: list) -> int:
             report = json.loads(path.read_text())
         except (json.JSONDecodeError, UnicodeDecodeError, OSError):
             continue
-        if not is_gated_report(report):
+        gated = is_gated_report(report)
+        live = is_liveness_report(report)
+        if not (gated or live):
             continue
-        if is_legacy_report(report):
+        # The schema-2 legacy grandfather applies only to the NIAH acceptance
+        # report; liveness reports (§4 contract) are always asserted.
+        if gated and is_legacy_report(report):
             legacy += 1
             print(f"[legacy] {path}: schema<2 — grandfathered, NON-EVIDENCE "
                   "(rerun with the hardened harness to make claims)")

tests/inference_engine/bench/test_k3_report_gate.py

@@ -19,22 +19,108 @@
 from inference_engine.bench.k3_report_gate import (
     CLAIM_ORACLE_DECODE_LOOP,
     GATED_SCHEMA_VERSION,
+    LIVENESS_REPORT_KINDS,
     MAC_REPORT_KIND,
     MAX_PREFILL_SPREAD,
     MIN_MEDIAN_DECODE_TOKENS,
     MIN_PERF_SAMPLES,
     NATIVE_BASELINE_LABEL,
     GateViolation,
+    assert_liveness,
     decode_only_block,
     is_gated_report,
     is_legacy_report,
+    is_liveness_report,
     prefill_spread,
     row_prefill_seconds,
     summarize_violations,
     validate_report,
 )
 
 
+# ---------------------------------------------------------------------------
+# §4 liveness contract (proposer / f_θ / no-fallback)
+# ---------------------------------------------------------------------------
+
+
+def _live_report(**over: Any) -> Dict[str, Any]:
+    """A fused-chat liveness report that passes the §4 contract."""
+    rep = {
+        "kind": next(iter(LIVENESS_REPORT_KINDS)),
+        "schema_version": 1,
+        "f_theta_intended": True,
+        "fallbacks_taken": [],
+        "turns": [
+            {"user": "q1", "blocks": 2, "mean_accept_len": 4.0,
+             "f_theta_ran": True, "fallbacks_taken": []},
+            {"user": "q2", "blocks": 4, "mean_accept_len": 3.5,
+             "f_theta_ran": True, "fallbacks_taken": []},
+        ],
+    }
+    rep.update(over)
+    return rep
+
+
+def test_liveness_report_detection_and_pass():
+    rep = _live_report()
+    assert is_liveness_report(rep) and not is_gated_report(rep)
+    assert validate_report(rep) == []      # dispatches to assert_liveness
+    assert assert_liveness(rep) == []
+
+
+def test_liveness_missing_turns_is_invalid():
+    codes = {v.code for v in assert_liveness(_live_report(turns=[]))}
+    assert codes == {"MISSING_LIVENESS"}
+
+
+def test_liveness_proposer_never_ran():
+    rep = _live_report(turns=[
+        {"blocks": 0, "f_theta_ran": True}, {"blocks": 0, "f_theta_ran": True}])
+    codes = {v.code for v in assert_liveness(rep)}
+    assert "PROPOSER_NEVER_RAN" in codes
+
+
+def test_liveness_missing_blocks_field():
+    rep = _live_report(turns=[{"f_theta_ran": True}])  # no 'blocks'
+    codes = {v.code for v in assert_liveness(rep)}
+    assert "MISSING_LIVENESS" in codes
+
+
+def test_liveness_bool_blocks_not_counted_as_int():
+    # True is an int subclass — must NOT be accepted as a block count.
+    rep = _live_report(turns=[{"blocks": True, "f_theta_ran": True}])
+    codes = {v.code for v in assert_liveness(rep)}
+    assert "MISSING_LIVENESS" in codes
+
+
+def test_liveness_ftheta_not_run_when_intended():
+    rep = _live_report(turns=[
+        {"blocks": 2, "f_theta_ran": True}, {"blocks": 2, "f_theta_ran": False}])
+    codes = {v.code for v in assert_liveness(rep)}
+    assert "FTHETA_NOT_RUN" in codes
+
+
+def test_liveness_ftheta_missing_flag_when_intended():
+    rep = _live_report(turns=[{"blocks": 2}])  # f_theta_intended True but no flag
+    codes = {v.code for v in assert_liveness(rep)}
+    assert "MISSING_LIVENESS" in codes
+
+
+def test_liveness_ftheta_not_required_when_not_intended():
+    # all-MLX fast path: f_θ bypassed by design → no FTHETA_NOT_RUN.
+    rep = _live_report(f_theta_intended=False, turns=[
+        {"blocks": 2, "f_theta_ran": False}, {"blocks": 3, "f_theta_ran": False}])
+    assert assert_liveness(rep) == []
+
+
+def test_liveness_silent_fallback_report_and_turn_level():
+    rep = _live_report(fallbacks_taken=["proposer->ar"])
+    assert any(v.code == "SILENT_FALLBACK" for v in assert_liveness(rep))
+    rep2 = _live_report(turns=[
+        {"blocks": 2, "f_theta_ran": True, "fallbacks_taken": ["f_theta->identity"]}])
+    assert any(v.code == "SILENT_FALLBACK" for v in assert_liveness(rep2))
+
+
 def _valid_report(n: int = MIN_PERF_SAMPLES) -> Dict[str, Any]:
     """A schema-2 report that passes every rule."""
     cross_rows = [

tests/inference_engine/bridge/test_manifest.py

@@ -103,6 +103,9 @@ def test_harness_presets_validate_reports_others_do_not():
     assert gated == {
         "k3-step1-incremental", "k3-step2-fused", "k3-native-baseline",
         "k3-step2-fused-allmlx",
+        # §4 liveness gate runs on-device for the fused-chat presets too:
+        "mlx-kakeya-fused-chat-smoke", "mlx-kakeya-fused-chat-ftheta",
+        "mlx-kakeya-launcher-smoke",
     }