fix(codex): decline permission escalation with a valid empty grant

agent/transports/codex_app_server_session.py

@@ -706,7 +706,8 @@ def _respond_elicitation(self, params: dict) -> dict:
     _SERVER_REQUEST_HANDLERS: dict[str, Callable[..., dict]] = {
         "item/commandExecution/requestApproval": lambda self, p: {"decision": self._decide_exec_approval(p)},
         "item/fileChange/requestApproval": lambda self, p: {"decision": self._decide_apply_patch_approval(p)},
-        "item/permissions/requestApproval": lambda self, p: {"decision": "decline"},
+        # PermissionsRequestApprovalResponse declines by granting nothing, not by a decision enum.
+        "item/permissions/requestApproval": lambda self, p: {"permissions": {}},
         "mcpServer/elicitation/request": _respond_elicitation,
     }
 

tests/agent/transports/fixtures/codex_permissions_server.py

@@ -0,0 +1,36 @@
+"""Local stdio peer for the permission-denial wire contract; no provider access."""
+import json
+import sys
+
+
+def send(message):
+    print(json.dumps(message), flush=True)
+
+
+def note(method, **params):
+    send({"method": method, "params": {"threadId": "thread-1", "turnId": "turn-1", **params}})
+
+
+for line in sys.stdin:
+    message = json.loads(line)
+    method = message.get("method")
+    rid = message.get("id")
+    if method == "initialize":
+        send({"id": rid, "result": {"userAgent": "permission-fixture"}})
+    elif method == "thread/start":
+        send({"id": rid, "result": {"thread": {"id": "thread-1"}}})
+    elif method == "turn/start":
+        send({"id": rid, "result": {"turn": {"id": "turn-1"}}})
+        send({"id": "permission-1", "method": "item/permissions/requestApproval", "params": {
+            "threadId": "thread-1", "turnId": "turn-1", "itemId": "item-1",
+            "permissions": {"network": {"enabled": True}}, "reason": "needs network",
+        }})
+    elif rid == "permission-1":
+        # A deny must be a valid PermissionsRequestApprovalResponse and grant nothing.
+        if message.get("result") != {"permissions": {}}:
+            note("turn/completed", turn={"id": "turn-1", "status": "failed", "error": {
+                "message": "invalid permission denial: " + json.dumps(message),
+            }})
+        else:
+            note("item/completed", item={"id": "answer-1", "type": "agentMessage", "text": "DENIED-CLEANLY"})
+            note("turn/completed", turn={"id": "turn-1", "status": "completed", "error": None})

tests/agent/transports/test_codex_app_server_session.py

@@ -621,7 +621,40 @@ def test_compact_thread_ignores_foreign_child_completion(self):
 
 class TestServerRequestRouting:
 
+    @pytest.mark.parametrize("auto_approve", [False, True])
+    @pytest.mark.parametrize("permissions", [{}, {"network": {"enabled": True}}])
+    def test_permission_escalation_declines_with_empty_grant(self, auto_approve, permissions):
+        class PermissionClient(FakeClient):
+            def respond(self, request_id, result):
+                super().respond(request_id, result)
+                self.queue_notification(
+                    "turn/completed", threadId="t",
+                    turn={"id": "tu1", "status": "completed", "error": None},
+                )
+
+        client = PermissionClient()
+        client.queue_server_request(
+            "item/permissions/requestApproval", request_id="permission-1",
+            threadId="thread-fake-001", turnId="turn-fake-001",
+            itemId="item-1", permissions=permissions, reason="needs network",
+        )
+        callbacks = []
+
+        def approval(*args, **kwargs):
+            callbacks.append(args)
+            return "once"
 
+        session = make_session(
+            client, approval_callback=approval,
+            request_routing=_ServerRequestRouting(
+                auto_approve_exec=auto_approve, auto_approve_apply_patch=auto_approve,
+            ),
+        )
+        result = session.run_turn("hi", turn_timeout=2.0)
+        assert client.responses == [("permission-1", {"permissions": {}})]
+        assert callbacks == []  # Permission escalation remains unconditionally denied.
+        assert not client.error_responses
+        assert result.error is None
 
     def test_unknown_server_request_replied_with_error(self):
         client = FakeClient()

tests/agent/transports/test_codex_permission_wire.py

@@ -0,0 +1,29 @@
+"""Permission denial through the real session and JSON-RPC stdio client."""
+import subprocess
+import sys
+from pathlib import Path
+
+from agent.transports.codex_app_server_session import CodexAppServerSession
+
+
+def test_permission_denial_round_trip(tmp_path, monkeypatch):
+    # Only replace the executable at the spawn boundary. Session dispatch,
+    # JSON serialization, pipes, reader threads and reply correlation are real.
+    popen = subprocess.Popen
+    peer = Path(__file__).parent / "fixtures" / "codex_permissions_server.py"
+
+    def launch_peer(argv, **kwargs):
+        assert argv[:2] == ["permission-fixture", "app-server"]
+        return popen([sys.executable, "-u", str(peer)], **kwargs)
+
+    monkeypatch.setattr(subprocess, "Popen", launch_peer)
+    session = CodexAppServerSession(
+        cwd=str(tmp_path), codex_home=str(tmp_path), codex_bin="permission-fixture",
+    )
+    try:
+        result = session.run_turn("go", turn_timeout=10.0)
+        assert result.error is None
+        assert result.final_text == "DENIED-CLEANLY"
+        assert result.projected_messages == [{"role": "assistant", "content": "DENIED-CLEANLY"}]
+    finally:
+        session.close()

tests/e2e/core/providers/test_native_codex_app_server_faults.py

@@ -25,7 +25,6 @@
 
 KNOWN = {
     "q_approval": "#121296 approval in `chat -q` waits the full approvals.timeout instead of single_query_mode",
-    "permissions": "#121297 reply to item/permissions/requestApproval omits required `permissions`",
     "orphan": "#121298 `chat -q` exit never closes the codex session; own-session descendants orphaned",
     "failed_hidden": "#121299 failed turn after an agentMessage prints the message and hides the reason",
 }
@@ -126,7 +125,6 @@ def test_single_query_approval_resolves_without_waiting_for_a_human(runs):
         raise KnownSymptom(f"approval parked {waited:.1f}s on a prompt nobody can answer in -q")
 
 
-@pytest.mark.xfail(strict=True, raises=KnownSymptom, reason=KNOWN["permissions"])
 def test_permissions_request_reply_matches_protocol(runs):
     run = runs["permissions"]
     replies = run.fake.replies_to("item/permissions/requestApproval")