diff --git a/.github/workflows/codex_audit.yml b/.github/workflows/codex_audit.yml index 67c8b7dd..d6b31c17 100644 --- a/.github/workflows/codex_audit.yml +++ b/.github/workflows/codex_audit.yml @@ -26,8 +26,9 @@ on: description: "Audit provider" required: false type: choice - default: "auto" + default: "task_default" options: + - task_default - auto - api - anthropic @@ -68,7 +69,7 @@ jobs: ISSUE_NUMBER: ${{ github.event.client_payload.issue_number || inputs.issue_number }} SOURCE_REF: ${{ github.event.client_payload.source_ref || inputs.source_ref || 'main' }} CODEX_AUDIT_MODE: ${{ github.event.client_payload.mode || inputs.mode || 'review_and_fix' }} - CODEX_AUDIT_PROVIDER: ${{ github.event.client_payload.provider || inputs.provider || 'auto' }} + CODEX_AUDIT_PROVIDER: ${{ github.event.client_payload.provider || inputs.provider || 'task_default' }} CODEX_AUDIT_CODEX_BACKEND: service CODEX_AUDIT_TASK: ${{ github.event.client_payload.task || inputs.task || 'monthly_snapshot_audit' }} CODEX_AUDIT_AUTO_MERGE: ${{ github.event.client_payload.auto_merge || inputs.auto_merge || 'false' }} diff --git a/.github/workflows/codex_pr_review.yml b/.github/workflows/codex_pr_review.yml index b15c8f23..0e6191e9 100644 --- a/.github/workflows/codex_pr_review.yml +++ b/.github/workflows/codex_pr_review.yml @@ -18,6 +18,16 @@ on: required: false type: boolean default: false + api_fallback_enabled: + description: "Optional true/false override for direct API fallback after a recoverable Codex service failure. Empty defers to repository variables." + required: false + type: string + default: "" + direct_api_primary_enabled: + description: "Optional true/false override for API-only PR review when CODEX_AUDIT_SERVICE_URL is not configured. Empty defers to repository variables." + required: false + type: string + default: "" secrets: CODEX_AUDIT_REUSABLE_WORKFLOW_TOKEN: description: "Token that can read QuantStrategyLab/AIAuditBridge when this workflow is called from another private repo." @@ -81,6 +91,8 @@ jobs: CODEX_AUDIT_SERVICE_AUDIENCE: ${{ vars.CODEX_AUDIT_SERVICE_AUDIENCE || 'quant-codex-audit' }} CODEX_PR_REVIEW_REPO_ROOT: ${{ github.workspace }}/source CODEX_PR_REVIEW_ALLOW_UNCONFIGURED_BACKEND: ${{ inputs.allow_unconfigured_backend || 'false' }} + CODEX_PR_REVIEW_API_FALLBACK_ENABLED: ${{ github.event_name == 'workflow_call' && inputs.api_fallback_enabled != '' && inputs.api_fallback_enabled || vars.CODEX_PR_REVIEW_API_FALLBACK_ENABLED || 'true' }} + CODEX_PR_REVIEW_DIRECT_API_PRIMARY_ENABLED: ${{ github.event_name == 'workflow_call' && inputs.direct_api_primary_enabled != '' && inputs.direct_api_primary_enabled || vars.CODEX_PR_REVIEW_DIRECT_API_PRIMARY_ENABLED || 'true' }} working-directory: source run: | set -euo pipefail diff --git a/README.md b/README.md index 815addfc..64ad591d 100644 --- a/README.md +++ b/README.md @@ -73,10 +73,23 @@ Configure these values in `QuantStrategyLab/AIAuditBridge`: `openai,anthropic`. - Repository variable `OPENAI_MODEL` for OpenAI API fallback. - Repository variable `ANTHROPIC_MODEL` for Anthropic API fallback. +- Monthly audits default to `provider=auto` for `monthly_snapshot_audit` and + `provider=codex` for `long_horizon_signal_shadow`; override with + `CODEX_AUDIT_PROVIDER` when you need a specific provider. Workflow dispatch + uses `task_default` to defer provider selection to the task policy. - Monthly audits with `CODEX_AUDIT_PROVIDER=auto` fall back to the configured API reviewers when the Codex service hits quota/capacity failures. -- PR review workflows fall back to direct API review on recoverable Codex - service failures. +- PR review workflows can fall back to direct API review on recoverable Codex + service failures through `CODEX_PR_REVIEW_API_FALLBACK_ENABLED=true` or the + reusable workflow input `api_fallback_enabled`. The reusable workflow input + accepts string values `true`/`false`; when omitted it defers to repository + variables and then defaults to `true` for compatibility. Codex-only callers + should pass `api_fallback_enabled: "false"`. API-only PR review when no + service URL is configured is controlled separately by + `CODEX_PR_REVIEW_DIRECT_API_PRIMARY_ENABLED` or reusable workflow input + `direct_api_primary_enabled`; this uses the same `true`/`false`, variable, + and compatibility default rules and should be set to `"false"` for Codex-only + callers. - Repository variable `CODEX_AUDIT_SERVICE_MODEL` for the VPS Codex service primary path; `VPS Codex Service Ops` deploy writes it into the systemd unit. - Optional repository variable `CODEX_AUDIT_SERVICE_REASONING_EFFORT` for a diff --git a/README.zh-CN.md b/README.zh-CN.md index c9c17526..0590ff35 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -71,8 +71,21 @@ AIAuditBridge 只使用 service backend。workflow 运行在 `ubuntu-latest`， `openai,anthropic`。 - repository variable `OPENAI_MODEL`，API fallback 使用的 OpenAI 模型。 - repository variable `ANTHROPIC_MODEL`，API fallback 使用的 Anthropic 模型。 +- monthly audit 的默认 provider 随 task 而变：`monthly_snapshot_audit` + 默认 `auto`，`long_horizon_signal_shadow` 默认 `codex`；如需固定 provider， + 请显式设置 `CODEX_AUDIT_PROVIDER`。workflow dispatch 使用 `task_default` + 把 provider 选择交给 task policy。 - repository variable `CODEX_AUDIT_SERVICE_MODEL`，VPS Codex service 主路径模型； `VPS Codex Service Ops` deploy 会写入 systemd unit。 +- PR review 可以通过 repository variable + `CODEX_PR_REVIEW_API_FALLBACK_ENABLED=true` 或 reusable workflow input + `api_fallback_enabled` 在可恢复的 Codex service 失败后启用 direct API + fallback。reusable workflow input 使用字符串 `true`/`false`；省略时先使用 + repository variable，再为兼容旧调用方默认 `true`。只走 Codex 的调用方应传入 + `api_fallback_enabled: "false"`。当未配置 service URL 时，是否允许 API-only + PR review 由 `CODEX_PR_REVIEW_DIRECT_API_PRIMARY_ENABLED` 或 reusable + workflow input `direct_api_primary_enabled` 单独控制；该项使用同样的变量和 + 兼容默认规则，Codex-only 调用方应设为 `"false"`。 - workflow 已配置 `id-token: write`，用于向 service 提供 GitHub Actions OIDC token。 service host 启动示例： diff --git a/scripts/run_codex_pr_review.py b/scripts/run_codex_pr_review.py index 9d4e5950..632a12ad 100644 --- a/scripts/run_codex_pr_review.py +++ b/scripts/run_codex_pr_review.py @@ -492,6 +492,14 @@ def _allow_unconfigured_backend() -> bool: return parse_bool(env_value("CODEX_PR_REVIEW_ALLOW_UNCONFIGURED_BACKEND")) +def _api_fallback_enabled() -> bool: + return parse_bool(env_value("CODEX_PR_REVIEW_API_FALLBACK_ENABLED", "true")) + + +def _direct_api_primary_enabled() -> bool: + return parse_bool(env_value("CODEX_PR_REVIEW_DIRECT_API_PRIMARY_ENABLED", "true")) + + def run_codex_review_with_fallback( prompt: str, timeout_minutes: int, @@ -499,8 +507,6 @@ def run_codex_review_with_fallback( changed_file_count: int = 0, changed_line_count: int = 0, ) -> str: - # env_value() returns "" when CODEX_AUDIT_SERVICE_URL is unset, so this - # guard keeps the direct-API path intact without special error handling. service_url = env_value("CODEX_AUDIT_SERVICE_URL") service_failure: Exception | None = None if service_url: @@ -517,10 +523,15 @@ def run_codex_review_with_fallback( if not _service_review_should_fallback(exc): raise service_failure = exc - print(f"::warning::Codex service review failed; falling back to direct API: {exc}") + print(f"::warning::Codex service review failed: {exc}") except (json.JSONDecodeError, OSError, urllib.error.URLError) as exc: service_failure = exc - print(f"::error::Codex service review failed; falling back to direct API: {exc}") + print(f"::error::Codex service review failed: {exc}") + + if service_failure is not None and not _api_fallback_enabled(): + raise ReviewError(f"Codex service review failed and direct API fallback is disabled: {service_failure}") + if not service_url and not _direct_api_primary_enabled(): + raise ReviewError(NO_REVIEW_BACKEND_CONFIGURED) print("Running Codex review via direct API") try: diff --git a/scripts/run_monthly_codex_audit.py b/scripts/run_monthly_codex_audit.py index b82b9387..d46250a7 100644 --- a/scripts/run_monthly_codex_audit.py +++ b/scripts/run_monthly_codex_audit.py @@ -38,6 +38,11 @@ DEFAULT_TASK = "monthly_snapshot_audit" DEFAULT_MODE = "review_and_fix" DEFAULT_PROVIDER = "auto" +TASK_DEFAULT_PROVIDER = "task_default" +TASK_DEFAULT_PROVIDERS = { + "monthly_snapshot_audit": "auto", + "long_horizon_signal_shadow": "codex", +} API_PATCH_SYSTEM_PROMPT = ( "You are AIAuditBridge's API fallback patch provider. " "Return exactly one JSON object that matches the service patch contract. " @@ -384,13 +389,19 @@ def validate_task(task: str, source_repo: str) -> str: return normalized -def validate_provider(provider: str) -> str: - normalized = (provider or DEFAULT_PROVIDER).strip().lower() +def validate_provider(provider: str, task: str = DEFAULT_TASK) -> str: + normalized = (provider or TASK_DEFAULT_PROVIDER).strip().lower() + if normalized == TASK_DEFAULT_PROVIDER: + return default_provider_for_task(task) if normalized not in SUPPORTED_PROVIDERS: raise BridgeError(f"Unsupported CODEX_AUDIT_PROVIDER: {provider!r}") return normalized +def default_provider_for_task(task: str) -> str: + return TASK_DEFAULT_PROVIDERS.get(task, DEFAULT_PROVIDER) + + def api_fallback_allowed_source_repos() -> frozenset[str]: configured = env_value("CODEX_AUDIT_API_FALLBACK_ALLOWED_SOURCE_REPOSITORIES") if not configured: @@ -2772,7 +2783,7 @@ def main() -> int: mode = env_value("CODEX_AUDIT_MODE", DEFAULT_MODE) if mode not in {"review_only", "review_and_fix"}: raise BridgeError(f"Unsupported CODEX_AUDIT_MODE: {mode}") - provider = validate_provider(env_value("CODEX_AUDIT_PROVIDER", DEFAULT_PROVIDER)) + provider = validate_provider(env_value("CODEX_AUDIT_PROVIDER", TASK_DEFAULT_PROVIDER), task=task) codex_backend = validate_codex_backend(env_value("CODEX_AUDIT_CODEX_BACKEND", DEFAULT_CODEX_BACKEND)) issue_number_raw = env_value("ISSUE_NUMBER") if not issue_number_raw.isdigit(): diff --git a/service/automation_run_ledger.py b/service/automation_run_ledger.py new file mode 100644 index 00000000..ebaecd62 --- /dev/null +++ b/service/automation_run_ledger.py @@ -0,0 +1,300 @@ +"""Lightweight automation run ledger and runtime health policy runner.""" + +from __future__ import annotations + +import threading +import time +from collections import Counter +from copy import deepcopy +from typing import Any + +from service.task_state import TERMINAL_STATES + +CONTROL_CONTINUE = "continue" +CONTROL_REVIEW_ONLY = "review_only" +CONTROL_PAUSE_AUTO_FIX = "pause_auto_fix" +CONTROL_ESCALATE = "escalate" + +CONTROL_ACTIONS = frozenset( + { + CONTROL_CONTINUE, + CONTROL_REVIEW_ONLY, + CONTROL_PAUSE_AUTO_FIX, + CONTROL_ESCALATE, + } +) + +DEFAULT_MAX_RUNS = 500 +DEFAULT_MAX_EVENTS_PER_RUN = 50 +MAX_RUN_METADATA_FIELDS = 20 +MAX_RUN_METADATA_VALUE_LENGTH = 500 +MAX_EVENT_METADATA_FIELDS = 10 +MAX_EVENT_METADATA_VALUE_LENGTH = 200 +INTERNAL_ENTRY_KEYS = frozenset({"_ledger_sequence"}) + +QUOTA_STATUS_SEVERITY = { + "ok": 0, + "healthy": 0, + "unknown": 1, + "unavailable": 1, + "low": 2, + "constrained": 2, + "exhausted": 3, + "blocked": 3, +} + + +def _normalize_status(value: Any, default: str = "") -> str: + if isinstance(value, dict): + value = value.get("status", default) + return str(value or default).strip().lower() + + +def _normalize_quota_status(value: Any, default: str = "") -> str: + statuses = [_normalize_status(value)] + if isinstance(value, dict) and isinstance(value.get("quota"), dict): + statuses.append(_normalize_status(value["quota"])) + normalized = [status for status in statuses if status] + if not normalized: + return default + return max(normalized, key=lambda status: QUOTA_STATUS_SEVERITY.get(status, 1)) + + +def _is_omitted(value: Any) -> bool: + return value is None or (isinstance(value, str) and not value.strip()) + + +def _metadata_snapshot( + metadata: dict[str, Any], + *, + max_fields: int, + max_value_length: int, +) -> dict[str, Any]: + snapshot: dict[str, Any] = {} + omitted = 0 + for key, value in metadata.items(): + if len(snapshot) >= max_fields: + omitted += 1 + continue + if not isinstance(value, str | int | float | bool | type(None)): + omitted += 1 + continue + if isinstance(value, str) and len(value) > max_value_length: + value = value[:max_value_length] + "…" + snapshot[str(key)] = value + if omitted: + snapshot["_omitted_fields"] = omitted + return snapshot + + +def _run_metadata_snapshot(metadata: dict[str, Any]) -> dict[str, Any]: + return _metadata_snapshot( + metadata, + max_fields=MAX_RUN_METADATA_FIELDS, + max_value_length=MAX_RUN_METADATA_VALUE_LENGTH, + ) + + +def _event_metadata_snapshot(metadata: dict[str, Any]) -> dict[str, Any]: + return _metadata_snapshot( + metadata, + max_fields=MAX_EVENT_METADATA_FIELDS, + max_value_length=MAX_EVENT_METADATA_VALUE_LENGTH, + ) + + +def suggest_control_action( + service_health: Any = "", + quota_status: Any = "", + org_health_status: Any = "", +) -> dict[str, Any]: + """Convert health/quota/org-health signals into a control action.""" + health = _normalize_status(service_health, "unknown") + quota = _normalize_quota_status(quota_status, "unknown") + org_health = _normalize_status(org_health_status, "unknown") + + reasons: list[str] = [] + action = CONTROL_REVIEW_ONLY + + if health == "unhealthy": + action = CONTROL_ESCALATE + reasons.append("service health is unhealthy") + elif quota in {"exhausted", "blocked"}: + action = CONTROL_ESCALATE + reasons.append(f"quota status is {quota}") + elif org_health == "unhealthy": + action = CONTROL_ESCALATE + reasons.append("org health is unhealthy") + elif health == "degraded" or quota in {"low", "constrained"} or org_health == "degraded": + action = CONTROL_PAUSE_AUTO_FIX + if health == "degraded": + reasons.append("service health is degraded") + if quota in {"low", "constrained"}: + reasons.append(f"quota status is {quota}") + if org_health == "degraded": + reasons.append("org health is degraded") + elif health in {"healthy", "ok"} and quota in {"ok", "healthy"} and org_health in {"ok", "healthy"}: + action = CONTROL_CONTINUE + reasons.append("all runtime signals are healthy") + else: + reasons.append("runtime signals are incomplete") + + return { + "action": action, + "service_health": health, + "quota_status": quota, + "org_health_status": org_health, + "reasons": reasons, + "requires_human_review": action in {CONTROL_REVIEW_ONLY, CONTROL_PAUSE_AUTO_FIX, CONTROL_ESCALATE}, + "auto_fix_allowed": action == CONTROL_CONTINUE, + } + + +class AutomationRunLedger: + """In-memory ledger of automation runs and their latest task state.""" + + def __init__( + self, + *, + max_runs: int = DEFAULT_MAX_RUNS, + max_events_per_run: int = DEFAULT_MAX_EVENTS_PER_RUN, + ) -> None: + self._lock = threading.Lock() + self._runs: dict[str, dict[str, Any]] = {} + self._max_runs = max(1, int(max_runs)) + self._max_events_per_run = max(1, int(max_events_per_run)) + self._sequence = 0 + + def _evict_old_runs_locked(self) -> None: + overflow = len(self._runs) - self._max_runs + if overflow <= 0: + return + ordered = sorted( + self._runs.values(), + key=lambda item: ( + float(item.get("updated_at", 0.0)), + int(item.get("_ledger_sequence", 0)), + ), + ) + for entry in ordered[:overflow]: + self._runs.pop(str(entry["run_id"])) + + @staticmethod + def _public_entry(entry: dict[str, Any], *, include_events: bool = True) -> dict[str, Any]: + return { + key: deepcopy(value) + for key, value in entry.items() + if key not in INTERNAL_ENTRY_KEYS and (include_events or key != "events") + } + + def record( + self, + run_id: str, + task_state: str, + *, + task_name: str = "", + suggested_action: str = "", + service_health: Any = "", + quota_status: Any = "", + org_health_status: Any = "", + metadata: dict[str, Any] | None = None, + ) -> dict[str, Any]: + """Record or update one automation run.""" + if not run_id.strip(): + raise ValueError("run_id must be a non-empty string") + now = time.time() + entry = { + "run_id": run_id, + "task_name": task_name, + "task_state": str(task_state or "").strip().lower(), + "suggested_action": str(suggested_action or "").strip().lower(), + "service_health": _normalize_status(service_health), + "quota_status": _normalize_quota_status(quota_status), + "org_health_status": _normalize_status(org_health_status), + "metadata": _run_metadata_snapshot(metadata or {}), + "updated_at": now, + "events": [], + } + with self._lock: + current = self._runs.get(run_id) + if current: + entry["_ledger_sequence"] = current.get("_ledger_sequence", 0) + old_events = list(current.get("events", [])) + entry["events"] = ( + old_events[-(self._max_events_per_run - 1) :] if self._max_events_per_run > 1 else [] + ) + if not entry["task_name"]: + entry["task_name"] = str(current.get("task_name", "")) + if not entry["metadata"]: + entry["metadata"] = deepcopy(current.get("metadata", {})) + if not entry["suggested_action"]: + entry["suggested_action"] = str(current.get("suggested_action", "")) + if _is_omitted(service_health): + entry["service_health"] = str(current.get("service_health", "")) + if _is_omitted(quota_status): + entry["quota_status"] = str(current.get("quota_status", "")) + if _is_omitted(org_health_status): + entry["org_health_status"] = str(current.get("org_health_status", "")) + else: + self._sequence += 1 + entry["_ledger_sequence"] = self._sequence + entry["events"].append( + { + "task_state": entry["task_state"], + "suggested_action": entry["suggested_action"], + "service_health": entry["service_health"], + "quota_status": entry["quota_status"], + "org_health_status": entry["org_health_status"], + "metadata": _event_metadata_snapshot(entry["metadata"]), + "recorded_at": now, + } + ) + self._runs[run_id] = entry + self._evict_old_runs_locked() + return self._public_entry(self._runs[run_id]) + + def get(self, run_id: str) -> dict[str, Any] | None: + with self._lock: + entry = self._runs.get(run_id) + return self._public_entry(entry) if entry else None + + def snapshot(self, *, limit: int | None = 100, include_events: bool = False) -> dict[str, Any]: + with self._lock: + retained_runs = [ + self._public_entry(entry, include_events=include_events) + for entry in self._runs.values() + ] + max_runs = self._max_runs + max_events_per_run = self._max_events_per_run + + task_states = Counter(str(run.get("task_state", "")).strip().lower() for run in retained_runs if run.get("task_state")) + suggested_actions = Counter( + str(run.get("suggested_action", "")).strip().lower() + for run in retained_runs + if run.get("suggested_action") + ) + terminal_runs = sum(1 for run in retained_runs if str(run.get("task_state", "")).strip().lower() in TERMINAL_STATES) + ordered_runs = sorted( + retained_runs, + key=lambda item: (float(item.get("updated_at", 0.0)), str(item.get("run_id", ""))), + reverse=True, + ) + if limit is not None: + ordered_runs = ordered_runs[: max(0, int(limit))] + runs = ordered_runs + return { + "runs": runs, + "summary": { + "total_runs": len(retained_runs), + "returned_runs": len(runs), + "active_runs": len(retained_runs) - terminal_runs, + "terminal_runs": terminal_runs, + "task_states": dict(task_states), + "suggested_actions": dict(suggested_actions), + "retention": { + "max_runs": max_runs, + "max_events_per_run": max_events_per_run, + "events_included": include_events, + }, + }, + } diff --git a/tests/test_automation_run_ledger.py b/tests/test_automation_run_ledger.py new file mode 100644 index 00000000..939d1b2c --- /dev/null +++ b/tests/test_automation_run_ledger.py @@ -0,0 +1,192 @@ +"""Tests for service/automation_run_ledger.py.""" + +from __future__ import annotations + +import unittest +from unittest.mock import patch + +from service.automation_run_ledger import ( + AutomationRunLedger, + CONTROL_CONTINUE, + CONTROL_ESCALATE, + CONTROL_PAUSE_AUTO_FIX, + CONTROL_REVIEW_ONLY, + suggest_control_action, +) + + +class TestSuggestControlAction(unittest.TestCase): + def test_healthy_signals_continue(self) -> None: + result = suggest_control_action("healthy", {"status": "ok"}, {"status": "ok"}) + self.assertEqual(result["action"], CONTROL_CONTINUE) + self.assertTrue(result["auto_fix_allowed"]) + + def test_ok_service_health_is_healthy(self) -> None: + result = suggest_control_action({"status": "ok"}, {"status": "ok"}, {"status": "ok"}) + self.assertEqual(result["action"], CONTROL_CONTINUE) + self.assertTrue(result["auto_fix_allowed"]) + + def test_healthy_quota_status_is_healthy(self) -> None: + result = suggest_control_action({"status": "ok"}, {"status": "healthy"}, {"status": "ok"}) + self.assertEqual(result["action"], CONTROL_CONTINUE) + self.assertTrue(result["auto_fix_allowed"]) + + def test_unknown_org_health_falls_back_to_review_only(self) -> None: + result = suggest_control_action("healthy", "ok", {"status": "unavailable"}) + self.assertEqual(result["action"], CONTROL_REVIEW_ONLY) + self.assertFalse(result["auto_fix_allowed"]) + self.assertTrue(result["requires_human_review"]) + + def test_missing_signals_fall_back_to_review_only(self) -> None: + result = suggest_control_action() + self.assertEqual(result["action"], CONTROL_REVIEW_ONLY) + self.assertFalse(result["auto_fix_allowed"]) + self.assertIn("runtime signals are incomplete", result["reasons"]) + + def test_degraded_signals_pause_auto_fix(self) -> None: + result = suggest_control_action("degraded", "ok", "ok") + self.assertEqual(result["action"], CONTROL_PAUSE_AUTO_FIX) + self.assertIn("service health is degraded", result["reasons"]) + + def test_low_quota_pauses_auto_fix(self) -> None: + result = suggest_control_action("healthy", {"status": "low"}, "ok") + self.assertEqual(result["action"], CONTROL_PAUSE_AUTO_FIX) + self.assertIn("quota status is low", result["reasons"]) + + def test_nested_quota_snapshot_controls_action(self) -> None: + result = suggest_control_action( + "healthy", + {"status": "ok", "quota": {"status": "exhausted"}}, + "ok", + ) + self.assertEqual(result["action"], CONTROL_ESCALATE) + self.assertFalse(result["auto_fix_allowed"]) + + def test_quota_snapshot_keeps_most_severe_status(self) -> None: + result = suggest_control_action( + "healthy", + {"status": "blocked", "quota": {"status": "ok"}}, + "ok", + ) + self.assertEqual(result["action"], CONTROL_ESCALATE) + self.assertEqual(result["quota_status"], "blocked") + + def test_unhealthy_signals_escalate(self) -> None: + result = suggest_control_action("healthy", "ok", {"status": "unhealthy"}) + self.assertEqual(result["action"], CONTROL_ESCALATE) + self.assertIn("org health is unhealthy", result["reasons"]) + + +class TestAutomationRunLedger(unittest.TestCase): + def setUp(self) -> None: + self.ledger = AutomationRunLedger() + + def test_record_updates_latest_state_and_keeps_history(self) -> None: + first = self.ledger.record( + "run-1", + "queued", + task_name="monthly-audit", + suggested_action=CONTROL_REVIEW_ONLY, + service_health="healthy", + quota_status="ok", + org_health_status="ok", + metadata={"repo": "QuantStrategyLab/AIAuditBridge"}, + ) + second = self.ledger.record( + "run-1", + "running", + suggested_action=CONTROL_CONTINUE, + service_health="healthy", + quota_status="ok", + org_health_status="ok", + ) + + self.assertEqual(first["task_state"], "queued") + self.assertEqual(second["task_state"], "running") + self.assertNotIn("_ledger_sequence", second) + self.assertEqual(len(second["events"]), 2) + self.assertEqual(second["events"][0]["task_state"], "queued") + self.assertEqual(second["events"][1]["suggested_action"], CONTROL_CONTINUE) + + def test_snapshot_summarizes_terminal_and_active_runs(self) -> None: + self.ledger.record("run-1", "running", suggested_action=CONTROL_CONTINUE) + self.ledger.record("run-2", "merged", suggested_action=CONTROL_CONTINUE) + + snapshot = self.ledger.snapshot() + self.assertEqual(snapshot["summary"]["total_runs"], 2) + self.assertEqual(snapshot["summary"]["returned_runs"], 2) + self.assertEqual(snapshot["summary"]["active_runs"], 1) + self.assertEqual(snapshot["summary"]["terminal_runs"], 1) + self.assertEqual(snapshot["summary"]["suggested_actions"][CONTROL_CONTINUE], 2) + self.assertNotIn("events", snapshot["runs"][0]) + + def test_snapshot_can_include_bounded_history(self) -> None: + ledger = AutomationRunLedger(max_events_per_run=2) + ledger.record("run-1", "queued") + ledger.record("run-1", "running") + ledger.record("run-1", "merged") + + snapshot = ledger.snapshot(include_events=True) + self.assertEqual(len(snapshot["runs"][0]["events"]), 2) + self.assertEqual(snapshot["runs"][0]["events"][0]["task_state"], "running") + self.assertTrue(snapshot["summary"]["retention"]["events_included"]) + + def test_ledger_evicts_old_runs_by_count(self) -> None: + ledger = AutomationRunLedger(max_runs=2) + ledger.record("run-1", "queued") + ledger.record("run-2", "queued") + ledger.record("run-3", "queued") + + snapshot = ledger.snapshot(limit=None) + self.assertEqual(snapshot["summary"]["total_runs"], 2) + self.assertEqual({run["run_id"] for run in snapshot["runs"]}, {"run-2", "run-3"}) + + def test_ledger_eviction_keeps_new_run_when_timestamps_match(self) -> None: + ledger = AutomationRunLedger(max_runs=1) + with patch("service.automation_run_ledger.time.time", return_value=123.0): + ledger.record("run-1", "queued") + ledger.record("run-2", "queued") + + snapshot = ledger.snapshot(limit=None) + self.assertEqual([run["run_id"] for run in snapshot["runs"]], ["run-2"]) + + def test_update_preserves_control_fields_when_omitted(self) -> None: + self.ledger.record( + "run-1", + "queued", + suggested_action=CONTROL_PAUSE_AUTO_FIX, + service_health="degraded", + quota_status="low", + org_health_status="ok", + ) + + updated = self.ledger.record("run-1", "running") + + self.assertEqual(updated["suggested_action"], CONTROL_PAUSE_AUTO_FIX) + self.assertEqual(updated["service_health"], "degraded") + self.assertEqual(updated["quota_status"], "low") + self.assertEqual(updated["events"][-1]["suggested_action"], CONTROL_PAUSE_AUTO_FIX) + + def test_record_sanitizes_metadata(self) -> None: + metadata = { + "repo": "QuantStrategyLab/AIAuditBridge", + "repos": ["QuantStrategyLab/AIAuditBridge"], + "note": "x" * 600, + } + recorded = self.ledger.record("run-1", "queued", metadata=metadata) + metadata["repos"].append("mutated") + + stored = self.ledger.get("run-1") + + self.assertEqual(recorded["metadata"]["repo"], "QuantStrategyLab/AIAuditBridge") + self.assertNotIn("repos", recorded["metadata"]) + self.assertEqual(recorded["metadata"]["_omitted_fields"], 1) + self.assertTrue(recorded["metadata"]["note"].endswith("…")) + self.assertEqual(stored["metadata"], recorded["metadata"]) + self.assertEqual(stored["events"][0]["metadata"]["repo"], "QuantStrategyLab/AIAuditBridge") + self.assertNotIn("repos", stored["events"][0]["metadata"]) + self.assertEqual(stored["events"][0]["metadata"]["_omitted_fields"], 1) + + def test_record_rejects_blank_run_id(self) -> None: + with self.assertRaises(ValueError): + self.ledger.record(" ", "queued") diff --git a/tests/test_run_codex_pr_review.py b/tests/test_run_codex_pr_review.py index a7b3d0c4..4275d2bd 100644 --- a/tests/test_run_codex_pr_review.py +++ b/tests/test_run_codex_pr_review.py @@ -47,7 +47,14 @@ def test_load_policy_uses_trusted_base_ref(self) -> None: def test_service_failure_falls_back_to_direct_api(self) -> None: with ( - patch.dict(os.environ, {"CODEX_AUDIT_SERVICE_URL": "https://service.example"}, clear=True), + patch.dict( + os.environ, + { + "CODEX_AUDIT_SERVICE_URL": "https://service.example", + "CODEX_PR_REVIEW_API_FALLBACK_ENABLED": "true", + }, + clear=True, + ), patch( "scripts.run_codex_pr_review.run_codex_service_review", side_effect=ReviewError("HTTP 429 Too Many Requests"), @@ -65,6 +72,93 @@ def test_service_failure_falls_back_to_direct_api(self) -> None: self.assertEqual(output, "api review") direct_api.assert_called_once_with("Review this PR.", complexity="high") + def test_service_failure_does_not_fallback_to_direct_api_when_disabled(self) -> None: + with ( + patch.dict( + os.environ, + { + "CODEX_AUDIT_SERVICE_URL": "https://service.example", + "CODEX_PR_REVIEW_API_FALLBACK_ENABLED": "false", + }, + clear=True, + ), + patch( + "scripts.run_codex_pr_review.run_codex_service_review", + side_effect=ReviewError("HTTP 429 Too Many Requests"), + ), + patch("scripts.run_codex_pr_review.run_direct_api_review") as direct_api, + ): + with self.assertRaises(ReviewError) as raised: + run_codex_review_with_fallback( + "Review this PR.", + timeout_minutes=20, + complexity="high", + changed_file_count=3, + changed_line_count=120, + ) + + self.assertIn("direct API fallback is disabled", str(raised.exception)) + direct_api.assert_not_called() + + def test_direct_api_runs_when_service_url_is_unset(self) -> None: + with ( + patch.dict( + os.environ, + { + "OPENAI_API_KEY": "test-key", + "CODEX_PR_REVIEW_API_FALLBACK_ENABLED": "true", + }, + clear=True, + ), + patch("scripts.run_codex_pr_review.run_direct_api_review", return_value="api review") as direct_api, + ): + output = run_codex_review_with_fallback( + "Review this PR.", + timeout_minutes=20, + complexity="high", + changed_file_count=3, + changed_line_count=120, + ) + + self.assertEqual(output, "api review") + direct_api.assert_called_once_with("Review this PR.", complexity="high") + + def test_direct_api_runs_when_service_url_unset_even_if_service_fallback_disabled(self) -> None: + with ( + patch.dict( + os.environ, + { + "OPENAI_API_KEY": "test-key", + "CODEX_PR_REVIEW_API_FALLBACK_ENABLED": "false", + "CODEX_PR_REVIEW_DIRECT_API_PRIMARY_ENABLED": "true", + }, + clear=True, + ), + patch("scripts.run_codex_pr_review.run_direct_api_review", return_value="api review") as direct_api, + ): + output = run_codex_review_with_fallback("Review this PR.", timeout_minutes=20) + + self.assertEqual(output, "api review") + direct_api.assert_called_once() + + def test_direct_api_is_blocked_when_service_url_unset_and_primary_disabled(self) -> None: + with ( + patch.dict( + os.environ, + { + "OPENAI_API_KEY": "test-key", + "CODEX_PR_REVIEW_DIRECT_API_PRIMARY_ENABLED": "false", + }, + clear=True, + ), + patch("scripts.run_codex_pr_review.run_direct_api_review") as direct_api, + ): + with self.assertRaises(ReviewError) as raised: + run_codex_review_with_fallback("Review this PR.", timeout_minutes=20) + + self.assertEqual(str(raised.exception), run_codex_pr_review.NO_REVIEW_BACKEND_CONFIGURED) + direct_api.assert_not_called() + def test_service_fallback_without_api_keys_preserves_service_failure(self) -> None: with ( @@ -240,7 +334,22 @@ def test_reusable_workflow_runs_bridge_script_against_source_checkout(self) -> N self.assertIn("CODEX_AUDIT_REUSABLE_WORKFLOW_TOKEN", workflow) self.assertIn("caller_concurrency_key", workflow) self.assertIn("allow_unconfigured_backend", workflow) + self.assertIn("api_fallback_enabled", workflow) + self.assertIn("direct_api_primary_enabled", workflow) + self.assertIn("Empty defers to repository variables", workflow) + self.assertIn("default: false", workflow) + self.assertIn('default: ""', workflow) self.assertIn("CODEX_PR_REVIEW_ALLOW_UNCONFIGURED_BACKEND", workflow) + self.assertIn("CODEX_PR_REVIEW_API_FALLBACK_ENABLED", workflow) + self.assertIn("CODEX_PR_REVIEW_DIRECT_API_PRIMARY_ENABLED", workflow) + self.assertIn( + "github.event_name == 'workflow_call' && inputs.api_fallback_enabled != '' && inputs.api_fallback_enabled || vars.CODEX_PR_REVIEW_API_FALLBACK_ENABLED || 'true'", + workflow, + ) + self.assertIn( + "github.event_name == 'workflow_call' && inputs.direct_api_primary_enabled != '' && inputs.direct_api_primary_enabled || vars.CODEX_PR_REVIEW_DIRECT_API_PRIMARY_ENABLED || 'true'", + workflow, + ) self.assertIn("inputs.caller_concurrency_key || github.event.pull_request.number || github.run_id", workflow) self.assertNotIn("Validate bridge checkout token", workflow) self.assertIn("required: false", workflow) diff --git a/tests/test_run_monthly_codex_audit.py b/tests/test_run_monthly_codex_audit.py index bb5ae102..da89e57e 100644 --- a/tests/test_run_monthly_codex_audit.py +++ b/tests/test_run_monthly_codex_audit.py @@ -69,6 +69,7 @@ validate_provider, validate_repo, validate_task, + default_provider_for_task, write_codex_context, ) from scripts.codex_audit_service import CodexAuditServiceRequestHandler, _codex_env @@ -145,6 +146,8 @@ def test_validate_task_rejects_repo_task_mismatch(self) -> None: def test_validate_provider_accepts_supported_values(self) -> None: self.assertEqual(validate_provider(""), "auto") + self.assertEqual(validate_provider("", task="long_horizon_signal_shadow"), "codex") + self.assertEqual(validate_provider("task_default", task="long_horizon_signal_shadow"), "codex") self.assertEqual(validate_provider("codex"), "codex") self.assertEqual(validate_provider("OPENAI"), "openai") self.assertEqual(validate_provider("anthropic"), "anthropic") @@ -153,6 +156,11 @@ def test_validate_provider_accepts_supported_values(self) -> None: with self.assertRaises(Exception): validate_provider("claude") + def test_default_provider_for_task_is_task_specific(self) -> None: + self.assertEqual(default_provider_for_task("monthly_snapshot_audit"), "auto") + self.assertEqual(default_provider_for_task("long_horizon_signal_shadow"), "codex") + self.assertEqual(default_provider_for_task("unknown_task"), "auto") + def test_api_fallback_allowlist_requires_explicit_configuration(self) -> None: with patch.dict(os.environ, {}, clear=True): with self.assertRaisesRegex(BridgeError, "must explicitly list"): @@ -825,6 +833,56 @@ def test_main_codex_failure_uses_api_patch_remediation_for_auto_provider(self) - patch_remediation.assert_called_once() self.assertIs(patch_remediation.call_args.kwargs["workspace"], workspace) + def test_main_defaults_long_horizon_task_to_codex_provider(self) -> None: + issue = { + "number": 19, + "title": "Shadow", + "html_url": "https://example.test/issues/19", + "body": "Body", + "labels": [], + } + env = { + "SOURCE_REPO": "QuantStrategyLab/ResearchSignalContextPipelines", + "SOURCE_REF": "main", + "ISSUE_NUMBER": "19", + "CODEX_AUDIT_GH_TOKEN": "token", + "CODEX_AUDIT_TASK": "long_horizon_signal_shadow", + "CODEX_AUDIT_MODE": "review_and_fix", + "CODEX_AUDIT_CODEX_BACKEND": "service", + "CODEX_AUDIT_API_FALLBACK_ALLOWED_SOURCE_REPOSITORIES": ( + "QuantStrategyLab/ResearchSignalContextPipelines" + ), + "CODEX_AUDIT_API_FALLBACK_ALLOW_FIX": "true", + } + with ( + patch.dict(os.environ, env, clear=True), + patch("scripts.run_monthly_codex_audit.github_request", return_value=issue), + patch("scripts.run_monthly_codex_audit.fetch_issue_comments", return_value=[]), + patch("scripts.run_monthly_codex_audit.prepare_remediation_workspace") as prepare, + patch( + "scripts.run_monthly_codex_audit.run_codex_backend", + return_value=(1, "Codex audit service job failed [quota_or_capacity_failure]: budget", ""), + ), + patch("scripts.run_monthly_codex_audit.run_auto_provider_fallback") as patch_fallback, + patch("scripts.run_monthly_codex_audit.post_issue_comment") as post_comment, + ): + workspace = RemediationWorkspace( + repo_dir=Path("/tmp/source"), + branch_name="codex/long-horizon-signal-issue-19-test", + baseline_auto_merge_policy=dict(DEFAULT_GUARDED_AUTO_MERGE_POLICY), + feedback_retry_pr=None, + stale_auto_merge_label=GUARDED_AUTO_MERGE_LABEL, + stale_auto_merge_label_skip_reason="", + stale_auto_merge_label_removed=False, + prompt="prompt", + ) + prepare.return_value = workspace + exit_code = run_audit_main() + + self.assertEqual(exit_code, 1) + patch_fallback.assert_not_called() + post_comment.assert_called_once() + def test_main_codex_quota_failure_uses_api_fallback_for_auto_provider(self) -> None: issue = { "number": 19, @@ -2247,6 +2305,9 @@ def test_workflow_uses_service_backend_only(self) -> None: workflow = Path(".github/workflows/codex_audit.yml").read_text(encoding="utf-8") self.assertIn("runs-on: ubuntu-latest", workflow) + self.assertIn('default: "task_default"', workflow) + self.assertIn(" - task_default", workflow) + self.assertIn("CODEX_AUDIT_PROVIDER: ${{ github.event.client_payload.provider || inputs.provider || 'task_default' }}", workflow) self.assertIn("CODEX_AUDIT_CODEX_BACKEND: service", workflow) self.assertIn("CODEX_AUDIT_SERVICE_URL: ${{ secrets.CODEX_AUDIT_SERVICE_URL }}", workflow) self.assertNotIn("codex_backend:", workflow)