fix(codex): masked "invalid_prompt: Request blocked." replay rejection reaches the replay-strip recovery

The ChatGPT Codex backend answers a rejected encrypted-reasoning replay with the
same bare 400 {code: invalid_prompt, message: "Request blocked."} it uses for
genuine blocks. classify_api_error bucketed it as format_error, so the one-shot
repair that already exists for invalid_encrypted_content (disable replay, strip
cached codex_reasoning_items, retry once) never ran and the turn aborted.

Classify exactly that envelope from provider openai-codex — SDK 400 body, SSE
``error`` frame, or the ``response.failed`` text — as invalid_encrypted_content
while keeping format_error's abort-and-fallback hints, so the only behavioural
delta is turn_recovery's replay strip, which still requires cached reasoning
items. A block with nothing to strip aborts exactly as before; other providers,
other messages/codes and the #18028 safety refusals are untouched.

evals/codex_masked_replay_review.py drives a real AIAgent (codex_responses)
against a synthetic local Responses server for the before/after check.
Refs #92353, #92357, #94077.

agent/error_classifier.py

@@ -486,6 +486,13 @@ def _provider_special_cases(c: _Ctx) -> Optional[Verdict]:
     # to format_error and a status-less block isn't left retryable (#18028).
     if any(p in msg for p in _CONTENT_POLICY_BLOCKED_PATTERNS):
         return _V_CONTENT_BLOCKED
+    # ChatGPT Codex masks a rejected encrypted-reasoning replay behind the same bare
+    # ``invalid_prompt: Request blocked.`` it uses for real blocks (#92353). Exact envelope
+    # + provider only. The verdict keeps format_error's abort-and-fallback hints; the one
+    # extra thing it buys is turn_recovery's replay strip, which still requires cached
+    # ``codex_reasoning_items`` — a genuine block with nothing to strip behaves as before.
+    if _is_codex_masked_replay_rejection(c):
+        return _v(_R.invalid_encrypted_content, **_ABORT_FALLBACK)
     # Anthropic thinking-block 400s (signature mismatch after transcript
     # mutation). Not gated on provider — OpenRouter proxies Anthropic errors.
     if status == 400 and "thinking" in msg and any(p in msg for p in _THINKING_MUTATION_WORDS):
@@ -777,6 +784,22 @@ def _is_server_injected_param_rejection(error_msg: str, provider: str) -> bool:
     return False
 
 
+_CODEX_MASKED_REPLAY_MESSAGE = "request blocked."
+
+
+def _is_codex_masked_replay_rejection(c: "_Ctx") -> bool:
+    """HTTP 400 / status-less ``{code: invalid_prompt, message: "Request blocked."}`` from
+    ``openai-codex`` — as an SDK error body, a Responses ``error`` SSE frame, or the
+    ``response.failed`` text ``"invalid_prompt: Request blocked."``."""
+    if c.provider_slug != "openai-codex" or c.status_code not in (None, 400):
+        return False
+    # The OpenAI SDK unwraps ``body["error"]`` on status errors; stream frames keep the envelope.
+    body_msg = next((str(m).strip().lower() for m in _body_message_candidates(c.body or {}) if m), "")
+    return (c.code == "invalid_prompt" and body_msg == _CODEX_MASKED_REPLAY_MESSAGE) or (
+        c.msg.strip() == f"invalid_prompt: {_CODEX_MASKED_REPLAY_MESSAGE}"
+    )
+
+
 def _error_obj(body: Any) -> dict:
     """``body["error"]`` when it is a dict, else ``{}``."""
     err = body.get("error") if isinstance(body, dict) else None

evals/codex_masked_replay_review.py

@@ -0,0 +1,138 @@
+"""Local HTTP contract review of masked-error recovery; NOT live-provider proof.
+
+Run in an isolated HOME/HERMES_HOME under the campaign test lock. Responses are
+explicit synthetic fixtures; the client, transport, and AIAgent loop are real.
+"""
+import copy
+import hashlib
+import json
+import os
+from pathlib import Path
+import sys
+import threading
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+import run_agent
+from agent import error_classifier
+from openai import APIError
+import httpx
+
+ITEM = {"type": "reasoning", "id": "rs_review", "encrypted_content": "signed-control-opaque-do-not-alter", "summary": []}
+ERROR = {"message": "Request blocked.", "type": "invalid_request_error", "param": None, "code": "invalid_prompt"}
+
+
+def digest(value):
+    return hashlib.sha256(json.dumps(value, sort_keys=True).encode()).hexdigest()
+
+
+def scenario(name, provider="openai-codex", replay=True):
+    requests = []
+
+    class Handler(BaseHTTPRequestHandler):
+        def log_message(self, *_args):
+            pass
+
+        def do_POST(self):
+            payload = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
+            if "input" not in payload:
+                raw = json.dumps({"id": "chat-local", "object": "chat.completion", "created": 1,
+                                  "model": "gpt-5-codex", "choices": [{"index": 0, "finish_reason": "stop",
+                                  "message": {"role": "assistant", "content": "local auxiliary fixture"}}]}).encode()
+                self.send_response(200)
+                self.send_header("Content-Type", "application/json")
+                self.send_header("Content-Length", str(len(raw)))
+                self.end_headers()
+                self.wfile.write(raw)
+                return
+            requests.append(payload)
+            has_replay = any(i.get("type") == "reasoning" for i in payload["input"])
+            error = copy.deepcopy(ERROR)
+            if name == "explicit_encrypted":
+                error.update(code="invalid_encrypted_content", message="The encrypted content could not be verified.")
+            reject = name in {"unrelated_block", "failed_frame"} or (name not in {"success", "failed_frame"} and has_replay)
+            if reject and name == "failed_frame":
+                # The shape openai/codex issues show for a blocked turn: response.in_progress, then a
+                # ``response.failed`` terminal frame carrying the error (no HTTP status).
+                failed = {"id": "resp_review", "object": "response", "created_at": 1, "status": "failed",
+                          "model": "gpt-5-codex", "output": [], "error": error}
+                events = [{"type": "response.created", "response": dict(failed, status="in_progress", error=None), "sequence_number": 0},
+                          {"type": "response.in_progress", "response": dict(failed, status="in_progress", error=None), "sequence_number": 1},
+                          {"type": "response.failed", "response": failed, "sequence_number": 2}]
+                raw = "".join("data: " + json.dumps(e) + "\n\n" for e in events).encode()
+                self.send_response(200)
+                self.send_header("Content-Type", "text/event-stream")
+            elif reject:
+                raw = json.dumps({"error": error}).encode()
+                self.send_response(400)
+                self.send_header("Content-Type", "application/json")
+            else:
+                response = {"id": "resp_review", "object": "response", "created_at": 1,
+                            "status": "completed", "model": "gpt-5-codex", "output": [
+                                {"id": "msg_review", "type": "message", "role": "assistant", "status": "completed",
+                                 "content": [{"type": "output_text", "text": "local fixture success", "annotations": []}]}],
+                            "usage": {"input_tokens": 10, "output_tokens": 3, "total_tokens": 13}}
+                created = dict(response, status="in_progress", output=[])
+                events = [{"type": "response.created", "response": created, "sequence_number": 0},
+                          {"type": "response.output_item.done", "item": response["output"][0], "output_index": 0, "sequence_number": 1},
+                          {"type": "response.completed", "response": response, "sequence_number": 2}]
+                raw = "".join("data: " + json.dumps(e) + "\n\n" for e in events).encode()
+                self.send_response(200)
+                self.send_header("Content-Type", "text/event-stream")
+            self.send_header("Content-Length", str(len(raw)))
+            self.send_header("x-request-id", f"local-{name}-{len(requests)}")
+            self.end_headers()
+            self.wfile.write(raw)
+
+    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    agent = run_agent.AIAgent(
+        model="gpt-5-codex", provider=provider, api_mode="codex_responses",
+        base_url=f"http://127.0.0.1:{server.server_port}/v1", api_key="local-fixture-key",
+        enabled_toolsets=["terminal"], max_iterations=2, quiet_mode=True,
+        reasoning_config={"effort": "high"}, skip_context_files=True,
+        skip_memory=True, skip_background_review=True, save_trajectories=False,
+        session_id=f"local-review-{name}-{provider}-{replay}",
+    )
+    agent._api_max_retries = 3  # show whether a deterministic rejection burns identical retries
+    history = [{"role": "user", "content": "hello"}, {"role": "assistant", "content": "hello"}]
+    if replay:
+        history[1]["codex_reasoning_items"] = [copy.deepcopy(ITEM)]
+    try:
+        result = agent.run_conversation("reply briefly", system_message="Local contract review.", conversation_history=history)
+        reasoning_counts = [sum(i.get("type") == "reasoning" for i in p["input"]) for p in requests]
+        stable = {k: len({digest(p.get(k)) for p in requests}) <= 1
+                  for k in ["reasoning", "instructions", "tools"]}
+        signed_before = [i for i in requests[0]["input"] if i.get("type") == "reasoning"] if requests else []
+        return {"name": name, "provider": provider, "replay": replay, "completed": result.get("completed"),
+                "requests": len(requests), "reasoning_counts": reasoning_counts,
+                "stable_retry_fields": stable, "reasoning_parameters": [p.get("reasoning") for p in requests],
+                "tool_count": len(requests[0].get("tools", [])) if requests else 0,
+                "signed_payload_unchanged": [i.get("encrypted_content") for i in signed_before] == ([ITEM["encrypted_content"]] if replay else []),
+                "replay_enabled_after": agent._codex_reasoning_replay_enabled,
+                "canonical_replay_after": any(m.get("codex_reasoning_items") for m in result["messages"])}
+    finally:
+        agent.close()
+        server.shutdown()
+        server.server_close()
+        thread.join()
+
+
+def main():
+    print(json.dumps({"run_agent": run_agent.__file__, "classifier": error_classifier.__file__,
+                      "home": os.environ["HERMES_HOME"]}))
+    results = [scenario("success"), scenario("masked_replay"), scenario("explicit_encrypted"),
+               scenario("unrelated_block"), scenario("unrelated_block", replay=False),
+               scenario("masked_replay", provider="custom"), scenario("failed_frame"), scenario("failed_frame", replay=False)]
+    e = APIError("Request blocked.", request=httpx.Request("POST", "http://127.0.0.1"), body=ERROR)
+    output = {"surface": "real AIAgent and SDK against synthetic local HTTP/SSE server; not provider proof",
+              "statusless_reason": error_classifier.classify_api_error(e, provider="openai-codex").reason.value,
+              "scenarios": results}
+    Path(sys.argv[1]).write_text(json.dumps(output, indent=2) + "\n", encoding="utf-8")
+    print(json.dumps(output, indent=2))
+
+
+if __name__ == "__main__":
+    main()

tests/agent/test_error_classifier.py

@@ -691,6 +691,31 @@ class TestClassifyApiError:
         assert result.retryable is True
         assert result.should_fallback is False
 
+    # ── Codex masked encrypted-reasoning replay rejection (#92353) ──
+
+    _CODEX_MASKED = {"message": "Request blocked.", "type": "invalid_request_error", "param": None, "code": "invalid_prompt"}
+
+    @pytest.mark.parametrize("error", [
+        MockAPIError("Error code: 400 - Request blocked.", status_code=400, body=_CODEX_MASKED),  # SDK unwraps body["error"]
+        MockAPIError("Request blocked.", status_code=None, body={"error": _CODEX_MASKED}),  # SSE ``error`` frame
+        RuntimeError("invalid_prompt: Request blocked."),  # ``response.failed`` terminal frame
+    ], ids=["http400", "sse-frame", "response-failed"])
+    def test_codex_masked_replay_rejection_reaches_replay_strip(self, error):
+        result = classify_api_error(error, provider="openai-codex", model="gpt-5.5")
+        assert result.reason == FailoverReason.invalid_encrypted_content
+        assert result.retryable is False and result.should_fallback is True  # format_error's terminal hints kept
+
+    @pytest.mark.parametrize(("provider", "body", "expected"), [
+        ("custom", _CODEX_MASKED, FailoverReason.format_error),  # same envelope, other provider
+        ("openai-codex", {**_CODEX_MASKED, "message": "Invalid prompt: too long."}, FailoverReason.format_error),
+        ("openai-codex", {**_CODEX_MASKED, "code": "server_error"}, FailoverReason.format_error),
+        ("openai-codex", {**_CODEX_MASKED, "message": "Request blocked. Your request was flagged by our safety system."},
+         FailoverReason.content_policy_blocked),  # #18028 refusal still wins
+    ], ids=["other-provider", "other-message", "other-code", "safety-refusal"])
+    def test_codex_masked_replay_rejection_stays_narrow(self, provider, body, expected):
+        e = MockAPIError("Error code: 400 - " + body["message"], status_code=400, body=body)
+        assert classify_api_error(e, provider=provider, model="gpt-5.5").reason == expected
+
     # ── Reasoning-mandatory route rejecting a disable ──
 
     def test_reasoning_mandatory_400_is_retryable_not_format_error(self):