diff --git a/agentix/runtime/server/worker/process.py b/agentix/runtime/server/worker/process.py index 8f75b44..60a1c43 100644 --- a/agentix/runtime/server/worker/process.py +++ b/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-.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__": diff --git a/agentix/utils/log/__init__.py b/agentix/utils/log/__init__.py index b8b2718..10636e3 100644 --- a/agentix/utils/log/__init__.py +++ b/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-.log` (default `/tmp/agentix`) +inside the sandbox, so output survives a lost connection for +post-mortem reads. ## Delivery contract diff --git a/tests/_worker_target.py b/tests/_worker_target.py index 43a4ecb..3c1ba1f 100644 --- a/tests/_worker_target.py +++ b/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" diff --git a/tests/test_sio_namespace.py b/tests/test_sio_namespace.py index cbdb048..c62493c 100644 --- a/tests/test_sio_namespace.py +++ b/tests/test_sio_namespace.py @@ -20,7 +20,12 @@ emit_log_with_extra, fire_namespace_event, ) -from tests._worker_target import print_stdout +from tests._worker_target import ( + log_one_record, + print_stderr, + print_stdout, + spawn_stderr_writing_child, +) class _EchoHost(AsyncClientNamespace): @@ -229,6 +234,133 @@ def emit(self, record: logging.LogRecord) -> None: target_logger.removeHandler(handler) +@pytest.mark.asyncio +async def test_remote_stderr_arrives_on_host(live_server): + """fd 2 is captured like fd 1: a direct `sys.stderr` print inside the + remote fn replays on the host under `agentix.sandbox.stderr` (#138).""" + base_url = await live_server() + + captured: list[logging.LogRecord] = [] + + class _Capture(logging.Handler): + def emit(self, record: logging.LogRecord) -> None: + if record.name == "agentix.sandbox.stderr": + captured.append(record) + + target_logger = logging.getLogger("agentix.sandbox.stderr") + target_logger.setLevel(logging.INFO) + handler = _Capture() + target_logger.addHandler(handler) + try: + async with RuntimeClient(base_url) as c: + result = await c.remote(print_stderr, "hello from stderr") + assert result == "printed-stderr" + record = await _await_record(captured, "hello from stderr") + assert record is not None + assert getattr(record, "agentix_stream", None) == "stderr" + context = getattr(record, LOG_CONTEXT_ATTR, "") + assert context.startswith("sandbox-") + assert "-worker-" in context + finally: + target_logger.removeHandler(handler) + + +@pytest.mark.asyncio +async def test_child_process_stderr_arrives_on_host(live_server): + """Child processes inherit fd 2 — their stderr is exactly the output + stdlib logging cannot see, and it must still reach `/log` (#138).""" + base_url = await live_server() + + captured: list[logging.LogRecord] = [] + + class _Capture(logging.Handler): + def emit(self, record: logging.LogRecord) -> None: + if record.name == "agentix.sandbox.stderr": + captured.append(record) + + target_logger = logging.getLogger("agentix.sandbox.stderr") + target_logger.setLevel(logging.INFO) + handler = _Capture() + target_logger.addHandler(handler) + try: + async with RuntimeClient(base_url) as c: + result = await c.remote(spawn_stderr_writing_child, "child stderr line") + assert result == "spawned-stderr" + record = await _await_record(captured, "child stderr line") + assert record is not None + finally: + target_logger.removeHandler(handler) + + +@pytest.mark.asyncio +async def test_stdlib_records_are_not_recaptured_as_stderr(live_server): + """A stdlib record reaches the host exactly once — structured, via the + bridge. The console handler writes to the REAL stderr, so the record + must NOT come back a second time as a captured `agentix.sandbox.stderr` + line (#138 keeps the structured bridge; capture is additive).""" + base_url = await live_server() + + structured: list[logging.LogRecord] = [] + raw_stderr: list[logging.LogRecord] = [] + + class _Structured(logging.Handler): + def emit(self, record: logging.LogRecord) -> None: + if record.name == "tests.worker.dedup": + structured.append(record) + + class _Raw(logging.Handler): + def emit(self, record: logging.LogRecord) -> None: + if record.name == "agentix.sandbox.stderr": + raw_stderr.append(record) + + probe = "dedup probe 4242" + structured_logger = logging.getLogger("tests.worker.dedup") + structured_logger.setLevel(logging.INFO) + raw_logger = logging.getLogger("agentix.sandbox.stderr") + raw_logger.setLevel(logging.INFO) + s_handler, r_handler = _Structured(), _Raw() + structured_logger.addHandler(s_handler) + raw_logger.addHandler(r_handler) + try: + async with RuntimeClient(base_url) as c: + assert await c.remote(log_one_record, probe) == "logged" + assert await _await_record(structured, probe) is not None + # Grace period: a duplicate raw line would be the FORMATTED + # console line containing the probe text. + await asyncio.sleep(0.5) + assert not [r for r in raw_stderr if probe in r.getMessage()] + finally: + structured_logger.removeHandler(s_handler) + raw_logger.removeHandler(r_handler) + + +@pytest.mark.asyncio +async def test_sandbox_log_file_captures_records_and_stdio(live_server, tmp_path, monkeypatch): + """#139: the worker keeps a durable on-disk copy at + $AGENTIX_LOG_DIR/sandbox-.log — stdlib records AND captured + stdout/stderr lines — independent of `/log` stream delivery. The name + is per-worker so machine-shared dirs never race rotations.""" + monkeypatch.setenv("AGENTIX_LOG_DIR", str(tmp_path)) + base_url = await live_server() + + async with RuntimeClient(base_url) as c: + assert await c.remote(print_stdout, "file probe stdout") == "printed" + assert await c.remote(print_stderr, "file probe stderr") == "printed-stderr" + assert await c.remote(log_one_record, "file probe record") == "logged" + + deadline = asyncio.get_event_loop().time() + 3.0 + text = "" + while asyncio.get_event_loop().time() < deadline: + files = sorted(tmp_path.glob("sandbox-*.log")) + text = "".join(f.read_text(encoding="utf-8") for f in files) + if all(p in text for p in ("file probe stdout", "file probe stderr", "file probe record")): + break + await asyncio.sleep(0.05) + assert "file probe stdout" in text + assert "file probe stderr" in text + assert "file probe record" in text + + async def _await_record( captured: list[logging.LogRecord], message: str, @@ -284,9 +416,7 @@ def emit(self, record: logging.LogRecord) -> None: messages = [r.getMessage() for r in captured if r.getMessage().startswith("burst-")] expected = [f"burst-{i:03d}" for i in range(burst_count)] - assert messages == expected, ( - f"log stream lost or reordered events: got {len(messages)} of {burst_count}" - ) + assert messages == expected, f"log stream lost or reordered events: got {len(messages)} of {burst_count}" @pytest.mark.asyncio