fix(agent-gaia): end a turn parked on a confirmation when stdin closes

Closing the host while a permission prompt was on screen left the agent
process running forever, holding the model slot — only the TUI's two-second
kill grace ended it, so any host that closed stdin and waited politely got an
immortal child. Turning unattended approval on, or launching with it on, also
left no trace anywhere in a normal session: nobody could tell afterwards
whether a gated tool had been approved or had just run.

- stdin EOF now cancels the in-flight turn before queueing the run-loop
  sentinel, which sits behind the running turn and could never reach it. The
  turn still ends through its normal path, with its one terminal event.
- Permission-state changes and denied or dropped decisions log through a
  channel pinned above the user/dev level split, and only to the log file —
  stdout stays the wire.
- A dead stdout ends the run loop instead of raising a second time out of the
  handler that exists to keep the process alive.
- Smaller: the decision handoff is atomic, the stdin pump always queues its
  sentinel, a Lemonade client that fails to build says why instead of reading
  as a dead server, and the turn hands the console back when it ends.

Three "unit" tests were opening live sockets to Lemonade, and the wire-purity
test asserted json.dumps output was JSON rather than exercising the logging
lockdown that actually protects the wire. The stdin pump, the argv contract
the Go side pins, the post-terminal write clamp, and the error paths of
_terminal_error and _memory_dump_event had no coverage at all.

Author: Ovtcharov

hub/agents/gaia/python/gaia_agent/memory_dump.py

@@ -1,3 +1,5 @@
+# Copyright(C) 2025-2026 Advanced Micro Devices, Inc. All rights reserved.
+# SPDX-License-Identifier: MIT
 """Builds the read-only snapshot the TUI's ``/memory`` view renders.
 
 The agent process already holds an open ``MemoryStore`` (via ``MemoryMixin``),

hub/agents/gaia/python/gaia_agent/stdio.py

@@ -1,3 +1,5 @@
+# Copyright(C) 2025-2026 Advanced Micro Devices, Inc. All rights reserved.
+# SPDX-License-Identifier: MIT
 """Run the flagship agent over stdin/stdout as newline-delimited JSON.
 
 The collapsed transport: the TUI spawns this process once and keeps it, writing
@@ -60,6 +62,7 @@
 from __future__ import annotations
 
 import json
+import logging
 import os
 import queue
 import sys
@@ -76,6 +79,20 @@
 
 logger = get_logger(__name__)
 
+#: Level the permission audit trail is pinned at, independent of --dev.
+AUDIT_LEVEL = logging.INFO
+
+#: Logger carrying permission-state history: bypass toggles and every
+#: decision that was denied or dropped.
+#:
+#: It needs a channel of its own because user mode logs ERROR only and the
+#: control channel writes nothing to stdout by design (see ``apply_control``)
+#: — so without this, turning unattended approval ON leaves no record
+#: anywhere. ``_configure_logging`` pins it to the log FILE; it must never
+#: reach stdout, which is the wire.
+AUDIT_LOGGER_NAME = "gaia_agent.stdio.audit"
+audit = get_logger(AUDIT_LOGGER_NAME)
+
 AGENT_ID = "gaia"
 
 #: Key that marks a stdin line as a control message rather than a query.
@@ -123,6 +140,10 @@ def __init__(self, bypass: bool = False) -> None:
         self._bypass = bypass
         self._grants: set = set()
         self._handler: Any = None
+        if bypass:
+            # Starting unattended is the same security event as toggling it on
+            # mid-session, and it never went through set_bypass.
+            audit.warning("Bypass permissions ENABLED at launch")
 
     @property
     def bypass(self) -> bool:
@@ -139,7 +160,7 @@ def set_bypass(self, enabled: bool) -> None:
             self._bypass = enabled
             if self._handler is not None:
                 self._handler.auto_approve_gated_tools = enabled
-        logger.warning("Bypass permissions %s", "ENABLED" if enabled else "disabled")
+        audit.warning("Bypass permissions %s", "ENABLED" if enabled else "disabled")
 
     def attach(self, handler: Any) -> None:
         """Hand a turn's handler the session's accumulated permission state."""
@@ -159,17 +180,42 @@ def detach(self, handler: Any) -> None:
                 self._handler = None
 
     def resolve(self, decision: str, confirm_id: Optional[str]) -> None:
-        """Answer the confirmation the agent thread is parked on."""
+        """Answer the confirmation the agent thread is parked on.
+
+        The lock is held across the resolve: dropping it first lets the turn
+        thread detach and the next turn attach a different handler in between,
+        and a decision carrying no ``confirm_id`` would then be accepted by a
+        handler nobody is waiting on while the live prompt keeps waiting.
+        """
         with self._lock:
             handler = self._handler
-        if handler is None:
-            logger.warning("Dropped a '%s' tool decision: no turn is running", decision)
-            return
-        handler.resolve_tool_confirmation(
-            approved=decision in (DECISION_ALLOW, DECISION_ALWAYS),
-            always=decision == DECISION_ALWAYS,
-            confirm_id=confirm_id,
-        )
+            if handler is None:
+                audit.warning(
+                    "Dropped a '%s' tool decision: no turn is running", decision
+                )
+                return
+            handler.resolve_tool_confirmation(
+                approved=decision in (DECISION_ALLOW, DECISION_ALWAYS),
+                always=decision == DECISION_ALWAYS,
+                confirm_id=confirm_id,
+            )
+
+    def cancel_active(self) -> bool:
+        """Cancel the turn currently running, if any. True if one was cancelled.
+
+        stdin closing means the host is gone, but the sentinel that ends the run
+        loop sits BEHIND the running turn in the query queue — so a turn parked
+        on a confirmation nobody can answer would keep the process alive forever,
+        holding the model slot. Cancelling unblocks the wait, which lets the turn
+        finish through its normal path and emit its one terminal event.
+        """
+        with self._lock:
+            handler = self._handler
+            if handler is None:
+                return False
+            handler.cancelled.set()
+        audit.warning("stdin closed mid-turn — cancelled the in-flight turn")
+        return True
 
 
 def parse_control(line: str) -> Optional[Dict[str, Any]]:
@@ -221,7 +267,7 @@ def apply_control(message: Dict[str, Any], state: PermissionState) -> None:
         decision = str(message.get("decision") or DECISION_DENY)
         if decision not in (DECISION_ALLOW, DECISION_DENY, DECISION_ALWAYS):
             # Fail closed: an unreadable decision is not consent.
-            logger.warning("Unknown tool decision %r — denying", decision)
+            audit.warning("Unknown tool decision %r — denying", decision)
             decision = DECISION_DENY
         confirm_id = message.get("confirm_id")
         state.resolve(decision, str(confirm_id) if confirm_id else None)
@@ -303,8 +349,13 @@ def _lemonade_health(base_url: Optional[str]) -> Dict[str, Any]:
     """
     try:
         client = LemonadeClient(base_url=base_url, verbose=False)
-    except Exception:  # pylint: disable=broad-exception-caught
-        return {"lemonade_reachable": False}
+    except Exception as exc:  # pylint: disable=broad-exception-caught
+        # A malformed base_url reads to the user as "Lemonade isn't running",
+        # so name it rather than reporting a bare unreachable.
+        logger.warning(
+            "[lemonade] client construction failed for %r: %s", base_url, exc
+        )
+        return {"lemonade_base_url": base_url, "lemonade_reachable": False}
     state: Dict[str, Any] = {"lemonade_base_url": client.base_url}
     try:
         health = client.health_check() or {}
@@ -409,9 +460,7 @@ def _apply_switch(
     Snapshots everything first and restores it on ANY exception —
     ``rebuild_system_prompt()`` runs inside this same guarded block, so a bug
     in prompt composition rolls back the client swap too, instead of leaving
-    the session on a working new client with a half-composed prompt (the
-    original version of this function called rebuild AFTER mutating, which
-    could not roll back — caught in review).
+    the session on a working new client with a half-composed prompt.
     """
     snapshot = _snapshot_switch_state(agent)
     chat = agent.chat
@@ -599,23 +648,32 @@ def _pump_stdin(queries: "queue.Queue", state: PermissionState) -> None:
     itself, so while a turn ran nothing was reading — which is exactly when a
     confirmation decision needs to arrive. Control messages are handled here,
     inline, while the agent thread is still parked on the prompt.
+
+    The teardown in ``finally`` is the process's only exit signal, so it has to
+    run even if iterating stdin itself raises: without it ``main`` waits on a
+    queue nothing will ever fill again.
     """
-    for raw in sys.stdin:
-        line = raw.strip()
-        if not line:
-            continue
-        control = parse_control(line)
-        if control is None:
-            queries.put(parse_query(line))
-            continue
-        try:
-            apply_control(control, state)
-        except Exception:  # pylint: disable=broad-exception-caught
-            # A malformed control line must never take the pump down: losing
-            # this thread means every later confirmation hangs with nothing
-            # able to answer it.
-            logger.exception("control message failed: %s", line)
-    queries.put(None)  # stdin closed
+    try:
+        for raw in sys.stdin:
+            line = raw.strip()
+            if not line:
+                continue
+            control = parse_control(line)
+            if control is None:
+                queries.put(parse_query(line))
+                continue
+            try:
+                apply_control(control, state)
+            except Exception:  # pylint: disable=broad-exception-caught
+                # A malformed control line must never take the pump down:
+                # losing this thread means every later confirmation hangs with
+                # nothing able to answer it.
+                logger.exception("control message failed: %s", line)
+    finally:
+        # Cancel BEFORE the sentinel: the sentinel is queued behind the running
+        # turn, so a turn parked on a confirmation would never reach it.
+        state.cancel_active()
+        queries.put(None)  # stdin closed
 
 
 def _write(event: Dict[str, Any], out) -> None:
@@ -646,14 +704,9 @@ def _record_turn(agent: Any, query: str, answer: str) -> None:
 
     Without this the flagship is amnesiac over stdio. ``Agent`` composes each
     request as ``[system, *conversation_history, user]`` (see
-    ``_build_messages``), and nothing in the base class ever appends to
-    ``conversation_history`` — the HTTP surface fills it in per request
-    (``gaia/ui/agent_loop.py``), and this transport did not. So every TUI turn
-    reached the model as exactly two messages, system + the current question.
-
-    The user-visible cost was not subtle: asked "print issue 2975" one turn
-    after a triage of amd/gaia, the agent replied "I need to know which
-    repository it belongs to". It was not being told.
+    ``_build_messages``) and nothing in the base class ever appends to
+    ``conversation_history``, so a turn this transport does not record reaches
+    the model as system + the current question and nothing else.
 
     Only the question and the final answer are kept. Tool calls and their
     results belong to the turn that made them and the agent already threads
@@ -689,7 +742,6 @@ def log_path() -> "Path":
     against yours. Every line also carries its pid (see ``_configure_logging``)
     so the shared default stays attributable when no override is set.
     """
-    import os
     from pathlib import Path
 
     override = os.environ.get(LOG_PATH_ENV, "").strip()
@@ -710,9 +762,9 @@ def _configure_logging(real_stdout, *, dev: bool) -> "Path":
     User mode logs errors only — a healthy run should leave a boring file.
     ``--dev`` turns on DEBUG for the whole tree, because the questions a
     developer asks (which tool, how long, why that step) are answered by the
-    records user mode drops.
+    records user mode drops. The permission audit trail is deliberately outside
+    that split — see ``AUDIT_LOGGER_NAME``.
     """
-    import logging
     from pathlib import Path
 
     sys.stdout = sys.stderr
@@ -752,6 +804,21 @@ def _configure_logging(real_stdout, *, dev: bool) -> "Path":
             # process-wide, so every debug call builds its LogRecord just for
             # the handler to drop it.
             lg.setLevel(logging.NOTSET)
+
+    # Configured last, so the NOTSET sweep above cannot clear it. Its own
+    # handler at AUDIT_LEVEL is what keeps a bypass toggle on the record in
+    # user mode, where the shared handler drops everything below ERROR.
+    audit_handler = logging.FileHandler(path, encoding="utf-8")
+    audit_handler.setLevel(AUDIT_LEVEL)
+    audit_handler.setFormatter(
+        logging.Formatter(
+            "%(asctime)s | pid:%(process)d | AUDIT | %(name)s | %(message)s"
+        )
+    )
+    audit_log = logging.getLogger(AUDIT_LOGGER_NAME)
+    audit_log.handlers = [audit_handler]
+    audit_log.setLevel(AUDIT_LEVEL)
+    audit_log.propagate = False
     return Path(path)
 
 
@@ -832,6 +899,7 @@ def run_turn(
     from gaia.ui.sse_handler import SSEOutputHandler
 
     handler = SSEOutputHandler()
+    previous_console = getattr(agent, "console", None)
     agent.console = handler
     if state is not None:
         state.attach(handler)
@@ -936,6 +1004,10 @@ def _run() -> None:
         # is no longer listening, and the prompt after it would hang.
         if state is not None:
             state.detach(handler)
+        # Base-agent threads outlive their turn (``_call_tool_bounded`` leaves a
+        # timed-out worker running), so a handler left attached between turns
+        # accumulates events on a queue nobody drains.
+        agent.console = previous_console
 
 
 def dispatch_query(
@@ -1059,16 +1131,27 @@ def main(argv: Optional[list] = None) -> int:
             break
         try:
             dispatch_query(agent, query, out, dev=args.dev, state=state)
+        except BrokenPipeError:
+            # The wire is the parent. It being gone is not a turn to report.
+            logger.warning("stdout closed mid-turn — ending the run loop")
+            break
         except Exception as exc:  # never let one bad turn kill the process
             logger.exception("stdio turn crashed outside the run loop")
-            _write(_terminal_error(exc), out)
+            try:
+                _write(_terminal_error(exc), out)
+            except OSError:
+                logger.warning("stdout closed while reporting — ending the run loop")
+                break
 
     # stdin closed: the parent is done with us. The agent leaves non-daemon
     # threads behind (memory extraction, the filesystem watcher), so a plain
     # return would hang the interpreter at shutdown waiting on them — a one-shot
     # `run --query` sat for 400s until its caller killed it. Nothing here owns
     # unflushed state: events are flushed per line and the DBs commit per write.
-    out.flush()
+    try:
+        out.flush()
+    except OSError as exc:
+        logger.warning("stdout was already gone at exit: %s", exc)
     sys.stderr.flush()
     os._exit(0)