fix: fail closed on unreadable automation ledger

Co-Authored-By: Codex <noreply@openai.com>

Author: Pigbibi

service/automation_decision.py

@@ -442,7 +442,12 @@ def decide_automation_execution(
         "failure_history_complete": failure_history_complete,
         "human_review_required": human_review_required,
         "auto_fix_allowed": action == EXECUTION_RUN and effective_mode == MODE_REVIEW_AND_FIX and not human_review_required,
-        "auto_merge_allowed": action == EXECUTION_RUN and effective_autonomy == AUTONOMY_AUTO_MERGE and not human_review_required,
+        "auto_merge_allowed": (
+            action == EXECUTION_RUN
+            and effective_mode == MODE_REVIEW_AND_FIX
+            and effective_autonomy == AUTONOMY_AUTO_MERGE
+            and not human_review_required
+        ),
         "defer": defer,
         "reasons": reasons or ["execution allowed"],
     }

service/automation_run_ledger.py

@@ -225,7 +225,10 @@ def __init__(
         self._load_from_disk()
 
     def _load_from_disk(self) -> None:
-        if self._storage_path is None or not self._storage_path.exists():
+        if self._storage_path is None:
+            return
+        if not self._storage_path.exists():
+            self._history_completeness_unknown = True
             return
         runs, sequence, evicted_runs, evicted_runs_by_repo, history_unknown = self._read_from_disk_unlocked()
         with self._lock:
@@ -238,14 +241,14 @@ def _load_from_disk(self) -> None:
 
     def _read_from_disk_unlocked(self) -> tuple[dict[str, dict[str, Any]], int, int, dict[str, int], bool]:
         if self._storage_path is None or not self._storage_path.exists():
-            return {}, 0, 0, {}, False
+            return {}, 0, 0, {}, self._storage_path is not None
         try:
             payload = json.loads(self._storage_path.read_text(encoding="utf-8"))
         except (OSError, json.JSONDecodeError):
-            return {}, 0, 0, {}, False
+            return {}, 0, 0, {}, True
         runs = payload.get("runs") if isinstance(payload, dict) else None
         if not isinstance(runs, dict):
-            return {}, 0, 0, {}, False
+            return {}, 0, 0, {}, True
         clean_runs = {str(key): value for key, value in runs.items() if isinstance(value, dict)}
         sequence = _safe_int(payload.get("sequence"), len(clean_runs))
         history_unknown = "evicted_runs" not in payload or "evicted_runs_by_repo" not in payload
@@ -278,7 +281,10 @@ def _persist_locked(self) -> None:
         self._persist_with_owner_guard_locked()
 
     def _refresh_from_disk_locked(self) -> None:
-        if self._storage_path is None or not self._storage_path.exists():
+        if self._storage_path is None:
+            return
+        if not self._storage_path.exists():
+            self._history_completeness_unknown = True
             return
         lock_handle = None
         try:

tests/test_ai_gateway_service_get_routes.py

@@ -672,7 +672,7 @@ def test_automation_triage_reports_retryable_incident(self) -> None:
                     self.assertTrue(triage["retry_allowed"])
                     self.assertEqual(triage["recommended_action"], "retry")
                     self.assertEqual(triage["file_risk"], "low")
-                    self.assertEqual(triage["control"]["execution"]["action"], "review_only")
+                    self.assertEqual(triage["control"]["execution"]["action"], "human_review")
                     self.assertIn("run_id=incident-123", triage["summary"])
                 finally:
                     server.shutdown()

tests/test_automation_decision.py

@@ -82,6 +82,22 @@ def test_auto_merge_requires_matching_repo_autonomy(self) -> None:
         self.assertEqual(result["effective_autonomy"], "auto_merge")
         self.assertTrue(result["auto_merge_allowed"])
 
+    def test_auto_merge_is_disabled_when_effective_mode_is_review_only(self) -> None:
+        result = decide_automation_execution(
+            repo="QuantStrategyLab/AIAuditBridge",
+            requested_mode="auto_merge",
+            control_action=CONTROL_CONTINUE,
+            service_health="degraded",
+            quota_status="ok",
+            org_health_status="ok",
+            policy={"default": {"max_autonomy": "auto_merge"}},
+        )
+
+        self.assertEqual(result["action"], EXECUTION_RUN)
+        self.assertEqual(result["effective_mode"], MODE_REVIEW_ONLY)
+        self.assertFalse(result["auto_fix_allowed"])
+        self.assertFalse(result["auto_merge_allowed"])
+
     def test_degraded_health_forces_review_only(self) -> None:
         result = decide_automation_execution(
             repo="QuantStrategyLab/AIAuditBridge",

tests/test_automation_run_ledger.py

@@ -248,6 +248,22 @@ def test_pre_migration_ledger_marks_history_completeness_unknown(self) -> None:
 
         self.assertTrue(snapshot["summary"]["retention"]["history_completeness_unknown"])
 
+    def test_missing_persisted_ledger_marks_history_completeness_unknown(self) -> None:
+        with TemporaryDirectory() as tmp:
+            ledger = AutomationRunLedger(max_runs=2, storage_path=Path(tmp) / "missing.json")
+            snapshot = ledger.snapshot(limit=None)
+
+        self.assertTrue(snapshot["summary"]["retention"]["history_completeness_unknown"])
+
+    def test_corrupt_persisted_ledger_marks_history_completeness_unknown(self) -> None:
+        with TemporaryDirectory() as tmp:
+            path = Path(tmp) / "automation_runs.json"
+            path.write_text("{not-json", encoding="utf-8")
+            ledger = AutomationRunLedger(max_runs=2, storage_path=path)
+            snapshot = ledger.snapshot(limit=None)
+
+        self.assertTrue(snapshot["summary"]["retention"]["history_completeness_unknown"])
+
     def test_update_preserves_control_fields_when_omitted(self) -> None:
         self.ledger.record(
             "run-1",