log: capture sandbox stderr and keep a durable on-disk copy

agentix/runtime/server/worker/process.py

@@ -16,6 +16,7 @@
 import sys
 import time
 import traceback
+from logging.handlers import RotatingFileHandler
 from typing import Any
 
 from agentix import sio as _sio
@@ -27,7 +28,7 @@
 from agentix.runtime.shared.models import RemoteError, RemoteRequest
 from agentix.utils import log as _log
 from agentix.utils.log._bridge import emit_worker_record
-from agentix.utils.log._config import LOG_CONTEXT_ATTR, get_log_context
+from agentix.utils.log._config import DEFAULT_LOG_FORMAT, LOG_CONTEXT_ATTR, get_log_context
 from agentix.utils.trace._bridge import install_worker_bridge
 
 logger = logging.getLogger("agentix.runtime.server.worker.process")
@@ -53,6 +54,7 @@ def __init__(self) -> None:
         self._outbound_q: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
         self._drainer: asyncio.Task | None = None
         self._stdio_tasks: list[asyncio.Task] = []
+        self._outbound_write_failed = False
 
     async def run(self) -> None:
         loop = asyncio.get_running_loop()
@@ -64,18 +66,32 @@ async def run(self) -> None:
         # desyncing the protocol and hanging every later call.
         #
         # Move the framing onto private fds and point fd 0 at /dev/null, so
-        # inherited stdin is harmless. fd 1 becomes a user-output pipe:
-        # `print()` and child-process stdout are drained separately and
-        # forwarded through the `/log` side channel instead of corrupting
-        # the control frame stream.
+        # inherited stdin is harmless. fd 1 / fd 2 become user-output pipes:
+        # `print()`, child-process output, and C-extension writes are
+        # drained separately and forwarded through the `/log` side channel
+        # instead of corrupting the control frame stream (fd 1) or vanishing
+        # into the container log (fd 2).
         frame_in_fd = os.dup(0)
         frame_out_fd = os.dup(1)
+        # Save the real stderr before fd 2 becomes a capture pipe — stdlib
+        # logging's console output is repointed there (see
+        # `_redirect_internal_logging`), and `main()` restores fd 2 from it
+        # on the way out so interpreter-level output (crash tracebacks,
+        # finalization errors) lands in the container log instead of a
+        # reader-less pipe.
+        global _real_stderr_fd
+        real_stderr_fd = _real_stderr_fd = os.dup(2)
         stdout_read_fd, stdout_write_fd = os.pipe()
+        stderr_read_fd, stderr_write_fd = os.pipe()
         devnull = os.open(os.devnull, os.O_RDWR)
         os.dup2(devnull, 0)
         os.dup2(stdout_write_fd, 1)
+        os.dup2(stderr_write_fd, 2)
         os.close(stdout_write_fd)
+        os.close(stderr_write_fd)
         os.close(devnull)
+        _redirect_internal_logging(real_stderr_fd)
+        _attach_sandbox_log_file()
         _make_stdout_eager()
 
         reader = asyncio.StreamReader()
@@ -99,7 +115,8 @@ async def run(self) -> None:
         # extensions registered on top of agentix.sio.
         install_worker_bridge()
         _log.install_worker_bridge()
-        self._stdio_tasks.append(loop.create_task(self._drain_stdout(stdout_read_fd)))
+        self._stdio_tasks.append(loop.create_task(self._drain_stream(stdout_read_fd, "stdout")))
+        self._stdio_tasks.append(loop.create_task(self._drain_stream(stderr_read_fd, "stderr")))
         await self._send({"type": "ready"})
 
         while not self._shutdown.is_set():
@@ -122,7 +139,7 @@ async def run(self) -> None:
         if self._calls:
             await asyncio.gather(*self._calls.values(), return_exceptions=True)
         if self._stdio_tasks:
-            _close_stdout_pipe()
+            _close_stdio_pipes()
             _, pending = await asyncio.wait(self._stdio_tasks, timeout=1.0)
             for task in pending:
                 task.cancel()
@@ -144,7 +161,15 @@ async def _drain_outbound(self) -> None:
                 try:
                     await write_frame(self._writer, frame)
                 except Exception:
-                    logger.exception("outbound frame write failed")
+                    # Log the FIRST failure at exception level only: each
+                    # logged traceback becomes a /log frame on this same
+                    # queue, so per-failure logging self-sustains against a
+                    # broken pipe and floods the durable file until shutdown.
+                    if not self._outbound_write_failed:
+                        self._outbound_write_failed = True
+                        logger.exception("outbound frame write failed")
+                    else:
+                        logger.debug("outbound frame write failed", exc_info=True)
                     self._recover_failed_frame(frame)
                 finally:
                     self._outbound_q.task_done()
@@ -178,7 +203,12 @@ def _recover_failed_frame(self, frame: dict[str, Any]) -> None:
     async def _send(self, payload: dict[str, Any]) -> None:
         await self._outbound_q.put(payload)
 
-    async def _drain_stdout(self, fd: int) -> None:
+    async def _drain_stream(self, fd: int, stream: str) -> None:
+        # The drainer shares the event loop: a remote `async def` that BLOCKS
+        # the loop (subprocess.run, ...) while its child spews ≥64 KiB into
+        # this fd wedges both sides — the same standing constraint the fd 1
+        # capture has always had. Async callables must not block the loop;
+        # sync callables run in threads and are fine.
         loop = asyncio.get_running_loop()
         reader = asyncio.StreamReader()
         await loop.connect_read_pipe(
@@ -187,8 +217,8 @@ async def _drain_stdout(self, fd: int) -> None:
         )
         # Read fixed-size chunks and split into lines ourselves. `readline()`
         # raises on a line longer than the StreamReader limit (64 KiB); that
-        # error was swallowed and KILLED this loop, so fd 1 stopped draining
-        # and the next `print()` blocked on a full pipe — deadlocking the
+        # error was swallowed and KILLED this loop, so the fd stopped draining
+        # and the next write blocked on a full pipe — deadlocking the
         # in-flight call. Chunked reads can never overflow, so the pipe is
         # always drained regardless of line length.
         buf = bytearray()
@@ -200,20 +230,20 @@ async def _drain_stdout(self, fd: int) -> None:
                 buf.extend(chunk)
                 *lines, buf_rest = bytes(buf).split(b"\n")
                 for line in lines:
-                    _emit_stdio_line("stdout", line)
+                    _emit_stdio_line(stream, line)
                 buf = bytearray(buf_rest)
                 # A newline-less spew (e.g. a binary blob) must not grow `buf`
                 # without bound — flush it as a partial line.
                 if len(buf) >= 65536:
-                    _emit_stdio_line("stdout", bytes(buf))
+                    _emit_stdio_line(stream, bytes(buf))
                     buf.clear()
         except asyncio.CancelledError:
             pass
         except Exception:
-            logger.debug("stdout drain failed", exc_info=True)
+            logger.debug("%s drain failed", stream, exc_info=True)
         finally:
             if buf:
-                _emit_stdio_line("stdout", bytes(buf))
+                _emit_stdio_line(stream, bytes(buf))
 
     def _enqueue_frame(self, frame: dict[str, Any]) -> None:
         """Sync put for the agentix.sio bridge — must never block."""
@@ -302,6 +332,103 @@ async def _amain() -> None:
     await worker.run()
 
 
+def _redirect_internal_logging(real_stderr_fd: int) -> None:
+    """Point stdlib logging's console output at the REAL stderr.
+
+    fd 2 is now a capture pipe whose lines replay on the host under
+    `agentix.sandbox.stderr`. stdlib records already reach the host
+    STRUCTURED (with ack/replay) via the `/log` bridge, so a console handler
+    left on fd 2 would deliver every record twice — and a worker diagnostic
+    emitted while the outbound pipe is broken would self-amplify (write
+    fails → logged to stderr → captured → enqueued → fails → …). Raw fd-2
+    capture is for the writers stdlib logging cannot see: child-process
+    stderr, C extensions, direct `sys.stderr` prints."""
+    with contextlib.suppress(Exception):
+        real_stderr = os.fdopen(real_stderr_fd, "w", buffering=1)
+        for handler in logging.getLogger().handlers:
+            if isinstance(handler, logging.StreamHandler) and handler.stream is sys.stderr:
+                handler.setStream(real_stderr)
+
+
+class _SandboxLogFileHandler(RotatingFileHandler):
+    """Best-effort durable log: any write failure detaches the handler for
+    good. It must never report its own errors — stdlib's `handleError`
+    prints to `sys.stderr`, which is the capture pipe, so a failing write
+    per captured line would amplify into a loop."""
+
+    def handleError(self, record: logging.LogRecord) -> None:
+        _detach_sandbox_log_file(self)
+
+
+_sandbox_log_handler: logging.Handler | None = None
+_real_stderr_fd: int | None = None
+
+
+def _restore_real_stderr() -> None:
+    """Point fd 2 back at the real stderr saved in `run()`.
+
+    Called on the way out of `main()`: after the loop closes, nothing drains
+    the capture pipe, so interpreter output written to fd 2 — the crash
+    traceback of an exception escaping `asyncio.run`, 'Exception ignored'
+    finalization messages — would either block, break the pipe, or be
+    discarded with the buffer. Restoring fd 2 sends it to the container log,
+    as on master."""
+    if _real_stderr_fd is not None:
+        with contextlib.suppress(Exception):
+            os.dup2(_real_stderr_fd, 2)
+
+
+def _attach_sandbox_log_file() -> None:
+    """Durable in-sandbox log at `$AGENTIX_LOG_DIR/sandbox-<worker>.log` (#139).
+
+    The `/log` stream buffer is bounded — output that outlives a long
+    disconnect (or the host itself) is otherwise unrecoverable. Attached to
+    the root logger so stdlib records land in the file; captured
+    stdout/stderr lines are written via a direct `emit()` from
+    `_emit_stdio_line`. Size-bounded with one rotation — a post-mortem
+    artifact, not an archive. Set `AGENTIX_LOG_DIR=` (empty) to disable."""
+    global _sandbox_log_handler
+    log_dir = os.environ.get("AGENTIX_LOG_DIR", "/tmp/agentix")
+    if not log_dir:
+        return
+    # Per-worker filename: the default dir is machine-shared, and two
+    # processes rotating ONE file race in `doRollover` (the loser's rename
+    # fails → the handler detaches). Every spawn gets a fresh worker id, so
+    # respawns never share a file either.
+    worker_id = os.environ.get("AGENTIX_WORKER_ID", str(os.getpid()))
+    try:
+        os.makedirs(log_dir, exist_ok=True)
+        handler = _SandboxLogFileHandler(
+            os.path.join(log_dir, f"sandbox-{worker_id}.log"),
+            maxBytes=64 * 1024 * 1024,
+            backupCount=1,
+            encoding="utf-8",
+            delay=True,
+        )
+    except Exception:
+        return
+    handler.setFormatter(logging.Formatter(os.environ.get("AGENTIX_LOG_FORMAT", DEFAULT_LOG_FORMAT)))
+    _sandbox_log_handler = handler
+    logging.getLogger().addHandler(handler)
+
+
+def _detach_sandbox_log_file(handler: logging.Handler) -> None:
+    global _sandbox_log_handler
+    _sandbox_log_handler = None
+    with contextlib.suppress(Exception):
+        logging.getLogger().removeHandler(handler)
+    with contextlib.suppress(Exception):
+        handler.close()
+    # Announce the loss on `/log` so a truncated post-mortem file is
+    # distinguishable from a sandbox that went quiet. The wire path is safe
+    # here — it is the FILE we can no longer write, not the stream.
+    with contextlib.suppress(Exception):
+        _emit_stdio_line_wire(
+            "stderr",
+            "agentix: durable sandbox log detached after a write failure; later output is stream-only",
+        )
+
+
 def _make_stdout_eager() -> None:
     """Make regular `print()` visible without requiring `flush=True`."""
     with contextlib.suppress(Exception):
@@ -310,20 +437,45 @@ def _make_stdout_eager() -> None:
             reconfigure(line_buffering=True, write_through=True)
 
 
-def _close_stdout_pipe() -> None:
-    """Flush fd 1 and detach it from the capture pipe so the drainer reaches EOF."""
-    with contextlib.suppress(Exception):
-        sys.stdout.flush()
-    with contextlib.suppress(Exception):
-        devnull = os.open(os.devnull, os.O_WRONLY)
-        try:
-            os.dup2(devnull, 1)
-        finally:
-            os.close(devnull)
+def _close_stdio_pipes() -> None:
+    """Flush fd 1 / fd 2 and detach them from the capture pipes so the
+    drainers reach EOF."""
+    for stream, fd in ((sys.stdout, 1), (sys.stderr, 2)):
+        with contextlib.suppress(Exception):
+            stream.flush()
+        with contextlib.suppress(Exception):
+            devnull = os.open(os.devnull, os.O_WRONLY)
+            try:
+                os.dup2(devnull, fd)
+            finally:
+                os.close(devnull)
 
 
 def _emit_stdio_line(stream: str, raw: bytes) -> None:
     text = raw.decode("utf-8", "replace").rstrip("\r\n")
+    handler = _sandbox_log_handler
+    if handler is not None:
+        # handler.handle(), not a logger call: routing through a logger would
+        # multiply the line into the console/bridge handlers. handle() (unlike
+        # a bare emit()) takes the handler lock — sync remote fns run in
+        # threads, so their stdlib records reach this same handler locked, and
+        # an unlocked emit racing a rollover kills the file for good.
+        with contextlib.suppress(Exception):
+            handler.handle(
+                logging.makeLogRecord(
+                    {
+                        "name": f"agentix.sandbox.{stream}",
+                        "msg": text,
+                        "levelno": logging.INFO,
+                        "levelname": "INFO",
+                        LOG_CONTEXT_ATTR: get_log_context(),
+                    }
+                )
+            )
+    _emit_stdio_line_wire(stream, text)
+
+
+def _emit_stdio_line_wire(stream: str, text: str) -> None:
     emit_worker_record(
         {
             "name": f"agentix.sandbox.{stream}",
@@ -355,6 +507,8 @@ def main() -> None:
         asyncio.run(_amain())
     except KeyboardInterrupt:
         pass
+    finally:
+        _restore_real_stderr()
 
 
 if __name__ == "__main__":

agentix/utils/log/__init__.py

@@ -14,8 +14,14 @@
 the root logger that emits each `LogRecord` on the `/log` SIO
 namespace. The host's `RuntimeClient` auto-registers a consumer that
 forwards records into the host's own `logging` system, so they appear
-in host logs untouched. The worker runtime also captures stdout and sends
-each line through the same `/log` stream as `agentix.sandbox.stdout`.
+in host logs untouched. The worker runtime also captures raw stdout and
+stderr (fd 1 / fd 2 — `print()`, child-process output, C-extension writes)
+and sends each line through the same `/log` stream as
+`agentix.sandbox.stdout` / `agentix.sandbox.stderr`; every record and
+captured line is also appended to a size-bounded on-disk copy at
+`$AGENTIX_LOG_DIR/sandbox-<worker-id>.log` (default `/tmp/agentix`)
+inside the sandbox, so output survives a lost connection for
+post-mortem reads.
 
 ## Delivery contract
 

tests/_worker_target.py

@@ -90,3 +90,31 @@ def spawn_stdin_reading_child() -> int:
         timeout=10,
     )
     return proc.returncode
+
+
+def print_stderr(message: str) -> str:
+    import sys
+
+    print(message, file=sys.stderr)
+    return "printed-stderr"
+
+
+def spawn_stderr_writing_child(message: str) -> str:
+    """A child process inheriting fd 2 — its stderr must reach the host
+    `/log` stream even though stdlib logging never sees it."""
+    import subprocess
+    import sys
+
+    subprocess.run(
+        [sys.executable, "-c", f"import sys; print({message!r}, file=sys.stderr)"],
+        check=True,
+        timeout=10,
+    )
+    return "spawned-stderr"
+
+
+def log_one_record(message: str) -> str:
+    import logging
+
+    logging.getLogger("tests.worker.dedup").warning(message)
+    return "logged"