feat(gateway): bridge managed Codex TUI approvals

gateway/run.py

@@ -6917,6 +6917,7 @@ def __init__(self, config: Optional[GatewayConfig] = None):
         self.delivery_router = DeliveryRouter(self.config)
         self._running = False
         self._gateway_loop: Optional[asyncio.AbstractEventLoop] = None
+        self._managed_codex_approval_sink = None
         self._shutdown_event = asyncio.Event()
         self._exit_cleanly = False
         self._exit_with_failure = False
@@ -13249,6 +13250,18 @@ async def _connect_one_startup(p, p_cfg, adp):
         self._install_plugin_message_injector()
         self._update_runtime_status("running")
 
+        # Native Codex approvals can outlive the agent turn that launched the
+        # PTY, so they use a process-lifetime sink rather than a per-turn
+        # register_gateway_notify callback. Install it only after startup
+        # succeeds so an aborted start cannot leave a stale global callback.
+        from tools.process_registry import process_registry
+
+        def _managed_codex_sink(session, approval_data) -> None:
+            self._notify_managed_codex_approval_sync(session, approval_data)
+
+        self._managed_codex_approval_sink = _managed_codex_sink
+        process_registry.on_approval = _managed_codex_sink
+
         self._start_loop_heartbeat_task()
 
         # Emit gateway:startup hook
@@ -14773,6 +14786,18 @@ async def stop(
             return
 
         async def _stop_impl() -> None:
+            # Stop accepting new PTY approval prompts before subprocess
+            # teardown. kill_all() below denies any already-pending request.
+            try:
+                from tools.process_registry import process_registry
+
+                if process_registry.on_approval is getattr(
+                    self, "_managed_codex_approval_sink", None
+                ):
+                    process_registry.on_approval = None
+            except Exception as _e:
+                logger.debug("Codex PTY approval sink cleanup error: %s", _e)
+
             def _kill_tool_subprocesses(phase: str) -> list:
                 """Kill tool subprocesses + tear down terminal envs + browsers.
 
@@ -25148,6 +25173,80 @@ async def _transcribe_and_echo_pending_voice(
             logger.warning("%s transcription failed: %s", log_context, trans_exc)
             return text, []
 
+    def _notify_managed_codex_approval_sync(self, session, approval_data: dict) -> None:
+        """Bridge a PTY reader thread onto the gateway event loop."""
+        loop = self._gateway_loop
+        future = safe_schedule_threadsafe(
+            self._send_managed_codex_approval(session, approval_data),
+            loop,
+            logger=logger,
+            log_message="managed Codex PTY approval scheduling error",
+        )
+        if future is None:
+            raise RuntimeError("gateway event loop is unavailable")
+        sent = future.result(timeout=15)
+        if not sent:
+            raise RuntimeError("Discord approval prompt could not be delivered")
+
+    async def _send_managed_codex_approval(self, session, approval_data: dict) -> bool:
+        """Send an owned Codex PTY approval to its exact Discord origin."""
+        evt = {
+            "type": "managed_codex_approval",
+            "session_id": session.id,
+            "session_key": session.session_key,
+            "platform": session.watcher_platform,
+            "chat_id": session.watcher_chat_id,
+            "thread_id": session.watcher_thread_id,
+            "user_id": session.watcher_user_id,
+            "user_name": session.watcher_user_name,
+        }
+        source = self._build_process_event_source(evt)
+        if source is None or _gateway_platform_value(source.platform) != "discord":
+            logger.warning(
+                "Managed Codex PTY %s has no resolvable Discord origin",
+                session.id,
+            )
+            return False
+        adapter = self._adapter_for_source(source)
+        if adapter is None:
+            return False
+
+        adapter.pause_typing_for_chat(source.chat_id)
+        command = _redact_approval_command(approval_data.get("command", ""))
+        description = approval_data.get(
+            "description", "Native Codex TUI approval"
+        )
+        metadata = self._thread_metadata_for_source(source)
+        if getattr(type(adapter), "send_exec_approval", None) is not None:
+            try:
+                result = await adapter.send_exec_approval(
+                    chat_id=source.chat_id,
+                    command=command,
+                    session_key=session.session_key,
+                    description=description,
+                    metadata=metadata,
+                    allow_permanent=False,
+                    allow_session=True,
+                    smart_denied=False,
+                )
+                if result and result.success:
+                    return True
+            except Exception as exc:
+                logger.warning(
+                    "Managed Codex button approval failed; using text: %s", exc
+                )
+
+        prefix = getattr(adapter, "typed_command_prefix", "/")
+        message = _format_exec_approval_fallback(
+            command,
+            description,
+            prefix,
+            allow_permanent=False,
+            allow_session=True,
+        )
+        result = await adapter.send(source.chat_id, message, metadata=metadata)
+        return bool(result and result.success)
+
     def _build_process_event_source(self, evt: dict):
         """Resolve the canonical source for a synthetic background-process event.
 

tests/gateway/test_codex_pty_approval_bridge.py

@@ -0,0 +1,124 @@
+"""Gateway routing tests for managed native Codex PTY approvals."""
+
+from types import SimpleNamespace
+
+import pytest
+
+from gateway.config import Platform
+from gateway.run import GatewayRunner
+from tools.process_registry import ProcessSession
+
+
+class _DiscordAdapter:
+    def __init__(self):
+        self.calls = []
+        self.paused = []
+
+    def pause_typing_for_chat(self, chat_id):
+        self.paused.append(chat_id)
+
+    async def send_exec_approval(self, **kwargs):
+        self.calls.append(kwargs)
+        return SimpleNamespace(success=True)
+
+
+class _TextDiscordAdapter:
+    typed_command_prefix = "/"
+
+    def __init__(self):
+        self.sent = []
+
+    def pause_typing_for_chat(self, _chat_id):
+        pass
+
+    async def send(self, chat_id, message, metadata=None):
+        self.sent.append((chat_id, message, metadata))
+        return SimpleNamespace(success=True)
+
+
+@pytest.mark.asyncio
+async def test_managed_codex_prompt_uses_process_session_origin_and_queue_key():
+    runner = object.__new__(GatewayRunner)
+    source = SimpleNamespace(
+        platform=Platform.DISCORD,
+        chat_id="channel-123",
+        thread_id="thread-9",
+        message_id=None,
+        profile=None,
+    )
+    adapter = _DiscordAdapter()
+    captured_event = {}
+
+    def build_source(event):
+        captured_event.update(event)
+        return source
+
+    runner._build_process_event_source = build_source
+    runner._adapter_for_source = lambda actual: adapter if actual is source else None
+    runner._thread_metadata_for_source = lambda actual: {"thread_id": actual.thread_id}
+    session = ProcessSession(
+        id="proc_codex",
+        command="codex",
+        session_key="agent:main:discord:thread:thread-9:user:42",
+    )
+
+    sent = await runner._send_managed_codex_approval(
+        session,
+        {"command": "$ make deploy", "description": "Codex command execution"},
+    )
+
+    assert sent is True
+    assert captured_event["session_key"] == session.session_key
+    assert adapter.paused == [source.chat_id]
+    assert adapter.calls == [
+        {
+            "chat_id": source.chat_id,
+            "command": "$ make deploy",
+            "session_key": session.session_key,
+            "description": "Codex command execution",
+            "metadata": {"thread_id": source.thread_id},
+            "allow_permanent": False,
+            "allow_session": True,
+            "smart_denied": False,
+        }
+    ]
+
+
+@pytest.mark.asyncio
+async def test_managed_codex_prompt_refuses_non_discord_origin():
+    runner = object.__new__(GatewayRunner)
+    runner._build_process_event_source = lambda _event: SimpleNamespace(
+        platform=Platform.TELEGRAM,
+    )
+    session = ProcessSession(
+        id="proc_codex",
+        command="codex",
+        session_key="agent:main:telegram:dm:123:user:42",
+    )
+
+    assert await runner._send_managed_codex_approval(session, {}) is False
+
+
+@pytest.mark.asyncio
+async def test_managed_codex_prompt_text_fallback_names_channel_commands():
+    runner = object.__new__(GatewayRunner)
+    source = SimpleNamespace(
+        platform=Platform.DISCORD,
+        chat_id="channel-123",
+        thread_id=None,
+    )
+    adapter = _TextDiscordAdapter()
+    runner._build_process_event_source = lambda _event: source
+    runner._adapter_for_source = lambda _source: adapter
+    runner._thread_metadata_for_source = lambda _source: None
+    session = ProcessSession(
+        id="proc_codex",
+        command="codex",
+        session_key="agent:main:discord:group:channel-123:user:42",
+    )
+
+    assert await runner._send_managed_codex_approval(
+        session, {"command": "$ make deploy"}
+    )
+    assert "`/approve`" in adapter.sent[0][1]
+    assert "`/deny`" in adapter.sent[0][1]