Classify Codex exec failures separately

scripts/codex_audit_service.py

@@ -670,8 +670,29 @@ def _job_dedupe_key(payload: dict[str, Any]) -> str:
     return hashlib.sha256(json.dumps(key_parts, separators=(",", ":")).encode("utf-8")).hexdigest()
 
 
+def _classify_codex_exec_failure(text: str) -> str:
+    if any(word in text for word in ("quota", "rate limit", "too many active", "budget")):
+        return "quota_or_capacity_failure"
+    if any(word in text for word in ("timeout", "timed out", "temporarily", "unavailable", "connection", "network")):
+        return "transient_service_failure"
+    codex_auth_config_signals = (
+        "not authenticated",
+        "authentication failed",
+        "login required",
+        "please log in",
+        "invalid api key",
+        "401 unauthorized",
+        "403 forbidden",
+    )
+    if any(signal in text for signal in codex_auth_config_signals):
+        return "auth_or_config_failure"
+    return "unknown_failure"
+
+
 def _classify_failure(error: str) -> str:
     text = error.lower()
+    if "codex exec failed" in text:
+        return _classify_codex_exec_failure(text)
     auth_config_signals = (
         "permission denied",
         "unauthorized",

scripts/run_monthly_codex_audit.py

@@ -169,8 +169,29 @@ class BridgeError(RuntimeError):
     pass
 
 
+def _classify_codex_exec_failure(text: str) -> str:
+    if any(word in text for word in ("quota", "rate limit", "too many active", "budget")):
+        return "quota_or_capacity_failure"
+    if any(word in text for word in ("timeout", "timed out", "temporarily", "unavailable", "connection", "network")):
+        return "transient_service_failure"
+    codex_auth_config_signals = (
+        "not authenticated",
+        "authentication failed",
+        "login required",
+        "please log in",
+        "invalid api key",
+        "401 unauthorized",
+        "403 forbidden",
+    )
+    if any(signal in text for signal in codex_auth_config_signals):
+        return "auth_or_config_failure"
+    return "unknown_failure"
+
+
 def classify_service_failure(error: str) -> str:
     text = error.lower()
+    if "codex exec failed" in text:
+        return _classify_codex_exec_failure(text)
     auth_config_signals = (
         "permission denied",
         "unauthorized",

service/ai_gateway_service.py

@@ -434,8 +434,29 @@ def _job_dedupe_key(payload: dict[str, Any]) -> str:
     return hashlib.sha256(json.dumps(parts, separators=(",", ":")).encode("utf-8")).hexdigest()
 
 
+def _classify_codex_exec_failure(text: str) -> str:
+    if any(word in text for word in ("quota", "rate limit", "too many active", "budget")):
+        return "quota_or_capacity_failure"
+    if any(word in text for word in ("timeout", "timed out", "temporarily", "unavailable", "connection", "network")):
+        return "transient_service_failure"
+    codex_auth_config_signals = (
+        "not authenticated",
+        "authentication failed",
+        "login required",
+        "please log in",
+        "invalid api key",
+        "401 unauthorized",
+        "403 forbidden",
+    )
+    if any(signal in text for signal in codex_auth_config_signals):
+        return "auth_or_config_failure"
+    return "unknown_failure"
+
+
 def _classify_failure(error: str) -> str:
     text = error.lower()
+    if "codex exec failed" in text:
+        return _classify_codex_exec_failure(text)
     auth_config_signals = (
         "permission denied",
         "unauthorized",

tests/test_run_monthly_codex_audit.py

@@ -538,6 +538,14 @@ def test_service_failure_classification_identifies_infra_failures(self) -> None:
     def test_service_failure_classification_ignores_source_code_secret_words(self) -> None:
         message = "codex exec failed: BLOCKED_PATH_RE = r'.*token.*|.*secret.*'"
         self.assertEqual(classify_service_failure(message), "unknown_failure")
+        allowlist_message = "codex exec failed: raise ValueError('not allowed by source allowlist')"
+        self.assertEqual(classify_service_failure(allowlist_message), "unknown_failure")
+        self.assertEqual(codex_audit_service._classify_failure(allowlist_message), "unknown_failure")
+        self.assertEqual(classify_service_failure("codex exec failed: too many active requests"), "quota_or_capacity_failure")
+        self.assertEqual(
+            classify_service_failure("source_repository foo is not allowed by service allowlist"),
+            "auth_or_config_failure",
+        )
 
     def test_codex_audit_service_async_job_lifecycle(self) -> None:
         with tempfile.TemporaryDirectory() as tmp: