fix: fail closed automation policy loading

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

Author: Pigbibi

docs/async_service_deployment.md

@@ -72,8 +72,9 @@ bash scripts/deploy_codex_audit_service.sh deploy
 
 The job directory should be owned by the service user and mode `0700`.
 The deploy script points `CODEX_AUDIT_SERVICE_EXECUTION_POLICY_PATH` to
-`${CODEX_AUDIT_SERVICE_JOB_DIR}/execution_policy.json`. If present, this
-service-owned file can cap repo autonomy without trusting the reviewed checkout:
+`${CODEX_AUDIT_SERVICE_JOB_DIR}/execution_policy.json` and creates a conservative
+default file if it is missing. This service-owned file can cap repo autonomy
+without trusting the reviewed checkout:
 
 ```json
 {
@@ -91,6 +92,9 @@ service-owned file can cap repo autonomy without trusting the reviewed checkout:
 }
 ```
 
+If this configured file later becomes unreadable or malformed, the service
+fails closed for execution decisions until the file is repaired.
+
 The service should rely on an authenticated Codex CLI session and must not
 inject OpenAI/Codex API keys into the Codex subprocess.
 With `CODEX_AUDIT_SERVICE_CODEX_ACCOUNT_USAGE=1`, `/v1/ai/quota` includes a

scripts/deploy_codex_audit_service.sh

@@ -214,6 +214,28 @@ write_admin_env_file_if_needed() {
   trap - RETURN
 }
 
+write_default_execution_policy_if_missing() {
+  local owner="$1"
+  local policy_path="${JOB_DIR}/execution_policy.json"
+  if [ -e "$policy_path" ]; then
+    return
+  fi
+  local tmp
+  tmp="$(mktemp)"
+  cat >"$tmp" <<'EOF_POLICY'
+{
+  "default": {
+    "max_autonomy": "auto_pr",
+    "max_consecutive_failures": 3,
+    "low_cost_model": "gpt-5.4-mini"
+  },
+  "repositories": {}
+}
+EOF_POLICY
+  sudo install -m 0600 -o "$owner" -g "$owner" "$tmp" "$policy_path"
+  rm -f "$tmp"
+}
+
 write_audit_service_unit() {
   local runner_user runner_home
   runner_user="$(id -un)"
@@ -510,6 +532,7 @@ deploy() {
   install_file "scripts/codex_audit_service.py" "${DEPLOY_DIR}/scripts/codex_audit_service.py" "0755"
   install_service_package
   sudo install -d -m 0700 -o "$runner_user" -g "$runner_user" "$JOB_DIR"
+  write_default_execution_policy_if_missing "$runner_user"
   write_admin_env_file_if_needed
   write_audit_service_unit
   write_managed_audit_service_dropin

service/ai_gateway_service.py

@@ -35,6 +35,7 @@
 
 from service.auth import authenticate
 from service.contracts import (
+    MODE_REVIEW_AND_FIX,
     MODE_REVIEW_ONLY,
     TASK_ANALYZE,
     TASK_EXECUTE,
@@ -508,7 +509,7 @@ def _public_job_payload(job: dict[str, Any]) -> dict[str, object]:
     return payload
 
 
-def _automation_control_snapshot(repo: str, *, task_name: str = "", requested_mode: str = MODE_REVIEW_ONLY) -> dict[str, Any]:
+def _automation_control_snapshot(repo: str, *, task_name: str = "", requested_mode: str = MODE_REVIEW_AND_FIX) -> dict[str, Any]:
     try:
         org_health = read_org_health()
     except Exception:
@@ -520,8 +521,10 @@ def _automation_control_snapshot(repo: str, *, task_name: str = "", requested_mo
     control = suggest_control_action(get_health_monitor().status, quota_status, org_health)
     try:
         recent_runs = get_automation_run_ledger().snapshot(limit=None)["runs"]
+        ledger_unavailable = False
     except Exception:
         recent_runs = []
+        ledger_unavailable = True
     execution = decide_automation_execution(
         repo=repo or "unknown",
         task_name=task_name,
@@ -533,6 +536,14 @@ def _automation_control_snapshot(repo: str, *, task_name: str = "", requested_mo
         recent_runs=recent_runs,
         policy=load_execution_policy(),
     )
+    if ledger_unavailable:
+        execution["action"] = EXECUTION_HUMAN_REVIEW
+        execution["effective_mode"] = MODE_REVIEW_ONLY
+        execution["human_review_required"] = True
+        execution["auto_fix_allowed"] = False
+        reasons = execution.get("reasons") if isinstance(execution.get("reasons"), list) else []
+        reasons.append("automation ledger unavailable; forcing human review")
+        execution["reasons"] = reasons
     original_action = str(control.get("action") or CONTROL_REVIEW_ONLY)
     strict_action = original_action
     if execution.get("action") == EXECUTION_HUMAN_REVIEW:
@@ -1459,7 +1470,7 @@ def _handle_automation_control(self) -> None:
             else:
                 repo = claims_repo
         repo = repo or "unknown"
-        mode = str(params.get("mode", [MODE_REVIEW_ONLY])[0] or MODE_REVIEW_ONLY)
+        mode = str(params.get("mode", [MODE_REVIEW_AND_FIX])[0] or MODE_REVIEW_AND_FIX)
         _json_response(self, HTTPStatus.OK, {"status": "ok", "control": _automation_control_snapshot(repo, requested_mode=mode)})
 
     def _handle_automation_triage(self, claims: dict[str, Any], payload: dict[str, Any]) -> None:
@@ -1541,7 +1552,7 @@ def _handle_record_automation_run(self, claims: dict[str, Any], payload: dict[st
         control = _automation_control_snapshot(
             repo,
             task_name=str(payload.get("task") or payload.get("task_name") or ""),
-            requested_mode=str(payload.get("mode") or MODE_REVIEW_ONLY),
+            requested_mode=str(payload.get("mode") or MODE_REVIEW_AND_FIX),
         )
         metadata = payload.get("metadata") if isinstance(payload.get("metadata"), dict) else {}
         ledger = get_automation_run_ledger()

service/automation_decision.py

@@ -29,6 +29,7 @@
 DEFAULT_MAX_CONSECUTIVE_FAILURES = 3
 DEFAULT_LOW_COST_MODEL = "gpt-5.4-mini"
 EXECUTION_POLICY_PATH_ENV = "CODEX_AUDIT_SERVICE_EXECUTION_POLICY_PATH"
+POLICY_LOAD_ERROR_KEY = "_load_error"
 QUOTA_STATUS_SEVERITY = {
     "ok": 0,
     "healthy": 0,
@@ -83,18 +84,32 @@ def _repo_from_run(run: dict[str, Any]) -> str:
     return str(metadata.get("source_repository") or metadata.get("repository") or "")
 
 
+def _fail_closed_policy(reason: str) -> dict[str, Any]:
+    return {
+        POLICY_LOAD_ERROR_KEY: reason,
+        "default": {
+            "max_autonomy": AUTONOMY_MANUAL,
+            "max_consecutive_failures": 1,
+        },
+    }
+
+
 def load_execution_policy(path: Path | None = None) -> dict[str, Any]:
     """Load service-owned execution policy for repo autonomy thresholds."""
     if path is None:
         configured = os.environ.get(EXECUTION_POLICY_PATH_ENV, "").strip()
         if not configured:
             return {}
         path = Path(configured).expanduser()
+    if not path.exists():
+        return _fail_closed_policy("execution policy file is unavailable")
     try:
         payload = json.loads(path.read_text(encoding="utf-8"))
     except (OSError, json.JSONDecodeError):
-        return {}
-    return payload if isinstance(payload, dict) else {}
+        return _fail_closed_policy("execution policy file is unreadable")
+    if not isinstance(payload, dict):
+        return _fail_closed_policy("execution policy file is invalid")
+    return payload
 
 
 def repo_execution_policy(repo: str, policy: dict[str, Any] | None = None) -> dict[str, Any]:
@@ -146,6 +161,7 @@ def decide_automation_execution(
 ) -> dict[str, Any]:
     """Produce a safe execution decision from health, quota, failures, and repo policy."""
     repo_policy = repo_execution_policy(repo, policy)
+    policy_load_error = str((policy or {}).get(POLICY_LOAD_ERROR_KEY) or "") if isinstance(policy, dict) else ""
     max_autonomy, autonomy_config_error = _parse_autonomy(repo_policy.get("max_autonomy"), AUTONOMY_AUTO_PR)
     max_failures = _safe_positive_int(repo_policy.get("max_consecutive_failures"), DEFAULT_MAX_CONSECUTIVE_FAILURES)
     low_cost_model = str(repo_policy.get("low_cost_model") or DEFAULT_LOW_COST_MODEL)
@@ -170,6 +186,8 @@ def decide_automation_execution(
         reasons.append(f"repo max autonomy is {max_autonomy}")
     if autonomy_config_error:
         reasons.append(autonomy_config_error)
+    if policy_load_error:
+        reasons.append(policy_load_error)
     if max_autonomy == AUTONOMY_MANUAL:
         action = EXECUTION_HUMAN_REVIEW
 
@@ -199,16 +217,18 @@ def decide_automation_execution(
     if quota in {"low", "constrained"}:
         effective_model = effective_model or low_cost_model or recommend_model(0.0)
         if quota_low_behavior == "defer":
-            action = EXECUTION_DEFER
-            defer = True
+            if action != EXECUTION_HUMAN_REVIEW:
+                action = EXECUTION_DEFER
+                defer = True
             effective_mode = MODE_REVIEW_ONLY
             human_review_required = True
             reasons.append(f"quota status is {quota}; deferring automation")
         else:
             reasons.append(f"quota status is {quota}; recommending low-cost model")
     elif quota in {"exhausted", "blocked"}:
-        action = EXECUTION_DEFER
-        defer = True
+        if action != EXECUTION_HUMAN_REVIEW:
+            action = EXECUTION_DEFER
+            defer = True
         effective_mode = MODE_REVIEW_ONLY
         human_review_required = True
         reasons.append(f"quota status is {quota}; deferring automation")

tests/test_ai_gateway_automation_control.py

@@ -13,6 +13,24 @@
 
 
 class TestAutomationControlSnapshot(unittest.TestCase):
+    def test_control_snapshot_preserves_continue_for_healthy_default_mode(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": []}})()
+
+        with (
+            patch("service.ai_gateway_service.read_org_health", return_value={"status": "ok"}),
+            patch("service.ai_gateway_service.get_health_monitor", return_value=health),
+            patch("service.ai_gateway_service.get_quota_manager", return_value=quota),
+            patch("service.ai_gateway_service.get_automation_run_ledger", return_value=ledger),
+            patch("service.ai_gateway_service.load_execution_policy", return_value={}),
+        ):
+            control = _automation_control_snapshot("QuantStrategyLab/TargetRepo")
+
+        self.assertEqual(control["action"], "continue")
+        self.assertEqual(control["execution"]["effective_mode"], "review_and_fix")
+        self.assertTrue(control["execution"]["auto_fix_allowed"])
+
     def test_control_snapshot_applies_service_owned_execution_policy(self) -> None:
         with TemporaryDirectory() as tmp:
             policy_path = Path(tmp) / "execution_policy.json"
@@ -95,6 +113,23 @@ def snapshot(self, limit=100):
         self.assertEqual(control["execution"]["action"], "human_review")
         self.assertEqual(control["execution"]["consecutive_failures"], 2)
 
+    def test_control_snapshot_fails_closed_when_ledger_is_unavailable(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: (_ for _ in ()).throw(RuntimeError("boom"))})()
+
+        with (
+            patch("service.ai_gateway_service.read_org_health", return_value={"status": "ok"}),
+            patch("service.ai_gateway_service.get_health_monitor", return_value=health),
+            patch("service.ai_gateway_service.get_quota_manager", return_value=quota),
+            patch("service.ai_gateway_service.get_automation_run_ledger", return_value=ledger),
+            patch("service.ai_gateway_service.load_execution_policy", return_value={}),
+        ):
+            control = _automation_control_snapshot("QuantStrategyLab/TargetRepo")
+
+        self.assertEqual(control["action"], "escalate")
+        self.assertEqual(control["execution"]["action"], "human_review")
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_automation_decision.py

@@ -49,11 +49,11 @@ def test_degraded_health_forces_review_only(self) -> None:
         self.assertFalse(result["auto_fix_allowed"])
         self.assertTrue(result["human_review_required"])
 
-    def test_exhausted_quota_defers_execution(self) -> None:
+    def test_exhausted_quota_defers_execution_without_stronger_guard(self) -> None:
         result = decide_automation_execution(
             repo="QuantStrategyLab/AIAuditBridge",
             requested_mode=MODE_REVIEW_AND_FIX,
-            control_action=CONTROL_ESCALATE,
+            control_action=CONTROL_CONTINUE,
             service_health="healthy",
             quota_status={"status": "exhausted"},
             org_health_status="ok",
@@ -63,6 +63,20 @@ def test_exhausted_quota_defers_execution(self) -> None:
         self.assertTrue(result["defer"])
         self.assertEqual(result["effective_mode"], MODE_REVIEW_ONLY)
 
+    def test_human_review_dominates_exhausted_quota_defer(self) -> None:
+        result = decide_automation_execution(
+            repo="QuantStrategyLab/AIAuditBridge",
+            requested_mode=MODE_REVIEW_AND_FIX,
+            control_action=CONTROL_ESCALATE,
+            service_health="healthy",
+            quota_status={"status": "exhausted"},
+            org_health_status="ok",
+        )
+
+        self.assertEqual(result["action"], EXECUTION_HUMAN_REVIEW)
+        self.assertFalse(result["defer"])
+        self.assertEqual(result["effective_mode"], MODE_REVIEW_ONLY)
+
     def test_low_quota_recommends_low_cost_model(self) -> None:
         result = decide_automation_execution(
             repo="QuantStrategyLab/AIAuditBridge",
@@ -168,12 +182,12 @@ def test_invalid_failure_threshold_falls_back_safely(self) -> None:
         self.assertEqual(result["max_consecutive_failures"], 3)
         self.assertEqual(result["action"], EXECUTION_RUN)
 
-    def test_load_execution_policy_ignores_malformed_files(self) -> None:
+    def test_load_execution_policy_fails_closed_for_malformed_files(self) -> None:
         with TemporaryDirectory() as tmp:
             path = Path(tmp) / "policy.json"
             path.write_text("not-json", encoding="utf-8")
 
-            self.assertEqual(load_execution_policy(path), {})
+            self.assertEqual(load_execution_policy(path)["default"]["max_autonomy"], "manual")
 
             path.write_text(json.dumps({"default": {"max_autonomy": "review_only"}}), encoding="utf-8")
             self.assertEqual(load_execution_policy(path)["default"]["max_autonomy"], "review_only")

tests/test_run_monthly_codex_audit.py

@@ -2343,6 +2343,8 @@ def test_vps_deploy_adds_nginx_audit_route_without_router_service(self) -> None:
         self.assertIn("location ^~ /v1/codex-audit/", deploy_script)
         self.assertIn("CODEX_AUDIT_SERVICE_JOB_DIR", deploy_script)
         self.assertIn("CODEX_AUDIT_SERVICE_EXECUTION_POLICY_PATH", deploy_script)
+        self.assertIn("write_default_execution_policy_if_missing", deploy_script)
+        self.assertIn('"max_consecutive_failures": 3', deploy_script)
         self.assertIn("codex_pr_review.yml@refs/pull/*/merge", deploy_script)
         self.assertIn("refs/pull/*/merge", deploy_script)
         self.assertIn("QuantStrategyLab/AIAuditBridge,QuantStrategyLab/CryptoLivePoolPipelines", deploy_script)