diff --git a/service/ai_gateway_service.py b/service/ai_gateway_service.py index 89ecab43..f760d8ce 100644 --- a/service/ai_gateway_service.py +++ b/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"]) diff --git a/service/autonomy.py b/service/autonomy.py index a6d6981d..6d1ee9a1 100644 --- a/service/autonomy.py +++ b/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, } diff --git a/tests/test_autonomy.py b/tests/test_autonomy.py index 1f6915ca..83314ab7 100644 --- a/tests/test_autonomy.py +++ b/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"])