feat: centralize AI monthly budget guard

.github/workflows/codex_pr_review.yml

@@ -110,7 +110,7 @@ jobs:
           CODEX_PR_REVIEW_REUSABLE_CALL: ${{ github.repository != 'QuantStrategyLab/AIAuditBridge' && 'true' || 'false' }}
           CODEX_PR_REVIEW_API_FALLBACK_INPUT: ${{ inputs.api_fallback_enabled }}
           CODEX_PR_REVIEW_DIRECT_API_PRIMARY_INPUT: ${{ inputs.direct_api_primary_enabled }}
-          CODEX_PR_REVIEW_API_FALLBACK_DEFAULT: ${{ vars.CODEX_PR_REVIEW_API_FALLBACK_ENABLED || 'true' }}
+          CODEX_PR_REVIEW_API_FALLBACK_DEFAULT: ${{ vars.CODEX_PR_REVIEW_API_FALLBACK_ENABLED || 'false' }}
           CODEX_PR_REVIEW_DIRECT_API_PRIMARY_DEFAULT: ${{ vars.CODEX_PR_REVIEW_DIRECT_API_PRIMARY_ENABLED || 'true' }}
         working-directory: source
         run: |

docs/ai_autonomy_architecture.md

@@ -424,3 +424,24 @@ trusted review comment 只保存最近固定轮数、固定字节上限且脱敏
 2. 再补一份“自治边界表”，把自动 / 人工 / 禁止 三类动作列清楚。
 3. 然后补指标持久化和 dashboard 视图。
 4. 最后再决定是否扩大 auto-merge 覆盖面。
+
+## 7. AI Budget Guard v1（中央执行前门）
+
+所有 `research`、`optimization`、`review` 和 `auto_fix` 入口在启动前必须生成
+`ai_budget_decision.v1` 并取得 `decision=allow`。这是 AIAuditBridge 的中央执行门，
+不是各仓库自行复制的 branch rule；确定性监控、kill-switch、暂停和回滚不依赖 AI 额度。
+
+- API-key 预算按 provider、key/project scope、repo、task class 和明确的 billing period
+  （默认 `UTC` 日历月）记账。未配置月度预算时可用上限为 `0`；内部 hard limit 取用户月度预算
+  与 provider/project limit 的 `80%` 两者较小值。
+- usage/cost 快照缺失、过期或无法解析时 fail closed。每次启动先做原子 reservation，并发
+  reservation 不能共同突破 hard limit。reservation 的终态由 provider dispatch 证据决定：明确未
+  dispatch/provider 未接收时才 release；明确已 dispatch、accepted 或 billable 时按 actual 或保守
+  estimate settle；发送边界崩溃、timeout 或重启后无法证明时持久化为 `pending_uncertain`，继续占用
+  额度并 fail closed，直到 provider usage、idempotency 或显式 reconciliation 给出证据。不得静默
+  release，也不得把不确定状态直接永久 settle。
+- Codex 同时读取 primary/secondary rate-limit window，使用更紧张的剩余比例；research、
+  maintenance/review/auto-fix、incident 的门槛分别是 `30%/20%/10%`，并永久保留最后 `10%`。
+  快照不可用时不启动 research 或 auto-fix。
+- Codex 或 API 额度不足的状态是 `deferred_budget`，禁止自动切换 OpenAI/Anthropic、换 key 或
+  推荐付费低成本模型；只有人工显式批准才允许一次性 fallback。reset 后可安全恢复延后队列。

docs/async_service_deployment.md

@@ -64,8 +64,6 @@ CODEX_AUDIT_SERVICE_AUDIENCE=quant-codex-audit \
 CODEX_AUDIT_SERVICE_MODEL=gpt-5.4 \
 CODEX_AUDIT_SERVICE_REASONING_EFFORT=auto \
 CODEX_AUDIT_SERVICE_CODEX_ACCOUNT_USAGE=1 \
-CODEX_AUDIT_SERVICE_OPENAI_USAGE_WINDOW_DAYS=7 \
-CODEX_AUDIT_SERVICE_ANTHROPIC_USAGE_WINDOW_DAYS=7 \
 CODEX_AUDIT_SERVICE_JOB_DIR=/var/lib/codex-audit-bridge/jobs \
 bash scripts/deploy_codex_audit_service.sh deploy
 ```

scripts/run_codex_pr_review.py

@@ -527,15 +527,21 @@ def _review_capacity_is_unavailable(exc: ReviewError) -> bool:
         "[quota_or_capacity_failure]" in message
         or "usage limit" in message
         or "daily budget exceeded" in message
+        or "deferred_budget" in message
         or "codex service job failed [unknown_failure]: codex exec failed" in message
     )
 
 
 def _api_fallback_enabled() -> bool:
-    return parse_bool(env_value("CODEX_PR_REVIEW_API_FALLBACK_ENABLED", "true"))
+    # Codex quota exhaustion must become deferred_budget; never consume a
+    # paid API key unless a human explicitly enables the exception.
+    return parse_bool(env_value("CODEX_PR_REVIEW_API_FALLBACK_ENABLED", "false"))
 
 
 def _direct_api_primary_enabled() -> bool:
+    # Preserve the legacy API-only primary path when no Codex service URL is
+    # configured; only the secondary fallback after Codex quota failure is off
+    # by default.
     return parse_bool(env_value("CODEX_PR_REVIEW_DIRECT_API_PRIMARY_ENABLED", "true"))
 
 

service/adapters/codex_adapter.py

@@ -18,13 +18,35 @@
 
 SECRET_ENV_MARKERS = ("TOKEN", "SECRET", "PASSWORD", "PRIVATE_KEY", "CREDENTIAL", "API_KEY", "ADMIN_KEY")
 CODEX_REASONING_EFFORTS = frozenset({"minimal", "low", "medium", "high", "xhigh"})
+_LOCAL_CODEX_FAILURE_MARKERS = (
+    "authentication",
+    "bootstrap",
+    "command not found",
+    "configuration",
+    "invalid option",
+    "invalid sandbox",
+    "no such file",
+    "not logged in",
+    "permission denied",
+    "unrecognized option",
+    "unknown option",
+    "unsupported",
+)
 
 
 @dataclass(frozen=True)
 class CodexResult:
     success: bool
     output: str = ""
     error: str = ""
+    dispatch_started: bool = False
+    dispatch_uncertain: bool = False
+
+
+def _codex_dispatch_is_uncertain(detail: str) -> bool:
+    """Classify known local CLI failures as definitely not dispatched."""
+    text = detail.lower()
+    return not any(marker in text for marker in _LOCAL_CODEX_FAILURE_MARKERS)
 
 
 def _codex_env() -> dict[str, str]:
@@ -122,17 +144,21 @@ def execute(
 
         with tempfile.TemporaryDirectory() as tmp:
             output_last_message = Path(tmp) / "codex-final-message.md"
+            try:
+                command = _codex_command(
+                    output_last_message,
+                    sandbox=sandbox,
+                    model=model,
+                    reasoning_effort=reasoning_effort,
+                    output_schema=output_schema,
+                    cwd=cwd,
+                    images=images,
+                )
+            except (RuntimeError, ValueError) as exc:
+                return CodexResult(success=False, error=f"codex command configuration failed: {exc}")
             try:
                 completed = subprocess.run(
-                    _codex_command(
-                        output_last_message,
-                        sandbox=sandbox,
-                        model=model,
-                        reasoning_effort=reasoning_effort,
-                        output_schema=output_schema,
-                        cwd=cwd,
-                        images=images,
-                    ),
+                    command,
                     input=prompt,
                     text=True,
                     stdout=subprocess.PIPE,
@@ -142,14 +168,22 @@ def execute(
                     env=_codex_env(),
                 )
             except subprocess.TimeoutExpired as exc:
-                return CodexResult(success=False, error=f"codex exec timed out after {timeout}s: {exc}")
+                return CodexResult(
+                    success=False,
+                    error=f"codex exec timed out after {timeout}s: {exc}",
+                    dispatch_uncertain=True,
+                )
             except FileNotFoundError as exc:
                 return CodexResult(success=False, error=f"codex command not found: {exc}")
 
             if completed.returncode != 0:
                 detail = (completed.stdout[-4000:] + completed.stderr[-4000:]).strip()
-                return CodexResult(success=False, error=f"codex exec failed (rc={completed.returncode})" + (f":\n{detail}" if detail else ""))
+                return CodexResult(
+                    success=False,
+                    error=f"codex exec failed (rc={completed.returncode})" + (f":\n{detail}" if detail else ""),
+                    dispatch_uncertain=_codex_dispatch_is_uncertain(detail),
+                )
 
             if output_last_message.exists() and output_last_message.read_text(encoding="utf-8").strip():
-                return CodexResult(success=True, output=output_last_message.read_text(encoding="utf-8"))
-            return CodexResult(success=True, output=completed.stdout)
+                return CodexResult(success=True, output=output_last_message.read_text(encoding="utf-8"), dispatch_started=True)
+            return CodexResult(success=True, output=completed.stdout, dispatch_started=True)

service/adapters/llm_adapter.py

@@ -55,6 +55,8 @@ class LlmResult:
     success: bool = True
     error: str = ""
     latency_seconds: float = 0.0
+    dispatch_started: bool = False
+    dispatch_uncertain: bool = False
 
 
 class LlmAdapterError(RuntimeError):
@@ -105,6 +107,23 @@ def resolve_model(model: str) -> tuple[str, str]:
     return PROVIDER_ANTHROPIC, DEFAULT_ANTHROPIC_MODEL
 
 
+def _dispatch_is_uncertain(error: str) -> bool:
+    """Keep capacity only when provider acceptance cannot be disproven."""
+    text = error.lower()
+    if "api_key is not configured" in text:
+        return False
+    if any(f"http {status}" in text for status in range(400, 500)):
+        return False
+    return not any(marker in text for marker in (
+        "invalid model",
+        "unsupported model",
+        "validation",
+        "malformed",
+        "missing required",
+        "not configured",
+    ))
+
+
 # ── OpenAI ─────────────────────────────────────────────────────────────
 
 
@@ -273,7 +292,13 @@ def complete(
                 output = _anthropic_completion(resolved_model, system, user, max_tokens=max_tokens, timeout=timeout)
             else:
                 output = _openai_completion(resolved_model, system, user, max_tokens=max_tokens, timeout=timeout)
-            return LlmResult(provider=provider, model=resolved_model, output=output, latency_seconds=time.time() - started)
+            return LlmResult(
+                provider=provider,
+                model=resolved_model,
+                output=output,
+                latency_seconds=time.time() - started,
+                dispatch_started=True,
+            )
         except LlmAdapterError as exc:
             return LlmResult(
                 provider=provider,
@@ -282,6 +307,8 @@ def complete(
                 success=False,
                 error=str(exc),
                 latency_seconds=time.time() - started,
+                dispatch_started=False,
+                dispatch_uncertain=_dispatch_is_uncertain(str(exc)),
             )
 
     def parallel_review(
@@ -299,18 +326,24 @@ def parallel_review(
         """
         import concurrent.futures
 
-        results: list[LlmResult] = []
+        results: list[LlmResult | None] = [None] * len(reviewers)
         with concurrent.futures.ThreadPoolExecutor(max_workers=min(len(reviewers), 4)) as pool:
             futures = {
-                pool.submit(self.complete, model=model, system=system, user=user, max_tokens=max_tokens, timeout=timeout): (label, model)
-                for label, model in reviewers
+                pool.submit(self.complete, model=model, system=system, user=user, max_tokens=max_tokens, timeout=timeout): (index, label, model)
+                for index, (label, model) in enumerate(reviewers)
             }
             for f in concurrent.futures.as_completed(futures):
+                index, label, model = futures[f]
                 try:
-                    results.append(f.result())
+                    results[index] = f.result()
                 except Exception as exc:
-                    label, model = futures[f]
-                    results.append(
-                        LlmResult(provider=label, model=model, output="", success=False, error=str(exc))
+                    results[index] = (
+                        LlmResult(
+                            provider=label,
+                            model=model,
+                            output="",
+                            success=False,
+                            error=str(exc),
+                        )
                     )
-        return results
+        return [result for result in results if result is not None]