From d7434126d5b4c18e35b02ea559ef1e9f9bae5517 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 5 Jul 2026 06:14:44 +0800 Subject: [PATCH 1/4] feat: guard autonomy decisions with shared policy Co-Authored-By: Codex --- service/ai_gateway_service.py | 7 +- service/autonomy.py | 117 ++++++++++++++++++++++++++++++---- tests/test_autonomy.py | 32 ++++++++++ 3 files changed, 144 insertions(+), 12 deletions(-) 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..0f8abd64 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,22 @@ "LICENSE", ".gitignore", }) +DEFAULT_POLICY_PATH = Path(".github/codex_auto_merge_policy.json") + + +def load_autonomy_policy(path: Path = DEFAULT_POLICY_PATH) -> dict[str, Any]: + """Load the shared repo autonomy policy. + + The policy is intentionally best-effort: malformed or missing files fall + back to the built-in conservative classifier instead of disabling review. + """ + 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 +168,45 @@ 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: + patterns = (policy or {}).get("blocked_path_patterns") if isinstance(policy, dict) else None + raw_patterns = patterns if isinstance(patterns, list) and patterns else list(CRITICAL_PATTERNS) + 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 +231,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 +277,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 +340,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 +353,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 +373,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..b881d4d5 100644 --- a/tests/test_autonomy.py +++ b/tests/test_autonomy.py @@ -8,9 +8,11 @@ ACTION_AUTO_MERGE, ACTION_AUTO_PR, ACTION_ESCALATE, + classify_file_risk, RISK_HIGH, RISK_LOW, RISK_MEDIUM, + RISK_CRITICAL, DEFAULT_DECISION_MATRIX, recommended_action, ) @@ -38,3 +40,33 @@ 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) + + 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"]) From 3acf3a03d7e5a395547963dbb3d3d4b012692583 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 5 Jul 2026 06:18:59 +0800 Subject: [PATCH 2/4] fix: preserve default autonomy safety policy Co-Authored-By: Codex --- service/autonomy.py | 7 +++++-- tests/test_autonomy.py | 15 +++++++++++++++ 2 files changed, 20 insertions(+), 2 deletions(-) diff --git a/service/autonomy.py b/service/autonomy.py index 0f8abd64..a649e497 100644 --- a/service/autonomy.py +++ b/service/autonomy.py @@ -90,7 +90,8 @@ "LICENSE", ".gitignore", }) -DEFAULT_POLICY_PATH = Path(".github/codex_auto_merge_policy.json") +REPO_ROOT = Path(__file__).resolve().parents[1] +DEFAULT_POLICY_PATH = REPO_ROOT / ".github" / "codex_auto_merge_policy.json" def load_autonomy_policy(path: Path = DEFAULT_POLICY_PATH) -> dict[str, Any]: @@ -180,7 +181,9 @@ def _policy_matches(path: str, rule: dict[str, Any]) -> bool: def _blocked_by_policy(path: str, policy: dict[str, Any] | None) -> bool: patterns = (policy or {}).get("blocked_path_patterns") if isinstance(policy, dict) else None - raw_patterns = patterns if isinstance(patterns, list) and patterns else list(CRITICAL_PATTERNS) + 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 diff --git a/tests/test_autonomy.py b/tests/test_autonomy.py index b881d4d5..c0db378b 100644 --- a/tests/test_autonomy.py +++ b/tests/test_autonomy.py @@ -2,6 +2,8 @@ from __future__ import annotations +import os +import tempfile import unittest from service.autonomy import ( @@ -14,6 +16,7 @@ RISK_MEDIUM, RISK_CRITICAL, DEFAULT_DECISION_MATRIX, + load_autonomy_policy, recommended_action, ) @@ -55,6 +58,18 @@ def test_shared_policy_classifies_blocked_and_low_risk_paths(self) -> None: 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: + old_cwd = os.getcwd() + try: + os.chdir(tmp) + policy = load_autonomy_policy() + finally: + os.chdir(old_cwd) + + self.assertEqual(policy.get("version"), 1) def test_degraded_health_caps_auto_merge_to_auto_pr(self) -> None: result = recommended_action([{"confidence": 0.99}], ["docs/runbook.md"], health_status="degraded") From 655d5cc7fc4882c3e0b0f31618cad814f4197dfb Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 5 Jul 2026 06:24:15 +0800 Subject: [PATCH 3/4] chore: retrigger required checks Co-Authored-By: Codex From d150e220f82b8fa9825fc547c6c7009c984ce862 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 5 Jul 2026 06:30:18 +0800 Subject: [PATCH 4/4] fix: require trusted autonomy policy source Co-Authored-By: Codex --- service/autonomy.py | 22 +++++++++++++++++----- tests/test_autonomy.py | 35 +++++++++++++++++++++++++++++++++++ 2 files changed, 52 insertions(+), 5 deletions(-) diff --git a/service/autonomy.py b/service/autonomy.py index a649e497..6d1ee9a1 100644 --- a/service/autonomy.py +++ b/service/autonomy.py @@ -90,16 +90,26 @@ "LICENSE", ".gitignore", }) +CRITICAL_EXACT = frozenset({ + ".github/codex_auto_merge_policy.json", +}) REPO_ROOT = Path(__file__).resolve().parents[1] -DEFAULT_POLICY_PATH = REPO_ROOT / ".github" / "codex_auto_merge_policy.json" +AUTONOMY_POLICY_PATH_ENV = "CODEX_AUDIT_SERVICE_AUTONOMY_POLICY_PATH" -def load_autonomy_policy(path: Path = DEFAULT_POLICY_PATH) -> dict[str, Any]: - """Load the shared repo autonomy policy. +def load_autonomy_policy(path: Path | None = None) -> dict[str, Any]: + """Load the shared autonomy policy from a trusted service-owned path. - The policy is intentionally best-effort: malformed or missing files fall - back to the built-in conservative classifier instead of disabling review. + 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: @@ -180,6 +190,8 @@ def _policy_matches(path: str, rule: dict[str, Any]) -> bool: 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): diff --git a/tests/test_autonomy.py b/tests/test_autonomy.py index c0db378b..83314ab7 100644 --- a/tests/test_autonomy.py +++ b/tests/test_autonomy.py @@ -3,6 +3,7 @@ from __future__ import annotations import os +from pathlib import Path import tempfile import unittest @@ -15,6 +16,7 @@ RISK_LOW, RISK_MEDIUM, RISK_CRITICAL, + AUTONOMY_POLICY_PATH_ENV, DEFAULT_DECISION_MATRIX, load_autonomy_policy, recommended_action, @@ -62,15 +64,48 @@ def test_shared_policy_classifies_blocked_and_low_risk_paths(self) -> None: 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")