fix: scope ledger history safety checks

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

Author: Pigbibi

service/ai_gateway_service.py

@@ -556,7 +556,7 @@ def _automation_control_snapshot(
     )
     failure_history_complete = (
         not ledger_unavailable
-        and not bool(retention.get("history_completeness_unknown"))
+        and (not bool(retention.get("history_completeness_unknown")) or repo_history_has_terminal_boundary)
         and (repo_evictions <= 0 or repo_history_has_terminal_boundary)
     )
     execution = decide_automation_execution(

service/automation_run_ledger.py

@@ -351,8 +351,7 @@ def _persist_with_owner_guard_locked(
             self._storage_file_seen = True
             if not disk_read_ok:
                 self._history_completeness_unknown = True
-                self._evict_old_runs_locked()
-                return
+                raise OSError("automation ledger could not be refreshed from disk")
             if guard_run_id:
                 disk_entry = disk_runs.get(guard_run_id)
                 disk_owner = _entry_owner_repository(disk_entry) if isinstance(disk_entry, dict) else ""

tests/test_ai_gateway_automation_control.py

@@ -228,17 +228,20 @@ def test_control_snapshot_fails_closed_after_ledger_eviction(self) -> None:
         self.assertFalse(control["execution"]["failure_history_complete"])
         self.assertEqual(control["execution"]["consecutive_failures"], 1)
 
-    def test_control_snapshot_ignores_other_repo_ledger_eviction(self) -> None:
+    def test_control_snapshot_allows_known_repo_boundary_when_history_unknown(self) -> None:
         health = type("Health", (), {"status": "healthy"})()
         quota = type("Quota", (), {"runtime_status": lambda self, repo: {"status": "ok"}})()
         ledger = type(
             "Ledger",
             (),
             {
                 "snapshot": lambda self, limit=None: {
-                    "runs": [],
+                    "runs": [
+                        {"run_id": "merged-1", "task_state": "merged", "metadata": {"source_repository": "QuantStrategyLab/TargetRepo"}}
+                    ],
                     "summary": {
                         "retention": {
+                            "history_completeness_unknown": True,
                             "may_be_truncated": True,
                             "evicted_runs": 1,
                             "evicted_runs_by_repo": {"quantstrategylab/otherrepo": 1},

tests/test_automation_run_ledger.py

@@ -320,18 +320,19 @@ def test_corrupt_persisted_ledger_marks_history_completeness_unknown(self) -> No
 
         self.assertTrue(snapshot["summary"]["retention"]["history_completeness_unknown"])
 
-    def test_corrupt_disk_read_preserves_local_runs_without_overwriting_file(self) -> None:
+    def test_corrupt_disk_read_fails_record_without_overwriting_file(self) -> None:
         with TemporaryDirectory() as tmp:
             path = Path(tmp) / "automation_runs.json"
             ledger = AutomationRunLedger(max_runs=3, storage_path=path)
             ledger.record("run-1", "queued")
             path.write_text("{not-json", encoding="utf-8")
-            ledger.record("run-2", "queued")
+            with self.assertRaises(OSError):
+                ledger.record("run-2", "queued")
             snapshot = ledger.snapshot(limit=None)
 
             self.assertEqual(path.read_text(encoding="utf-8"), "{not-json")
 
-        self.assertEqual({run["run_id"] for run in snapshot["runs"]}, {"run-1", "run-2"})
+        self.assertEqual({run["run_id"] for run in snapshot["runs"]}, {"run-1"})
         self.assertTrue(snapshot["summary"]["retention"]["history_completeness_unknown"])
 
     def test_update_preserves_control_fields_when_omitted(self) -> None: