diff --git a/executive/watchdog.py b/executive/watchdog.py index 7123215..445f90c 100644 --- a/executive/watchdog.py +++ b/executive/watchdog.py @@ -21,6 +21,7 @@ sys.path.insert(0, str(_SCRIPT_DIR.parent)) from executive.judicial_log import JudicialLog, _log_path +from executive.proof import OverrideProof ACTIONS_LOG = "executive_actions.log" OVERRIDE_THRESHOLD = 5 # Alert if overrides in window exceed this @@ -82,6 +83,22 @@ def record_action( f.write(json.dumps(entry, ensure_ascii=False) + "\n") +def _override_proof_error(action: dict) -> Optional[str]: + """Return an error message when override proof is missing or invalid.""" + proof = action.get("override_proof") + if not proof: + return "missing override proof" + + ruling_id = action.get("ruling_id", "") + reason = action.get("override_reason") or action.get("description", "") + try: + if OverrideProof.verify(proof, ruling_id, reason) is None: + return "invalid override proof" + except FileNotFoundError: + return "cannot verify override proof (secret not configured)" + return None + + def _judicial_ruling_ids(log: JudicialLog) -> set[str]: """Extract ruling IDs from judicial log (from source or matter).""" ids = set() @@ -119,13 +136,17 @@ def run_checks() -> tuple[bool, list[str]]: for a in actions: rid = a.get("ruling_id", "") - if rid and rid not in approved and "override" not in a.get("action_type", "").lower(): - # Action references ruling not in judicial log - proof = a.get("override_proof") - if not proof: - alerts.append( - f"Action without judicial approval and no override proof: ruling_id={rid}" - ) + if not rid: + alerts.append("Action without ruling_id reference") + continue + if rid in approved: + continue + + proof_error = _override_proof_error(a) + if proof_error: + alerts.append( + f"Action without judicial approval ({proof_error}): ruling_id={rid}" + ) # (c) Override frequency overrides = [a for a in actions if a.get("override_proof")] diff --git a/litigation/run.py b/litigation/run.py index c86de43..8f5aa34 100644 --- a/litigation/run.py +++ b/litigation/run.py @@ -6,6 +6,7 @@ """ import argparse +import fcntl import os import re import sys @@ -71,21 +72,39 @@ def allocate_case_no(category: str = "DEL", *, deliberation: int = 1) -> str: import yaml year = datetime.now().strftime("%Y") + REGISTRY_PATH.parent.mkdir(parents=True, exist_ok=True) if not REGISTRY_PATH.exists(): - return f"{year}-{category}-001-{deliberation:03d}" - - data = yaml.safe_load(REGISTRY_PATH.read_text(encoding="utf-8")) or {} - year = str(data.get("year", year)) - categories = data.setdefault("categories", {}) - nnn = int(categories.get(category, 1)) - case_no = f"{year}-{category}-{nnn:03d}-{deliberation:03d}" - categories[category] = nnn + 1 - data["categories"] = categories - data["last_updated"] = datetime.now().strftime("%Y-%m-%d") - REGISTRY_PATH.write_text( - yaml.dump(data, default_flow_style=False, sort_keys=False), - encoding="utf-8", - ) + REGISTRY_PATH.write_text( + yaml.dump( + { + "year": int(year), + "categories": {category: 1}, + "last_updated": datetime.now().strftime("%Y-%m-%d"), + }, + default_flow_style=False, + sort_keys=False, + ), + encoding="utf-8", + ) + + with open(REGISTRY_PATH, "r+", encoding="utf-8") as registry_file: + fcntl.flock(registry_file.fileno(), fcntl.LOCK_EX) + try: + data = yaml.safe_load(registry_file.read()) or {} + year = str(data.get("year", year)) + categories = data.setdefault("categories", {}) + nnn = int(categories.get(category, 1)) + case_no = f"{year}-{category}-{nnn:03d}-{deliberation:03d}" + categories[category] = nnn + 1 + data["categories"] = categories + data["last_updated"] = datetime.now().strftime("%Y-%m-%d") + registry_file.seek(0) + registry_file.truncate() + registry_file.write( + yaml.dump(data, default_flow_style=False, sort_keys=False) + ) + finally: + fcntl.flock(registry_file.fileno(), fcntl.LOCK_UN) return case_no diff --git a/tests/test_executive_watchdog.py b/tests/test_executive_watchdog.py index 72e66d3..0529204 100644 --- a/tests/test_executive_watchdog.py +++ b/tests/test_executive_watchdog.py @@ -66,6 +66,112 @@ def test_run_checks_alerts_on_unapproved_action_without_proof( assert any("without judicial approval" in a for a in alerts) +def test_run_checks_alerts_on_override_action_without_proof( + tmp_path, monkeypatch: pytest.MonkeyPatch +) -> None: + log_path = tmp_path / "judicial_decisions.log" + actions_path = tmp_path / "executive_actions.log" + + JudicialLog(log_path=log_path) + + _write_action( + actions_path, + action_type="override", + ruling_id="2026-UNKNOWN-999", + description="rogue override", + ) + + monkeypatch.setattr(watchdog, "JudicialLog", lambda: JudicialLog(log_path=log_path)) + monkeypatch.setattr(watchdog, "_actions_log_path", lambda: actions_path) + + ok, alerts = watchdog.run_checks() + assert ok is False + assert any("missing override proof" in a for a in alerts) + + +def test_run_checks_alerts_on_invalid_override_proof( + tmp_path, monkeypatch: pytest.MonkeyPatch +) -> None: + log_path = tmp_path / "judicial_decisions.log" + actions_path = tmp_path / "executive_actions.log" + secret_path = tmp_path / ".executive_secret" + secret_path.write_bytes(bytes.fromhex("ab" * 32)) + + JudicialLog(log_path=log_path) + + _write_action( + actions_path, + action_type="execute", + ruling_id="2026-UNKNOWN-999", + description="rogue execution", + override_proof="fake-proof", + ) + + monkeypatch.setattr(watchdog, "JudicialLog", lambda: JudicialLog(log_path=log_path)) + monkeypatch.setattr(watchdog, "_actions_log_path", lambda: actions_path) + monkeypatch.setenv("EXECUTIVE_PROOF_SECRET", "0x" + ("ab" * 32)) + + ok, alerts = watchdog.run_checks() + assert ok is False + assert any("invalid override proof" in a for a in alerts) + + +def test_run_checks_accepts_valid_override_proof_for_unapproved_ruling( + tmp_path, monkeypatch: pytest.MonkeyPatch +) -> None: + log_path = tmp_path / "judicial_decisions.log" + actions_path = tmp_path / "executive_actions.log" + secret_path = tmp_path / ".executive_secret" + secret_path.write_bytes(bytes.fromhex("cd" * 32)) + + JudicialLog(log_path=log_path) + reason = "emergency rollback" + proof = OverrideProof.generate( + "2026-UNKNOWN-999", + reason, + timestamp="2026-05-25T12:00:00+00:00", + secret_path=secret_path, + ) + + _write_action( + actions_path, + action_type="override", + ruling_id="2026-UNKNOWN-999", + description=reason, + override_proof=proof, + ) + + monkeypatch.setattr(watchdog, "JudicialLog", lambda: JudicialLog(log_path=log_path)) + monkeypatch.setattr(watchdog, "_actions_log_path", lambda: actions_path) + monkeypatch.setenv("EXECUTIVE_PROOF_SECRET", "0x" + ("cd" * 32)) + + ok, alerts = watchdog.run_checks() + assert ok is True + assert alerts == [] + + +def test_run_checks_alerts_on_missing_ruling_id( + tmp_path, monkeypatch: pytest.MonkeyPatch +) -> None: + log_path = tmp_path / "judicial_decisions.log" + actions_path = tmp_path / "executive_actions.log" + + JudicialLog(log_path=log_path) + _write_action( + actions_path, + action_type="execute", + ruling_id="", + description="deploy without ruling reference", + ) + + monkeypatch.setattr(watchdog, "JudicialLog", lambda: JudicialLog(log_path=log_path)) + monkeypatch.setattr(watchdog, "_actions_log_path", lambda: actions_path) + + ok, alerts = watchdog.run_checks() + assert ok is False + assert any("without ruling_id reference" in a for a in alerts) + + def test_run_checks_alerts_on_tampered_judicial_log( tmp_path, monkeypatch: pytest.MonkeyPatch ) -> None: diff --git a/tests/test_litigation_run.py b/tests/test_litigation_run.py index 8e46431..5d101b8 100644 --- a/tests/test_litigation_run.py +++ b/tests/test_litigation_run.py @@ -76,3 +76,18 @@ def test_save_transcript_filename_suffix_independent_of_case_no( assert path.name.endswith("-1.md") assert "**Case No.:** 2026-DEL-020-001" in content + + +def test_allocate_case_no_initializes_missing_registry( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + registry = tmp_path / "case-registry.yaml" + monkeypatch.setattr(litigation_run, "REGISTRY_PATH", registry) + + first = litigation_run.allocate_case_no("DEL") + second = litigation_run.allocate_case_no("DEL") + + assert first == f"{litigation_run.datetime.now().strftime('%Y')}-DEL-001-001" + assert second == f"{litigation_run.datetime.now().strftime('%Y')}-DEL-002-001" + data = yaml.safe_load(registry.read_text(encoding="utf-8")) + assert data["categories"]["DEL"] == 3