diff --git a/gateway/run.py b/gateway/run.py index e7b5167ff495..efe658d246a6 100644 --- a/gateway/run.py +++ b/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. diff --git a/tests/gateway/test_codex_pty_approval_bridge.py b/tests/gateway/test_codex_pty_approval_bridge.py new file mode 100644 index 000000000000..1d44d845cff7 --- /dev/null +++ b/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] diff --git a/tests/tools/test_codex_tui_approval.py b/tests/tools/test_codex_tui_approval.py new file mode 100644 index 000000000000..03df6f41f07b --- /dev/null +++ b/tests/tools/test_codex_tui_approval.py @@ -0,0 +1,192 @@ +"""Focused tests for the Hermes-owned native Codex PTY approval bridge.""" + +import threading +import time + +import pytest + +from tools.codex_tui_approval import ( + APPROVE_ONCE_KEY, + DENY_KEY, + CodexTuiApprovalDetector, + prepare_managed_codex_tui_command, +) +from tools.process_registry import ProcessRegistry, ProcessSession + + +def test_prepare_bridge_is_process_local_and_discord_only(monkeypatch): + monkeypatch.setattr( + "tools.codex_tui_approval._supported_codex_executable", + lambda _executable: "/opt/codex/bin/codex", + ) + session_key = "agent:main:discord:channel:123:user:456" + rewritten, enabled = prepare_managed_codex_tui_command( + "codex --no-alt-screen", + session_key, + approval_sink_available=True, + ) + + assert enabled is True + assert rewritten.startswith("/opt/codex/bin/codex -c ") + assert 'tui.keymap.approval.approve="ctrl-g"' in rewritten + assert 'tui.keymap.approval.approve_for_session="ctrl-o"' in rewritten + assert 'tui.keymap.approval.deny="ctrl-x"' in rewritten + assert rewritten.endswith("--no-alt-screen") + + for command, key, sink in ( + ("codex", "agent:main:telegram:dm:123:user:456", True), + ("codex exec pwd", session_key, True), + ("codex remote-control", session_key, True), + ("codex doctor", session_key, True), + ("codex && echo unsafe", session_key, True), + ("codex", session_key, False), + ): + untouched, enabled = prepare_managed_codex_tui_command( + command, key, approval_sink_available=sink + ) + assert (untouched, enabled) == (command, False) + + +def test_detector_waits_for_full_prompt_and_deduplicates_redraws(): + detector = CodexTuiApprovalDetector() + assert detector.feed("\x1b[2JWould you like to run the following ") is None + assert detector.feed("command?\r\n $ rm build.tmp\r\n") is None + + prompt = detector.feed("\x1b[4;1H Yes, just this once") + assert prompt is not None + assert prompt.kind == "command execution" + assert "$ rm build.tmp" in prompt.command + assert detector.feed("\x1b[4;1H Yes, just this once") is None + + detector.mark_resolved() + assert detector.feed("ordinary Codex output") is None + + +class _FakePty: + def __init__(self): + self.writes = [] + + def write(self, value): + self.writes.append(value) + + +def _wait_for(predicate, timeout=2): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if predicate(): + return True + time.sleep(0.01) + return False + + +@pytest.mark.parametrize( + ("choice", "expected_key"), + (("once", APPROVE_ONCE_KEY), ("deny", DENY_KEY)), +) +def test_approve_and_deny_unblock_only_the_originating_session(choice, expected_key): + from tools.approval import ( + has_blocking_approval, + resolve_gateway_approval, + unregister_gateway_notify, + ) + + registry = ProcessRegistry() + notified = [] + registry.on_approval = lambda session, data: notified.append((session.id, data)) + session = ProcessSession( + id="proc_codex_a", + command="codex", + session_key="agent:main:discord:channel:111:user:7", + ) + session.managed_codex_tui = True + session._codex_approval_detector = CodexTuiApprovalDetector() + session._pty = _FakePty() + + output = ( + "Would you like to run the following command?\r\n" + " $ python deploy.py\r\n" + " Yes, just this once" + ) + worker = threading.Thread( + target=registry._check_codex_tui_approval, + args=(session, output), + ) + worker.start() + try: + assert _wait_for( + lambda: has_blocking_approval(session.session_key) and bool(notified) + ) + assert notified[0][0] == session.id + assert session._codex_approval_pending is True + + # The launching agent turn can finish before Codex asks. Its normal + # callback teardown must not expire this process-lifetime request. + unregister_gateway_notify(session.session_key) + assert has_blocking_approval(session.session_key) + + other_key = "agent:main:discord:channel:222:user:7" + assert resolve_gateway_approval(other_key, "once") == 0 + worker.join(timeout=0.05) + assert worker.is_alive(), "another Discord session crossed the approval boundary" + assert session._pty.writes == [] + + assert resolve_gateway_approval(session.session_key, choice) == 1 + worker.join(timeout=2) + assert not worker.is_alive() + assert session._pty.writes == [expected_key.encode()] + assert session._codex_approval_pending is False + finally: + resolve_gateway_approval(session.session_key, "deny") + worker.join(timeout=2) + + +def test_public_stdin_cannot_race_a_pending_codex_approval(): + registry = ProcessRegistry() + session = ProcessSession(id="proc_codex", command="codex") + session.managed_codex_tui = True + session._codex_approval_pending = True + session._pty = _FakePty() + registry._running[session.id] = session + + result = registry.write_stdin(session.id, "y") + + assert result["status"] == "approval_pending" + assert session._pty.writes == [] + + +def test_kill_cancels_only_that_process_approval(monkeypatch): + from tools.approval import ( + _ApprovalEntry, + _gateway_queues, + has_blocking_approval, + ) + + registry = ProcessRegistry() + pty = _FakePty() + pty.terminate = lambda force: None + session = ProcessSession( + id="proc_codex", + command="codex", + session_key="discord-session", + ) + session.managed_codex_tui = True + session._pty = pty + registry._running[session.id] = session + own = _ApprovalEntry( + { + "approval_source": "managed_codex_tui", + "approval_source_id": session.id, + } + ) + unrelated = _ApprovalEntry({"approval_source": "terminal_guard"}) + _gateway_queues[session.session_key] = [own, unrelated] + monkeypatch.setattr(registry, "_write_checkpoint", lambda: None) + + try: + result = registry.kill_process(session.id) + assert result["status"] == "killed" + assert own.event.is_set() and own.result == "deny" + assert not unrelated.event.is_set() + assert has_blocking_approval(session.session_key) + finally: + _gateway_queues.pop(session.session_key, None) diff --git a/tests/tools/test_process_registry.py b/tests/tools/test_process_registry.py index 5fbde444dcf7..d86f47eeeaa5 100644 --- a/tests/tools/test_process_registry.py +++ b/tests/tools/test_process_registry.py @@ -916,7 +916,7 @@ def fake_popen(args, **kwargs): assert "sleep 5 &" in shell_cmd def test_pty_path_uses_rewritten_command(self, registry): - """PTY spawn path must also use the rewritten command (issue #68915).""" + """An ineligible approval bridge must preserve the PTY rewrite.""" mock_pty_proc = MagicMock() mock_pty_proc.pid = 5555 @@ -925,6 +925,9 @@ def test_pty_path_uses_rewritten_command(self, registry): fake_thread = MagicMock() fake_thread.daemon = False + # Exercise the bridge-selection path with a live gateway sink. This + # non-Codex command must be returned unchanged *after* safe rewriting. + registry.on_approval = MagicMock() with patch("tools.process_registry._find_shell", return_value="/bin/bash"), \ patch.dict("sys.modules", {"ptyprocess": mock_pty_module}), \ diff --git a/tools/approval.py b/tools/approval.py index 775aa8597270..782a3d612f21 100644 --- a/tools/approval.py +++ b/tools/approval.py @@ -2811,12 +2811,23 @@ def register_gateway_notify(session_key: str, cb) -> None: def unregister_gateway_notify(session_key: str) -> None: """Unregister the per-session gateway approval callback. - Signals ALL blocked threads for this session so they don't hang forever - (e.g. when the agent run finishes or is interrupted). + Signals blocked agent-turn threads for this session so they don't hang + forever (e.g. when the run finishes or is interrupted). Managed-process + approvals have their own gateway-lifetime notifier and survive turn end. """ with _lock: _gateway_notify_cbs.pop(session_key, None) - entries = _gateway_queues.pop(session_key, []) + queue = _gateway_queues.get(session_key, []) + entries = [ + entry + for entry in queue + if entry.data.get("approval_lifetime") != "process" + ] + survivors = [entry for entry in queue if entry not in entries] + if survivors: + _gateway_queues[session_key] = survivors + else: + _gateway_queues.pop(session_key, None) for entry in entries: entry.event.set() @@ -2863,6 +2874,36 @@ def resolve_gateway_approval(session_key: str, choice: str, return len(targets) +def cancel_gateway_approvals( + session_key: str, + *, + source: str, + source_id: str, +) -> int: + """Deny pending approvals owned by one managed background process. + + Unlike :func:`resolve_gateway_approval`, this is an internal lifecycle + operation rather than a user choice. Matching both source fields keeps a + killed PTY from cancelling an unrelated tool approval in the same chat. + """ + with _lock: + queue = _gateway_queues.get(session_key, []) + targets = [ + entry + for entry in queue + if entry.data.get("approval_source") == source + and entry.data.get("approval_source_id") == source_id + ] + for entry in targets: + queue.remove(entry) + if not queue: + _gateway_queues.pop(session_key, None) + for entry in targets: + entry.result = "deny" + entry.event.set() + return len(targets) + + def list_gateway_approvals(session_key: str) -> list[dict]: """Return replay-safe snapshots of unresolved approvals for one session.""" with _lock: @@ -2885,6 +2926,20 @@ def has_blocking_approval(session_key: str) -> bool: return bool(_gateway_queues.get(session_key)) +def await_managed_process_approval( + session_key: str, + notify_cb, + approval_data: dict, +) -> dict: + """Queue a managed-process approval on the normal gateway FIFO.""" + return _await_gateway_decision( + session_key, + notify_cb, + approval_data, + surface="managed_codex_tui", + ) + + def get_pending_gateway_approval(session_key: str) -> dict | None: """Return a copy of the oldest unresolved gateway approval for a session. diff --git a/tools/codex_tui_approval.py b/tools/codex_tui_approval.py new file mode 100644 index 000000000000..a5b37ec82127 --- /dev/null +++ b/tools/codex_tui_approval.py @@ -0,0 +1,257 @@ +"""Approval bridge primitives for Hermes-managed native Codex TUI PTYs. + +This is intentionally *not* a general Codex attachment mechanism. A native +Codex TUI exposes its approval requests only inside its terminal; unrelated +TUI processes have no structured endpoint Hermes can attach to. The bridge +works by adding process-local key bindings when Hermes itself launches a +Codex PTY, then recognizing Codex's approval screen in that owned PTY. +""" + +from __future__ import annotations + +import os +import re +import shlex +import shutil +import subprocess +from dataclasses import dataclass +from functools import lru_cache +from typing import Optional + +from tools.ansi_strip import sanitize_display_text + + +# Codex 0.146 is the first version against which Hermes verifies these public +# keymap paths. Older/unknown versions retain completely native behavior. +_MIN_KEYMAP_VERSION = (0, 146, 0) + +APPROVE_ONCE_KEY = "\x07" # ctrl-g +APPROVE_SESSION_KEY = "\x0f" # ctrl-o +DENY_KEY = "\x18" # ctrl-x + +_KEYMAP_OVERRIDES = ( + 'tui.keymap.approval.approve="ctrl-g"', + 'tui.keymap.approval.approve_for_session="ctrl-o"', + 'tui.keymap.approval.deny="ctrl-x"', +) + +_APPROVAL_HEADERS = { + "Would you like to run the following command?": "command execution", + "Would you like to make the following edits?": "file changes", + "Would you like to grant these permissions?": "permission grant", +} +_READY_MARKERS = ( + "Yes, just this once", + "No, continue without", +) +_SHELL_PUNCTUATION = frozenset({";", "&", "|", "<", ">", "(", ")"}) +_NON_INTERACTIVE_SUBCOMMANDS = frozenset( + { + "exec", + "review", + "mcp", + "plugin", + "app-server", + "mcp-server", + "remote-control", + "login", + "logout", + "completion", + "update", + "doctor", + "sandbox", + "debug", + "features", + "apply", + "cloud", + "remote", + } +) + + +def _is_discord_session_key(session_key: str) -> bool: + """Recognize canonical gateway keys without importing gateway modules.""" + parts = str(session_key or "").split(":") + return len(parts) >= 4 and parts[2].lower() == "discord" + + +def _parse_version(text: str) -> Optional[tuple[int, int, int]]: + match = re.search(r"\b(\d+)\.(\d+)(?:\.(\d+))?\b", text or "") + if not match: + return None + return tuple(int(part or 0) for part in match.groups()) + + +@lru_cache(maxsize=16) +def _supported_codex_executable(executable: str) -> Optional[str]: + """Return a verified local Codex path without session or network I/O.""" + resolved = executable if os.path.isabs(executable) else shutil.which(executable) + if not resolved: + return None + try: + result = subprocess.run( + [resolved, "--version"], + capture_output=True, + text=True, + timeout=2, + check=False, + ) + except (OSError, subprocess.SubprocessError): + return None + version = _parse_version(f"{result.stdout}\n{result.stderr}") + if result.returncode != 0 or version is None or version < _MIN_KEYMAP_VERSION: + return None + return os.path.realpath(resolved) + + +def _split_direct_command(command: str) -> Optional[list[str]]: + """Return argv only for a single, direct shell command. + + Shell composition is rejected because rewriting a compound command could + change its meaning, and because Hermes could no longer prove which Codex + process owns the PTY. + """ + if not command or "\n" in command or "\r" in command: + return None + try: + lexer = shlex.shlex(command, posix=True, punctuation_chars=";&|<>()") + lexer.whitespace_split = True + tokens = list(lexer) + except ValueError: + return None + if not tokens or any( + token and set(token).issubset(_SHELL_PUNCTUATION) for token in tokens + ): + return None + return tokens + + +def _is_native_codex_tui(argv: list[str]) -> bool: + if not argv or os.path.basename(argv[0]).lower() not in {"codex", "codex.exe"}: + return False + # Do not compete with an explicit user keymap override. + if any("tui.keymap.approval." in token for token in argv[1:]): + return False + + # Find the first positional token while skipping the value-bearing global + # options Codex accepts before its optional initial prompt. + value_options = { + "-c", "--config", "-m", "--model", "-p", "--profile", "-s", + "--sandbox", "-a", "--ask-for-approval", "-C", "--cd", "--add-dir", + "--image", "--oss-provider", + } + i = 1 + while i < len(argv): + token = argv[i] + if token == "--": + i += 1 + break + if token in value_options: + i += 2 + continue + if token.startswith("-"): + i += 1 + continue + break + return i >= len(argv) or argv[i].lower() not in _NON_INTERACTIVE_SUBCOMMANDS + + +def prepare_managed_codex_tui_command( + command: str, + session_key: str, + *, + approval_sink_available: bool, +) -> tuple[str, bool]: + """Add per-process approval bindings when this PTY is safely bridgeable.""" + if not approval_sink_available or not _is_discord_session_key(session_key): + return command, False + argv = _split_direct_command(command) + if not argv or not _is_native_codex_tui(argv): + return command, False + resolved_executable = _supported_codex_executable(argv[0]) + if not resolved_executable: + return command, False + + # CLI overrides have higher precedence than config.toml and affect only + # this child. Put them before the existing argv so an initial prompt is + # never mistaken for an option value. + # Launch the exact binary we probed, rather than allowing a login-shell + # alias/function named ``codex`` to receive approval keystrokes. + bridged_argv = [resolved_executable] + for override in _KEYMAP_OVERRIDES: + bridged_argv.extend(("-c", override)) + bridged_argv.extend(argv[1:]) + return shlex.join(bridged_argv), True + + +@dataclass(frozen=True) +class CodexApprovalPrompt: + kind: str + command: str + description: str + + +class CodexTuiApprovalDetector: + """Recognize a fully-rendered Codex approval prompt in PTY output.""" + + _MAX_TEXT = 32_000 + + def __init__(self) -> None: + self._text = "" + self._scan_offset = 0 + self._pending = False + + def feed(self, chunk: str) -> Optional[CodexApprovalPrompt]: + if not chunk: + return None + clean = sanitize_display_text(chunk).replace("\r\n", "\n").replace("\r", "\n") + self._text += clean + if len(self._text) > self._MAX_TEXT: + removed = len(self._text) - self._MAX_TEXT + self._text = self._text[removed:] + self._scan_offset = max(0, self._scan_offset - removed) + if self._pending: + return None + + segment = self._text[self._scan_offset:] + candidates = [ + (segment.rfind(header), header, kind) + for header, kind in _APPROVAL_HEADERS.items() + ] + header_at, header, kind = max(candidates, key=lambda item: item[0]) + if header_at < 0: + return None + body = segment[header_at + len(header):] + marker_positions = [body.find(marker) for marker in _READY_MARKERS] + marker_positions = [pos for pos in marker_positions if pos >= 0] + if not marker_positions: + return None + + body = body[: min(marker_positions)] + lines = [line.strip() for line in body.splitlines()] + lines = [line for line in lines if line] + detail = "\n".join(lines).strip() + if len(detail) > 1800: + detail = detail[:1785] + "... [truncated]" + display = detail or f"Codex requested {kind} in its native TUI" + self._pending = True + return CodexApprovalPrompt( + kind=kind, + command=display, + description=f"Native Codex TUI requested {kind}", + ) + + def mark_resolved(self) -> None: + """Ignore prior screen redraws and begin looking for the next prompt.""" + self._scan_offset = len(self._text) + self._pending = False + + +def key_for_choice(choice: Optional[str]) -> str: + """Map Hermes approval scope onto Codex's process-local choices.""" + if choice == "once": + return APPROVE_ONCE_KEY + if choice in {"session", "always"}: + # Hermes must never turn a chat approval into a global Codex rule. + return APPROVE_SESSION_KEY + return DENY_KEY diff --git a/tools/process_registry.py b/tools/process_registry.py index ed2b6d581019..3fba94c788da 100644 --- a/tools/process_registry.py +++ b/tools/process_registry.py @@ -429,6 +429,9 @@ class ProcessSession: _lock: threading.Lock = field(default_factory=threading.Lock) _reader_thread: Optional[threading.Thread] = field(default=None, repr=False) _pty: Any = field(default=None, repr=False) # ptyprocess handle (when use_pty=True) + managed_codex_tui: bool = False # Hermes-owned native Codex approval bridge + _codex_approval_detector: Any = field(default=None, repr=False) + _codex_approval_pending: bool = field(default=False, repr=False) class ProcessRegistry: @@ -504,6 +507,9 @@ def __init__(self): # terminal tab. Distinct from kill — the process keeps running; only the # UI view is dropped (the user can reopen it from the status stack). self.on_close = None + # Managed native-Codex approval sink set by the messaging gateway. + # Called from a PTY reader thread with (session, approval_data). + self.on_approval = None @staticmethod def _clean_shell_noise(text: str) -> str: @@ -1069,6 +1075,23 @@ def spawn_local( if use_pty: # Try PTY mode for interactive CLI tools try: + from tools.codex_tui_approval import ( + CodexTuiApprovalDetector, + prepare_managed_codex_tui_command, + ) + + # ``safe_command`` already contains the compound-background + # rewrite from issue #68915. The Codex bridge must transform + # that value (and return it unchanged when ineligible), never + # reintroduce the raw command on the PTY path. + pty_command, bridge_enabled = prepare_managed_codex_tui_command( + safe_command, + session_key, + approval_sink_available=callable(self.on_approval), + ) + if bridge_enabled: + session.managed_codex_tui = True + session._codex_approval_detector = CodexTuiApprovalDetector() if _IS_WINDOWS: from winpty import PtyProcess as _PtyProcessCls else: @@ -1076,7 +1099,7 @@ def spawn_local( user_shell = _find_shell() pty_env = _sanitize_subprocess_env(os.environ, env_vars) pty_env["PYTHONUNBUFFERED"] = "1" - pty_argv = [user_shell, "-lic", f"set +m; {safe_command}"] + pty_argv = [user_shell, "-lic", f"set +m; {pty_command}"] # Cgroup isolation for PTY mode (#70716, reviewer gap #1): # Wrap the PTY command in a systemd scope so interactive @@ -1573,6 +1596,7 @@ def _append_text(text: str): session.output_buffer = session.output_buffer[-session.max_output_chars:] self._check_watch_patterns(session, text) self._emit_output(session, text) + self._check_codex_tui_approval(session, text) try: while pty.isalive(): @@ -1609,6 +1633,64 @@ def _append_text(text: str): session.completion_reason = "exited" self._move_to_finished(session) + def _check_codex_tui_approval(self, session: ProcessSession, chunk: str) -> None: + """Block an owned Codex PTY on the gateway's exact session queue.""" + detector = session._codex_approval_detector + if not session.managed_codex_tui or detector is None: + return + prompt = detector.feed(chunk) + if prompt is None: + return + + from tools.approval import await_managed_process_approval + from tools.codex_tui_approval import key_for_choice + + approval_data = { + "command": prompt.command, + "description": f"{prompt.description} (process {session.id})", + "pattern_key": f"codex-pty:{session.id}", + "pattern_keys": [f"codex-pty:{session.id}"], + "allow_permanent": False, + "allow_session": True, + "approval_source": "managed_codex_tui", + "approval_source_id": session.id, + "approval_lifetime": "process", + } + + with session._lock: + session._codex_approval_pending = True + try: + def _notify(data: dict) -> None: + sink = self.on_approval + if sink is None: + raise RuntimeError("gateway approval sink is unavailable") + sink(session, data) + + decision = await_managed_process_approval( + session.session_key, + _notify, + approval_data, + ) + key = key_for_choice( + decision.get("choice") if decision.get("resolved") else None + ) + self._write_managed_codex_choice(session, key) + finally: + with session._lock: + session._codex_approval_pending = False + detector.mark_resolved() + + @staticmethod + def _write_managed_codex_choice(session: ProcessSession, key: str) -> None: + """Write one injected approval shortcut, bypassing the public guard.""" + try: + if _IS_WINDOWS: + session._pty.write(key) + else: + session._pty.write(key.encode("utf-8")) + except Exception as exc: + logger.debug("Could not write Codex PTY approval choice: %s", exc) + def _move_to_finished(self, session: ProcessSession): """Move a session from running to finished. @@ -2297,6 +2379,21 @@ def kill_process( self._completion_consumed.add(session_id) return result + if session.managed_codex_tui: + # Wake only this process's approval waiter before tearing down its + # PTY. Other approvals in the originating Discord session remain + # untouched. + try: + from tools.approval import cancel_gateway_approvals + + cancel_gateway_approvals( + session.session_key, + source="managed_codex_tui", + source_id=session.id, + ) + except Exception as exc: + logger.debug("Could not cancel Codex PTY approval during kill: %s", exc) + # Kill via PTY, Popen (local), or env execute (non-local) try: if session._pty: @@ -2387,6 +2484,15 @@ def write_stdin(self, session_id: str, data: str) -> dict: if session.exited: return {"status": "already_exited", "error": "Process has already finished"} + if session.managed_codex_tui and session._codex_approval_pending: + return { + "status": "approval_pending", + "error": ( + "This Codex PTY is waiting for its originating Discord " + "session to use /approve or /deny" + ), + } + # PTY mode -- write through pty handle. if hasattr(session, '_pty') and session._pty: try: