[codex] Route app-server approvals through gateway

agent/codex_runtime.py

@@ -254,14 +254,43 @@ def run_codex_app_server_turn(
 
         cwd = getattr(agent, "session_cwd", None) or str(resolve_agent_cwd())
         # Approval callback: defer to Hermes' standard prompt flow if a
-        # CLI thread has installed one. Gateway / cron contexts get the
-        # codex-side fail-closed default.
+        # CLI thread has installed one. Gateway contexts do not have that
+        # terminal-local callback, so bridge through tools.approval's
+        # per-session queue instead. Cron / non-interactive contexts still
+        # get the codex-side fail-closed default.
         try:
             from tools.terminal_tool import _get_approval_callback
             approval_callback = _get_approval_callback()
         except Exception:
             approval_callback = None
 
+        if approval_callback is None:
+            try:
+                from tools.approval import (
+                    get_current_session_key,
+                    prompt_gateway_approval,
+                )
+                session_key = get_current_session_key(default="")
+                if session_key:
+                    def _gateway_approval_callback(
+                        command: str,
+                        description: str,
+                        *,
+                        allow_permanent: bool = True,
+                    ) -> str:
+                        return prompt_gateway_approval(
+                            command,
+                            description,
+                            session_key=session_key,
+                            pattern_key=f"codex_app_server:{command}",
+                            allow_permanent=allow_permanent,
+                            surface="codex_app_server",
+                        )
+
+                    approval_callback = _gateway_approval_callback
+            except Exception:
+                approval_callback = None
+
         def _on_codex_event(note: dict) -> None:
             # Bridge Codex app-server item/started notifications to Hermes
             # tool-progress so gateways show verbose "running X" breadcrumbs

gateway/run.py

@@ -292,9 +292,30 @@ def _redact_gateway_user_facing_secrets(text: str) -> str:
     redacted = str(text or "")
     for pattern in _GATEWAY_SECRET_PATTERNS:
         redacted = pattern.sub(lambda m: (m.group(1) if m.lastindex else "") + "[REDACTED]", redacted)
+    redacted = _redact_gateway_user_facing_cache_paths(redacted)
     return redacted
 
 
+_GATEWAY_CACHE_DOCUMENT_PATH_RE = re.compile(
+    r"(?<![\w:/.-])"
+    r"(?:"
+    r"(?:~|/home/[^/\s\]\)\"']+|/root)/\.hermes/cache/documents/"
+    r"|/tmp/hermes[^/\s\]\)\"']*/cache/documents/"
+    r")"
+    r"[^\s\]\)\"']+"
+)
+
+
+def _redact_gateway_user_facing_cache_paths(text: str) -> str:
+    """Hide inbound cache file paths from user-visible gateway output.
+
+    The model may need local cache paths internally to inspect uploaded files,
+    but those paths are implementation details and should not be echoed into
+    Telegram chats.
+    """
+    return _GATEWAY_CACHE_DOCUMENT_PATH_RE.sub("[cached document]", str(text or ""))
+
+
 def _redact_approval_command(cmd: "str | None") -> str:
     """Redact credentials from a command before it goes into an approval prompt.
 

plugins/platforms/telegram/adapter.py

@@ -5921,7 +5921,7 @@ async def _cache_replied_media(self, msg: Any, event: MessageEvent) -> None:
                 event.message_type = MessageType.VIDEO
         event.text = self._append_observed_note(
             event.text,
-            f"[Replied-to {cached.kind} '{cached.display_name}' saved at: {cached.path}]",
+            f"[Replied-to {cached.kind} '{cached.display_name}' is available as an internal attachment]",
         )
         logger.info("[Telegram] Cached replied-to %s at %s", cached.kind, cached.path)
 

tests/gateway/test_document_context_note.py

@@ -16,6 +16,7 @@
 
 gateway_run = importlib.import_module("gateway.run")
 _build_document_context_note = gateway_run._build_document_context_note
+_sanitize_gateway_final_response = gateway_run._sanitize_gateway_final_response
 
 
 class TestTextDocumentNote:
@@ -55,3 +56,22 @@ def test_binary_note_distinct_from_text_note(self):
         # The text path claims content is inlined; the binary path must not.
         assert "included below" in text_note
         assert "included below" not in pdf_note
+
+
+class TestTelegramDocumentPathRedaction:
+    def test_telegram_final_response_redacts_cached_document_paths(self):
+        response = (
+            "פתחתי את /home/gidon/.hermes/cache/documents/doc_a90e9ffc40fb_jobs.json "
+            "והקובץ תקין."
+        )
+
+        sanitized = _sanitize_gateway_final_response("telegram", response)
+
+        assert "/home/gidon/.hermes/cache/documents/" not in sanitized
+        assert "doc_a90e9ffc40fb_jobs.json" not in sanitized
+        assert "[cached document]" in sanitized
+
+    def test_non_telegram_final_response_keeps_existing_behavior(self):
+        response = "See /home/gidon/.hermes/cache/documents/doc_a90e9ffc40fb_jobs.json"
+
+        assert _sanitize_gateway_final_response("discord", response) == response

tests/gateway/test_telegram_group_gating.py

@@ -1050,6 +1050,9 @@ async def _run():
         assert event.media_urls == [str(cached_path)]
         assert event.media_types == ["image/png"]
         assert event.message_type == MessageType.PHOTO
+        assert "internal attachment" in event.text
+        assert str(cached_path) not in event.text
+        assert "saved at:" not in event.text
 
     asyncio.run(_run())
 

tests/run_agent/test_codex_app_server_integration.py

@@ -134,6 +134,65 @@ def fake_run_turn(self, user_input: str, **kwargs):
         assert agent.context_compressor.last_total_tokens == 130
         assert agent.context_compressor.context_length == 200000
 
+    def test_gateway_session_wires_approval_callback(self, monkeypatch):
+        from tools.approval import (
+            reset_current_session_key,
+            set_current_session_key,
+        )
+
+        captured = {}
+
+        def fake_prompt_gateway_approval(command, description, **kwargs):
+            captured["approval_prompt"] = {
+                "command": command,
+                "description": description,
+                **kwargs,
+            }
+            return "session"
+
+        def fake_init(self, **kwargs):
+            self._approval_callback = kwargs.get("approval_callback")
+            captured["callback"] = self._approval_callback
+
+        def fake_run_turn(self, user_input: str, **kwargs):
+            captured["choice"] = self._approval_callback(
+                "touch /tmp/codex-approval-test",
+                "Codex requests exec in /tmp",
+                allow_permanent=False,
+            )
+            return TurnResult(
+                final_text="ok",
+                projected_messages=[{"role": "assistant", "content": "ok"}],
+                turn_id="turn-stub-1",
+                thread_id="thread-stub-1",
+            )
+
+        monkeypatch.setattr(
+            "tools.terminal_tool._get_approval_callback",
+            lambda: None,
+        )
+        monkeypatch.setattr(
+            "tools.approval.prompt_gateway_approval",
+            fake_prompt_gateway_approval,
+        )
+        monkeypatch.setattr(CodexAppServerSession, "__init__", fake_init)
+        monkeypatch.setattr(CodexAppServerSession, "run_turn", fake_run_turn)
+
+        token = set_current_session_key("telegram:chat:123")
+        try:
+            agent = _make_codex_agent()
+            with patch.object(agent, "_spawn_background_review", return_value=None):
+                result = agent.run_conversation("needs filesystem escalation")
+        finally:
+            reset_current_session_key(token)
+
+        assert result["final_response"] == "ok"
+        assert callable(captured["callback"])
+        assert captured["choice"] == "session"
+        assert captured["approval_prompt"]["session_key"] == "telegram:chat:123"
+        assert captured["approval_prompt"]["allow_permanent"] is False
+        assert captured["approval_prompt"]["surface"] == "codex_app_server"
+
     def test_projected_messages_are_spliced(self, fake_session):
         agent = _make_codex_agent()
         with patch.object(agent, "_spawn_background_review", return_value=None):

tests/tools/test_approval.py

@@ -1685,6 +1685,71 @@ def _check():
         assert "NOT consented" in r["message"]
         assert "rephrase" in r["message"].lower()
 
+    def test_prompt_gateway_approval_uses_queue_and_returns_choice(self):
+        from tools import approval as mod
+
+        notified = []
+
+        def notify(data):
+            notified.append(data)
+
+        mod.register_gateway_notify(self.SESSION_KEY, notify)
+
+        result_holder = {}
+
+        def _prompt():
+            result_holder["choice"] = mod.prompt_gateway_approval(
+                "touch /tmp/codex-approval-test",
+                "Codex requests exec in /tmp",
+                session_key=self.SESSION_KEY,
+                pattern_key="codex:test",
+                allow_permanent=False,
+                surface="codex_app_server",
+            )
+
+        t = threading.Thread(target=_prompt)
+        t.start()
+        for _ in range(50):
+            if mod._gateway_queues.get(self.SESSION_KEY):
+                break
+            time.sleep(0.02)
+
+        mod.resolve_gateway_approval(self.SESSION_KEY, "once")
+        t.join(timeout=5)
+
+        assert result_holder["choice"] == "once"
+        assert len(notified) == 1
+        assert notified[0]["command"] == "touch /tmp/codex-approval-test"
+        assert notified[0]["allow_permanent"] is False
+
+    def test_prompt_gateway_approval_maps_disallowed_always_to_session(self):
+        from tools import approval as mod
+
+        mod.register_gateway_notify(self.SESSION_KEY, lambda data: None)
+
+        result_holder = {}
+
+        def _prompt():
+            result_holder["choice"] = mod.prompt_gateway_approval(
+                "apply_patch",
+                "Codex requests to apply a patch",
+                session_key=self.SESSION_KEY,
+                pattern_key="codex:patch",
+                allow_permanent=False,
+            )
+
+        t = threading.Thread(target=_prompt)
+        t.start()
+        for _ in range(50):
+            if mod._gateway_queues.get(self.SESSION_KEY):
+                break
+            time.sleep(0.02)
+
+        mod.resolve_gateway_approval(self.SESSION_KEY, "always")
+        t.join(timeout=5)
+
+        assert result_holder["choice"] == "session"
+
     def test_timeout_emits_post_hook_with_timeout_outcome(self, monkeypatch):
         """Plugins must be able to distinguish timeout from explicit deny.
 

tools/approval.py

@@ -1467,6 +1467,74 @@ def _drop_entry() -> None:
     return {"resolved": resolved, "choice": choice}
 
 
+def prompt_gateway_approval(
+    command: str,
+    description: str,
+    *,
+    session_key: str | None = None,
+    pattern_key: str | None = None,
+    pattern_keys: list[str] | None = None,
+    allow_permanent: bool = True,
+    surface: str = "gateway",
+) -> str:
+    """Prompt for approval through the active gateway session queue.
+
+    This is the public gateway counterpart to ``prompt_dangerous_approval``:
+    callers that are not running inside the CLI terminal thread can still
+    block synchronously while the gateway sends the request to Telegram,
+    Slack, etc. and resolves it through ``resolve_gateway_approval()``.
+
+    Returns ``once``, ``session``, ``always``, or ``deny``. Missing gateway
+    session/callback, notify failure, timeout, and explicit denial all fail
+    closed as ``deny``.
+    """
+    resolved_session_key = session_key or get_current_session_key(default="")
+    if not resolved_session_key:
+        return "deny"
+
+    primary_key = pattern_key or f"gateway:{command}"
+    all_keys = list(pattern_keys or [primary_key])
+    approval_data = {
+        "command": command,
+        "pattern_key": primary_key,
+        "pattern_keys": all_keys,
+        "description": description,
+        "allow_permanent": bool(allow_permanent),
+    }
+
+    with _lock:
+        notify_cb = _gateway_notify_cbs.get(resolved_session_key)
+    if notify_cb is None:
+        submit_pending(resolved_session_key, approval_data)
+        return "deny"
+
+    decision = _await_gateway_decision(
+        resolved_session_key,
+        notify_cb,
+        approval_data,
+        surface=surface,
+    )
+    if decision.get("notify_failed"):
+        return "deny"
+
+    if not decision.get("resolved"):
+        return "deny"
+    choice = decision.get("choice") or "deny"
+    if choice == "always" and not allow_permanent:
+        choice = "session"
+    if choice == "deny":
+        return "deny"
+
+    for key in all_keys:
+        if choice == "session":
+            approve_session(resolved_session_key, key)
+        elif choice == "always":
+            approve_session(resolved_session_key, key)
+            approve_permanent(key)
+            save_permanent_allowlist(_permanent_approved)
+    return choice
+
+
 def check_all_command_guards(command: str, env_type: str,
                              approval_callback=None) -> dict:
     """Run all pre-exec security checks and return a single approval decision.