feat(r8): operation-scoped lost-event fault-injection seam (disabled by default)

REAL_ASYNC_BRIDGE R8: narrowest reversible fault for lost-event
failure-recovery. scripts/r8_fault.py consults
$RUNE_WR_ROOT/r8_suppress_rules.json keyed by EXACT operation_id;
notify_event_bridge.py records the suppression decision (no .done file,
no bridge POST) and returns, so the waiter's watchdog/authority
recovery path is the only route to terminal truth. Fail-open design:
absent/corrupt rule file means normal delivery. Evidence appended to
r8_fault_log.jsonl. Healthy-path delivery semantics untouched
(guard clause only). Tests: tests/test_r8_fault_injection.py (7).

scripts/notify_event_bridge.py

@@ -42,6 +42,28 @@ def main() -> int:
         "exit_code": 0 if result == "success" else 1,
         "completed_at": time.time(),
     }
+    # R8 lost-event failure-recovery: operation-scoped fault injection.
+    # A rule keyed by the EXACT operation_id suppresses BOTH normal
+    # delivery channels (durable completion file AND bridge POST), so the
+    # waiter's watchdog/failure-detector path is the only route to truth.
+    # Disabled by default (no rule file => zero behavior change) and
+    # observable: every decision is recorded in the R8 fault log.
+    try:
+        import r8_fault  # noqa: E402 — sibling module
+        if r8_fault.should_suppress(payload["operation_id"]):
+            r8_fault.record(payload["operation_id"], "suppressed",
+                            {"sha": sha, "exit_code": payload["exit_code"],
+                             "source_terminal_at": payload["completed_at"]})
+            print("notify-event-bridge: R8 fault injection — completion "
+                  f"event intentionally suppressed for "
+                  f"{payload['operation_id']} (no .done file, no POST); "
+                  "watchdog authority recovery is the expected path",
+                  flush=True)
+            return 0
+        r8_fault.record(payload["operation_id"], "delivered_normally",
+                        {"sha": sha, "rule_active": False})
+    except Exception:  # noqa: BLE001 — fault seam must never break delivery
+        pass
     # R3 durable delivery: write the completion file FIRST — it works even
     # if the bridge daemon is down, and the atomic rename means any waiter
     # (inotify-watching or starting later) sees a complete file or none.

scripts/r8_fault.py

@@ -0,0 +1,82 @@
+#!/usr/bin/env python3
+"""r8_fault.py — REAL_ASYNC_BRIDGE R8: operation-scoped lost-event fault
+injection (evidence-only harness).
+
+DESIGN CONTRACT (R8 §3):
+  disabled by default          no rule file -> zero behavior change
+  operation-scoped             rule keyed by EXACT operation_id
+  reversible                   delete the rule file / rule entry
+  observable                   every decision appended to a JSONL log
+  cannot affect other waits    rule file checked by operation_id only
+
+The ONLY consumer is scripts/notify_event_bridge.py: before normal
+delivery it consults should_suppress(operation_id). Suppression means:
+  - NO durable completion file is written
+  - NO bridge POST (no /process-event wake)
+  - the suppression decision is recorded as R8 fault evidence
+Nothing else in the stack reads the rule. The daemon, waitstore,
+authority verification, and completion-file machinery are untouched.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import time
+from pathlib import Path
+
+RUNE_WR_ROOT = os.environ.get("RUNE_WR_ROOT") or os.environ.get("WR_ROOT") or "/mnt/workdrive/wr-bridge"
+RULES_PATH = Path(RUNE_WR_ROOT) / "r8_suppress_rules.json"
+FAULT_LOG = Path(RUNE_WR_ROOT) / "r8_fault_log.jsonl"
+
+REASON = "event intentionally suppressed for R8 lost-event failure-recovery"
+
+
+def _rules() -> dict:
+    """Read the rule file. Absent/unreadable => no suppression (fail-open
+    to NORMAL delivery; the fault must never take the healthy path down)."""
+    try:
+        return json.loads(RULES_PATH.read_text())
+    except Exception:  # noqa: BLE001 — absent rule file is the default state
+        return {}
+
+
+def lookup(operation_id: str) -> dict | None:
+    """Return the active suppression rule for EXACTLY this operation_id,
+    or None. Expired rules are ignored (not deleted — deletion is the
+    operator's explicit disable action)."""
+    spec = _rules()
+    if not spec.get("enabled"):
+        return None
+    rule = spec.get("operations", {}).get(operation_id)
+    if not rule:
+        return None
+    expires = rule.get("expires_at")
+    if expires:
+        try:
+            if time.time() > time.mktime(time.strptime(expires, "%Y-%m-%dT%H:%M:%SZ")) - time.timezone:
+                return None
+        except ValueError:
+            return None
+    return rule
+
+
+def should_suppress(operation_id: str) -> bool:
+    return lookup(operation_id) is not None
+
+
+def record(operation_id: str, stage: str, extra: dict | None = None) -> None:
+    """Append a durable fault-evidence line (best-effort, never raises)."""
+    try:
+        entry = {
+            "at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+            "operation_id": operation_id,
+            "reason": REASON,
+            "stage": stage,
+            **(extra or {}),
+        }
+        FAULT_LOG.parent.mkdir(parents=True, exist_ok=True)
+        with FAULT_LOG.open("a") as f:
+            f.write(json.dumps(entry, sort_keys=True) + "\n")
+    except Exception:  # noqa: BLE001 — evidence logging must never break CI
+        pass

tests/test_r8_fault_injection.py

@@ -0,0 +1,147 @@
+"""R8 fault-injection seam tests: operation-scoped, disabled-by-default,
+reversible, observable. The seam lives ONLY in notify_event_bridge.py via
+scripts/r8_fault.py; everything else must be untouched.
+"""
+
+from __future__ import annotations
+
+import importlib
+import json
+import sys
+import time
+from pathlib import Path
+
+import pytest
+
+SCRIPTS = Path(__file__).resolve().parent.parent / "scripts"
+sys.path.insert(0, str(SCRIPTS))
+r8_fault = importlib.import_module("r8_fault")
+
+
+@pytest.fixture()
+def isolated_root(tmp_path, monkeypatch):
+    monkeypatch.setattr(r8_fault, "RUNE_WR_ROOT", str(tmp_path))
+    monkeypatch.setattr(r8_fault, "RULES_PATH", tmp_path / "r8_suppress_rules.json")
+    monkeypatch.setattr(r8_fault, "FAULT_LOG", tmp_path / "r8_fault_log.jsonl")
+    return tmp_path
+
+
+def test_disabled_by_default(isolated_root):
+    # no rule file at all -> no suppression, no crash
+    assert r8_fault.should_suppress("ci-abc") is False
+    assert r8_fault.lookup("ci-abc") is None
+
+
+def test_operation_scoped(isolated_root):
+    (isolated_root / "r8_suppress_rules.json").write_text(
+        json.dumps(
+            {
+                "enabled": True,
+                "operations": {"ci-target": {"reason": r8_fault.REASON}},
+            }
+        )
+    )
+    assert r8_fault.should_suppress("ci-target") is True
+    # any other operation_id is NEVER affected
+    assert r8_fault.should_suppress("ci-other") is False
+    assert r8_fault.should_suppress("ci-targe") is False
+    assert r8_fault.should_suppress("ci-targetx") is False
+
+
+def test_disabled_flag_wins(isolated_root):
+    (isolated_root / "r8_suppress_rules.json").write_text(
+        json.dumps(
+            {
+                "enabled": False,
+                "operations": {"ci-target": {"reason": r8_fault.REASON}},
+            }
+        )
+    )
+    assert r8_fault.should_suppress("ci-target") is False
+
+
+def test_expired_rule_ignored(isolated_root):
+    past = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(time.time() - 10))
+    (isolated_root / "r8_suppress_rules.json").write_text(
+        json.dumps(
+            {
+                "enabled": True,
+                "operations": {"ci-target": {"reason": r8_fault.REASON, "expires_at": past}},
+            }
+        )
+    )
+    assert r8_fault.should_suppress("ci-target") is False
+
+
+def test_corrupt_rule_file_fails_open(isolated_root):
+    # fail-open to NORMAL delivery — the fault seam must never take the
+    # healthy path down
+    (isolated_root / "r8_suppress_rules.json").write_text("{broken")
+    assert r8_fault.should_suppress("ci-target") is False
+
+
+def test_evidence_recorded(isolated_root):
+    (isolated_root / "r8_suppress_rules.json").write_text(
+        json.dumps(
+            {
+                "enabled": True,
+                "operations": {"ci-target": {"reason": r8_fault.REASON}},
+            }
+        )
+    )
+    r8_fault.record("ci-target", "suppressed", {"sha": "abc"})
+    log = (isolated_root / "r8_fault_log.jsonl").read_text().strip().splitlines()
+    entry = json.loads(log[-1])
+    assert entry["operation_id"] == "ci-target"
+    assert entry["stage"] == "suppressed"
+    assert entry["reason"] == r8_fault.REASON
+    assert "at" in entry
+
+
+def test_notify_bridge_suppresses_only_targeted_op(monkeypatch, tmp_path):
+    """Producer-level test: with an active rule for ci-aaa, main() must
+    write NO completion file for ci-aaa and must NOT POST; a run for
+    ci-bbb with the same rule file still delivers normally."""
+    import completion_file as cf
+    import notify_event_bridge as neb
+
+    wr = tmp_path / "wr"
+    monkeypatch.setenv("RUNE_WR_ROOT", str(wr))
+    importlib.reload(r8_fault)
+    wr.mkdir(parents=True, exist_ok=True)
+    (wr / "r8_suppress_rules.json").write_text(
+        json.dumps(
+            {
+                "enabled": True,
+                "operations": {"ci-aaa": {"reason": r8_fault.REASON}},
+            }
+        )
+    )
+    monkeypatch.setattr(cf, "_wr_bridge", lambda: wr)
+    posted = []
+    monkeypatch.setattr(
+        "urllib.request.urlopen",
+        lambda req, timeout=5: (
+            posted.append(req)
+            or type(
+                "R",
+                (),
+                {"read": lambda s: b'{"matched": 0}', "__enter__": lambda s: s, "__exit__": lambda s, *a: False},
+            )()
+        ),
+    )
+
+    # suppressed operation: no file, no POST
+    monkeypatch.setenv("CI_SHA", "aaa")
+    monkeypatch.setenv("CI_RESULT", "success")
+    assert neb.main() == 0
+    assert cf.completion_exists("ci-aaa") is False
+    assert posted == []
+    log = (wr / "r8_fault_log.jsonl").read_text().strip().splitlines()
+    assert json.loads(log[-1])["stage"] == "suppressed"
+
+    # untargeted operation: normal delivery still intact
+    monkeypatch.setenv("CI_SHA", "bbb")
+    assert neb.main() == 0
+    assert cf.completion_exists("ci-bbb") is True
+    assert len(posted) == 1