feat: guard autonomy decisions with shared policy

service/ai_gateway_service.py

@@ -44,6 +44,7 @@
 from service.adapters.llm_adapter import LlmAdapter
 from service.adapters.codex_adapter import CodexAdapter
 from service.autonomy import (
+    load_autonomy_policy,
     recommended_action as compute_recommended_action,
 )
 from service.feedback import (
@@ -905,7 +906,11 @@ def _handle_review(self, payload: dict[str, Any]) -> None:
         # Autonomy decision: confidence + file risk → recommended action
         repo = str(payload.get("source_repository") or "")
         action = compute_recommended_action(
-            results, changed_paths, repo=repo if repo else None,
+            results,
+            changed_paths,
+            repo=repo if repo else None,
+            policy=load_autonomy_policy(),
+            health_status=get_health_monitor().status,
         )
         _audit_log("review_completed", consensus=consensus, all_success=all_ok,
                    action=action["action"], confidence=action["confidence"], risk=action["risk"])

service/autonomy.py

@@ -31,6 +31,7 @@
 
 import json
 import os
+import re
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any
@@ -44,6 +45,7 @@
 
 # ordered by increasing autonomy
 ACTION_ORDER = (ACTION_ESCALATE, ACTION_AUTO_NOTIFY, ACTION_AUTO_PR, ACTION_AUTO_MERGE)
+ACTION_RANK = {action: index for index, action in enumerate(ACTION_ORDER)}
 
 # ── risk tiers ──────────────────────────────────────────────────────────
 
@@ -88,6 +90,33 @@
     "LICENSE",
     ".gitignore",
 })
+CRITICAL_EXACT = frozenset({
+    ".github/codex_auto_merge_policy.json",
+})
+REPO_ROOT = Path(__file__).resolve().parents[1]
+AUTONOMY_POLICY_PATH_ENV = "CODEX_AUDIT_SERVICE_AUTONOMY_POLICY_PATH"
+
+
+def load_autonomy_policy(path: Path | None = None) -> dict[str, Any]:
+    """Load the shared autonomy policy from a trusted service-owned path.
+
+    The service must not read policy rules from the untrusted PR checkout being
+    reviewed. Set CODEX_AUDIT_SERVICE_AUTONOMY_POLICY_PATH to a deployment-owned
+    file or pass an explicit path in tests/tools. Missing or malformed files
+    fall back to the built-in conservative classifier.
+    """
+    if path is None:
+        env_path = os.environ.get(AUTONOMY_POLICY_PATH_ENV, "").strip()
+        if not env_path:
+            return {}
+        path = Path(env_path)
+    if not path.exists():
+        return {}
+    try:
+        payload = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError):
+        return {}
+    return payload if isinstance(payload, dict) else {}
 
 
 @dataclass(frozen=True)
@@ -150,17 +179,49 @@ def get_matrix(self, repo: str | None = None) -> list[tuple[str, float, str]]:
         return self.decision_matrix
 
 
-def classify_file_risk(path: str) -> str:
+def _policy_matches(path: str, rule: dict[str, Any]) -> bool:
+    exact = rule.get("exact")
+    if isinstance(exact, list) and path in {str(item) for item in exact}:
+        return True
+    prefixes = rule.get("prefixes")
+    if isinstance(prefixes, list) and any(path.startswith(str(prefix)) for prefix in prefixes):
+        return True
+    return False
+
+
+def _blocked_by_policy(path: str, policy: dict[str, Any] | None) -> bool:
+    if path in CRITICAL_EXACT:
+        return True
+    patterns = (policy or {}).get("blocked_path_patterns") if isinstance(policy, dict) else None
+    raw_patterns = list(CRITICAL_PATTERNS)
+    if isinstance(patterns, list):
+        raw_patterns.extend(pattern for pattern in patterns if isinstance(pattern, str))
+    for pattern in raw_patterns:
+        if not isinstance(pattern, str) or not pattern.strip():
+            continue
+        try:
+            if re.search(pattern, path, flags=re.IGNORECASE):
+                return True
+        except re.error:
+            continue
+    return False
+
+
+def classify_file_risk(path: str, *, policy: dict[str, Any] | None = None) -> str:
     """Classify a changed file path into a risk tier.
 
-    Mirrors the logic in codex_auto_merge_policy.json risk_policy.
+    Mirrors ``codex_auto_merge_policy.json`` when present, then falls back to
+    the built-in conservative rules.
     """
-    import re as _re
+    if _blocked_by_policy(path, policy):
+        return RISK_CRITICAL
 
-    # critical: secrets, credentials, keys
-    for pattern in CRITICAL_PATTERNS:
-        if _re.search(pattern, path):
-            return RISK_CRITICAL
+    risk_policy = (policy or {}).get("risk_policy") if isinstance(policy, dict) else None
+    if isinstance(risk_policy, dict):
+        for tier in (RISK_CRITICAL, RISK_HIGH, RISK_MEDIUM, RISK_LOW):
+            rule = risk_policy.get(tier)
+            if isinstance(rule, dict) and _policy_matches(path, rule):
+                return tier
 
     # low: exact match
     if path in LOW_RISK_EXACT:
@@ -185,14 +246,14 @@ def classify_file_risk(path: str) -> str:
     return RISK_MEDIUM
 
 
-def classify_changes_risk(changed_paths: list[str]) -> str:
+def classify_changes_risk(changed_paths: list[str], *, policy: dict[str, Any] | None = None) -> str:
     """Classify the overall risk of a set of changed file paths.
 
     Returns the highest risk tier among all changed files.
     """
     if not changed_paths:
         return RISK_LOW
-    tiers = {classify_file_risk(p) for p in changed_paths}
+    tiers = {classify_file_risk(p, policy=policy) for p in changed_paths}
     for tier in (RISK_CRITICAL, RISK_HIGH, RISK_MEDIUM, RISK_LOW):
         if tier in tiers:
             return tier
@@ -231,6 +292,45 @@ def decide_action(
     return ACTION_ESCALATE
 
 
+def _cap_action(action: str, maximum: str) -> str:
+    if ACTION_RANK.get(action, 0) > ACTION_RANK.get(maximum, 0):
+        return maximum
+    return action
+
+
+def apply_runtime_guards(
+    action: str,
+    *,
+    health_status: str | None = None,
+    quota_status: str | None = None,
+) -> tuple[str, list[str]]:
+    """Downgrade autonomy based on runtime health/quota state."""
+    guarded_action = action
+    guards: list[str] = []
+    health = (health_status or "healthy").strip().lower()
+    quota = (quota_status or "ok").strip().lower()
+
+    if health == "unhealthy":
+        guarded_action = ACTION_ESCALATE
+        guards.append("service health is unhealthy; forcing human review")
+    elif health == "degraded":
+        capped = _cap_action(guarded_action, ACTION_AUTO_PR)
+        if capped != guarded_action:
+            guards.append("service health is degraded; auto-merge capped at auto-pr")
+        guarded_action = capped
+
+    if quota in {"exhausted", "blocked"}:
+        guarded_action = ACTION_ESCALATE
+        guards.append(f"quota status is {quota}; forcing human review")
+    elif quota in {"low", "constrained"}:
+        capped = _cap_action(guarded_action, ACTION_AUTO_PR)
+        if capped != guarded_action:
+            guards.append(f"quota status is {quota}; auto-merge capped at auto-pr")
+        guarded_action = capped
+
+    return guarded_action, guards
+
+
 def extract_confidence(verdicts: list[dict[str, Any]]) -> float:
     """Extract an aggregated confidence score from a list of reviewer verdicts.
 
@@ -255,6 +355,9 @@ def recommended_action(
     *,
     config: AutonomyConfig | None = None,
     repo: str | None = None,
+    policy: dict[str, Any] | None = None,
+    health_status: str | None = None,
+    quota_status: str | None = None,
 ) -> dict[str, Any]:
     """Compute the recommended autonomous action from AI verdicts and file risks.
 
@@ -265,8 +368,10 @@ def recommended_action(
         reason: Human-readable explanation.
     """
     confidence = extract_confidence(verdicts)
-    risk = classify_changes_risk(changed_paths or [])
-    action = decide_action(confidence, risk, config=config, repo=repo)
+    active_policy = policy if policy is not None else load_autonomy_policy()
+    risk = classify_changes_risk(changed_paths or [], policy=active_policy)
+    initial_action = decide_action(confidence, risk, config=config, repo=repo)
+    action, runtime_guards = apply_runtime_guards(initial_action, health_status=health_status, quota_status=quota_status)
 
     reasons = {
         (ACTION_ESCALATE, RISK_CRITICAL): "Critical files changed — always escalates to human review",
@@ -283,7 +388,12 @@ def recommended_action(
 
     return {
         "action": action,
+        "initial_action": initial_action,
         "confidence": confidence,
         "risk": risk,
         "reason": reason,
+        "human_review_required": action == ACTION_ESCALATE,
+        "auto_merge_allowed": action == ACTION_AUTO_MERGE,
+        "runtime_guards": runtime_guards,
+        "policy_version": active_policy.get("version") if isinstance(active_policy, dict) else None,
     }

tests/test_autonomy.py

@@ -2,16 +2,23 @@
 
 from __future__ import annotations
 
+import os
+from pathlib import Path
+import tempfile
 import unittest
 
 from service.autonomy import (
     ACTION_AUTO_MERGE,
     ACTION_AUTO_PR,
     ACTION_ESCALATE,
+    classify_file_risk,
     RISK_HIGH,
     RISK_LOW,
     RISK_MEDIUM,
+    RISK_CRITICAL,
+    AUTONOMY_POLICY_PATH_ENV,
     DEFAULT_DECISION_MATRIX,
+    load_autonomy_policy,
     recommended_action,
 )
 
@@ -38,3 +45,78 @@ def test_default_matrix_reflects_safer_thresholds(self) -> None:
         self.assertIn((RISK_MEDIUM, 0.70, ACTION_AUTO_PR), DEFAULT_DECISION_MATRIX)
         self.assertIn((RISK_HIGH, 0.85, ACTION_AUTO_PR), DEFAULT_DECISION_MATRIX)
         self.assertNotIn((RISK_HIGH, 0.95, ACTION_AUTO_MERGE), DEFAULT_DECISION_MATRIX)
+
+    def test_shared_policy_classifies_blocked_and_low_risk_paths(self) -> None:
+        policy = {
+            "version": 7,
+            "blocked_path_patterns": [r"(^|/).*token.*$"],
+            "risk_policy": {
+                "low": {"prefixes": ["docs/"], "exact": ["CHANGELOG.md"]},
+                "high": {"prefixes": ["src/quant_"]},
+            },
+        }
+
+        self.assertEqual(classify_file_risk("docs/runbook.md", policy=policy), RISK_LOW)
+        self.assertEqual(classify_file_risk("CHANGELOG.md", policy=policy), RISK_LOW)
+        self.assertEqual(classify_file_risk("src/quant_alpha.py", policy=policy), RISK_HIGH)
+        self.assertEqual(classify_file_risk("config/token.txt", policy=policy), RISK_CRITICAL)
+        self.assertEqual(classify_file_risk("config/secret.pem", policy=policy), RISK_CRITICAL)
+
+    def test_policy_load_does_not_depend_on_cwd(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            policy_path = Path(__file__).resolve().parents[1] / ".github" / "codex_auto_merge_policy.json"
+            old_cwd = os.getcwd()
+            old_env = os.environ.get(AUTONOMY_POLICY_PATH_ENV)
+            os.environ[AUTONOMY_POLICY_PATH_ENV] = str(policy_path)
+            try:
+                os.chdir(tmp)
+                policy = load_autonomy_policy()
+            finally:
+                os.chdir(old_cwd)
+                if old_env is None:
+                    os.environ.pop(AUTONOMY_POLICY_PATH_ENV, None)
+                else:
+                    os.environ[AUTONOMY_POLICY_PATH_ENV] = old_env
+
+        self.assertEqual(policy.get("version"), 1)
+
+    def test_policy_is_not_loaded_from_repo_by_default(self) -> None:
+        old_env = os.environ.pop(AUTONOMY_POLICY_PATH_ENV, None)
+        try:
+            self.assertEqual(load_autonomy_policy(), {})
+        finally:
+            if old_env is not None:
+                os.environ[AUTONOMY_POLICY_PATH_ENV] = old_env
+
+    def test_autonomy_policy_file_cannot_be_downgraded_by_policy(self) -> None:
+        malicious_policy = {
+            "risk_policy": {
+                "low": {"exact": [".github/codex_auto_merge_policy.json"]},
+            },
+        }
+
+        self.assertEqual(
+            classify_file_risk(".github/codex_auto_merge_policy.json", policy=malicious_policy),
+            RISK_CRITICAL,
+        )
+        result = recommended_action(
+            [{"confidence": 0.99}],
+            [".github/codex_auto_merge_policy.json"],
+            policy=malicious_policy,
+        )
+        self.assertEqual(result["action"], ACTION_ESCALATE)
+
+    def test_degraded_health_caps_auto_merge_to_auto_pr(self) -> None:
+        result = recommended_action([{"confidence": 0.99}], ["docs/runbook.md"], health_status="degraded")
+
+        self.assertEqual(result["initial_action"], ACTION_AUTO_MERGE)
+        self.assertEqual(result["action"], ACTION_AUTO_PR)
+        self.assertFalse(result["auto_merge_allowed"])
+        self.assertTrue(result["runtime_guards"])
+
+    def test_unhealthy_health_forces_human_review(self) -> None:
+        result = recommended_action([{"confidence": 0.99}], ["docs/runbook.md"], health_status="unhealthy")
+
+        self.assertEqual(result["initial_action"], ACTION_AUTO_MERGE)
+        self.assertEqual(result["action"], ACTION_ESCALATE)
+        self.assertTrue(result["human_review_required"])