Merge pull request #21 from Agentiix/fix-examples-logging

Author: FatPigeorz

agentix/log/__init__.py

@@ -19,7 +19,9 @@
 
 import logging
 
-__all__ = ["install_worker_bridge"]
+from agentix.log._config import configure_logging
+
+__all__ = ["configure_logging", "install_worker_bridge"]
 
 
 def install_worker_bridge(level: int = logging.NOTSET) -> logging.Handler:

agentix/log/_bridge.py

@@ -2,13 +2,13 @@
 
 from __future__ import annotations
 
-import asyncio
 import logging
 from typing import Any
 
 import socketio
 
 from agentix import sio as _sio
+from agentix.log._config import LOG_CONTEXT_ATTR
 
 NAMESPACE = "/log"
 
@@ -48,10 +48,7 @@ def emit(self, record: logging.LogRecord) -> None:
         try:
             payload = _record_payload(record)
             ns = _get_worker_namespace()
-            asyncio.get_running_loop().create_task(ns.emit("record", payload))
-        except RuntimeError:
-            # No running loop — drop the record (worker is between async ticks).
-            pass
+            _sio._emit_nowait(ns.namespace, "record", payload)
         except Exception:
             self.handleError(record)
 
@@ -86,6 +83,7 @@ def emit(self, record: logging.LogRecord) -> None:
         # unconditionally so a record produced on 3.12+ doesn't try to
         # smuggle `taskName` through `extra=` into a fresh record.
         "taskName",
+        LOG_CONTEXT_ATTR,
     }
 )
 
@@ -105,6 +103,7 @@ def _record_payload(record: logging.LogRecord) -> dict[str, Any]:
         "exc_text": record.exc_text
         or (logging.Formatter().formatException(record.exc_info) if record.exc_info else None),
         "stack_info": record.stack_info,
+        LOG_CONTEXT_ATTR: getattr(record, LOG_CONTEXT_ATTR, None),
         "extras": extras or None,
     }
 
@@ -166,6 +165,8 @@ def _replay_record(payload: dict[str, Any]) -> None:
         record.exc_text = str(payload["exc_text"])
     if payload.get("stack_info"):
         record.stack_info = str(payload["stack_info"])
+    if payload.get(LOG_CONTEXT_ATTR):
+        setattr(record, LOG_CONTEXT_ATTR, str(payload[LOG_CONTEXT_ATTR]))
     logger.handle(record)
 
 

agentix/log/_config.py

@@ -0,0 +1,100 @@
+"""Small env-driven logging setup shared by host, runtime, and worker."""
+
+from __future__ import annotations
+
+import logging
+import os
+import socket
+import sys
+from typing import TextIO
+
+LOG_CONTEXT_ATTR = "agentix_context"
+DEFAULT_LOG_FORMAT = f"%(asctime)s [%({LOG_CONTEXT_ATTR})s] [%(name)s] %(message)s"
+
+_context = "host"
+_factory_installed = False
+_previous_factory = logging.getLogRecordFactory()
+
+
+class _SafeFormatMap(dict[str, str]):
+    def __missing__(self, key: str) -> str:
+        return "{" + key + "}"
+
+
+def configure_logging(
+    *,
+    default_context: str = "host",
+    stream: TextIO | None = None,
+    force: bool = False,
+) -> None:
+    """Configure stdlib logging from Agentix env vars.
+
+    Env vars:
+      AGENTIX_LOG_LEVEL: logging level name, default INFO.
+      AGENTIX_LOG_FORMAT: stdlib %-style format string.
+      AGENTIX_LOG_CONTEXT: context template for this process.
+
+    Context templates support `{uname}`, `{hostname}`, `{pid}`, and
+    `{id}`. The worker spawner sets `AGENTIX_WORKER_ID`, which backs
+    `{id}` inside worker processes.
+    """
+    set_log_context(os.environ.get("AGENTIX_LOG_CONTEXT", default_context))
+    level_name = os.environ.get("AGENTIX_LOG_LEVEL", "INFO").upper()
+    level = getattr(logging, level_name, logging.INFO)
+    log_format = os.environ.get("AGENTIX_LOG_FORMAT", DEFAULT_LOG_FORMAT)
+    logging.basicConfig(
+        level=level,
+        stream=stream or sys.stderr,
+        format=log_format,
+        force=force,
+    )
+
+
+def set_log_context(template: str) -> None:
+    global _context
+    _context = _expand_context(template)
+    _install_record_factory()
+
+
+def get_log_context() -> str:
+    return _context
+
+
+def _expand_context(template: str) -> str:
+    hostname = socket.gethostname()
+    values = _SafeFormatMap(
+        {
+            "hostname": hostname,
+            "uname": hostname,
+            "pid": str(os.getpid()),
+            "id": os.environ.get("AGENTIX_WORKER_ID", str(os.getpid())),
+        }
+    )
+    try:
+        return template.format_map(values)
+    except ValueError:
+        return template
+
+
+def _install_record_factory() -> None:
+    global _factory_installed
+    if _factory_installed:
+        return
+
+    def record_factory(*args, **kwargs):
+        record = _previous_factory(*args, **kwargs)
+        if not hasattr(record, LOG_CONTEXT_ATTR):
+            setattr(record, LOG_CONTEXT_ATTR, _context)
+        return record
+
+    logging.setLogRecordFactory(record_factory)
+    _factory_installed = True
+
+
+__all__ = [
+    "DEFAULT_LOG_FORMAT",
+    "LOG_CONTEXT_ATTR",
+    "configure_logging",
+    "get_log_context",
+    "set_log_context",
+]

agentix/runtime/server/app.py

@@ -23,12 +23,13 @@
 from fastapi import FastAPI
 
 from agentix import __version__
+from agentix.log import configure_logging
 from agentix.runtime.server.sio import make_sio
 from agentix.runtime.server.worker import RuntimeWorkerClient
 from agentix.runtime.shared.models import HealthResponse
 
+configure_logging(default_context="sandbox-{uname}")
 logger = logging.getLogger("agentix.runtime")
-logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(name)s] %(message)s")
 
 
 @asynccontextmanager

agentix/runtime/server/sio.py

@@ -190,9 +190,7 @@ async def on_cancel(sid: str, data: Any) -> None:
     # `sio_emit`  — worker wants to broadcast an event on a namespace;
     #               we pack the payload and call sio.emit there.
 
-    _broadcast_tasks: set[asyncio.Task] = set()
-
-    def _on_worker_sio_frame(frame: dict[str, Any]) -> None:
+    async def _on_worker_sio_frame(frame: dict[str, Any]) -> None:
         kind = frame.get("type")
         namespace = frame.get("namespace")
         if not isinstance(namespace, str) or not namespace.startswith("/"):
@@ -201,11 +199,7 @@ def _on_worker_sio_frame(frame: dict[str, Any]) -> None:
             event = frame.get("event")
             if not isinstance(event, str):
                 return
-            task = asyncio.create_task(
-                sio.emit(event, pack(frame.get("data")), namespace=namespace),
-            )
-            _broadcast_tasks.add(task)
-            task.add_done_callback(_broadcast_tasks.discard)
+            await sio.emit(event, pack(frame.get("data")), namespace=namespace)
         elif kind == "sio_open":
             if namespace in opened_namespaces or namespace == "/":
                 return

agentix/runtime/server/worker/client.py

@@ -10,6 +10,7 @@
 
 import asyncio
 import contextlib
+import inspect
 import logging
 import os
 import sys
@@ -128,7 +129,7 @@ async def send_inbound(self, namespace: str, event: str, data: Any) -> None: ...
     async def shutdown(self) -> None: ...
 
 
-SioFrameHandler = Callable[[dict[str, Any]], None]
+SioFrameHandler = Callable[[dict[str, Any]], Any]
 """Called from the worker read loop for each `sio_emit` or
 `sio_subscribe` frame the worker produces. The SIO server layer
 installs one; the in-process backend has no transport hop."""
@@ -173,6 +174,7 @@ def __init__(
     ) -> None:
         self._python = python
         self._runtime_bin_dir = runtime_bin_dir
+        self._worker_id = _new_id()[:8]
 
         self._proc: asyncio.subprocess.Process | None = None
         self._send_lock = asyncio.Lock()
@@ -188,6 +190,13 @@ def __init__(
     async def start(self) -> None:
         env = _clean_worker_env(self._runtime_bin_dir)
         env["AGENTIX_WORKER_IMPORT_ROOT"] = str(_WORKER_IMPORT_ROOT)
+        env["AGENTIX_WORKER_ID"] = self._worker_id
+        env["AGENTIX_LOG_CONTEXT"] = env.get(
+            "AGENTIX_WORKER_LOG_CONTEXT",
+            "sandbox-{uname}-worker-{id}",
+        )
+        if worker_log_format := env.get("AGENTIX_WORKER_LOG_FORMAT"):
+            env["AGENTIX_LOG_FORMAT"] = worker_log_format
         self._proc = await asyncio.create_subprocess_exec(
             self._python,
             "-c",
@@ -235,7 +244,7 @@ async def _read_loop(self) -> None:
                 frame = await read_frame(self._proc.stdout)
                 if frame is None:
                     break
-                self._on_frame(frame)
+                await self._on_frame(frame)
         except Exception:
             logger.exception("runtime worker read loop crashed")
         finally:
@@ -246,7 +255,7 @@ async def _read_loop(self) -> None:
                     fut.set_exception(RuntimeError(err.message))
             self._pending.clear()
 
-    def _on_frame(self, frame: dict[str, Any]) -> None:
+    async def _on_frame(self, frame: dict[str, Any]) -> None:
         kind = frame.get("type")
         if kind == "ready":
             self._ready.set()
@@ -268,7 +277,9 @@ def _on_frame(self, frame: dict[str, Any]) -> None:
         elif kind in ("sio_emit", "sio_open"):
             if self._sio_handler is not None:
                 try:
-                    self._sio_handler(frame)
+                    result = self._sio_handler(frame)
+                    if inspect.isawaitable(result):
+                        await result
                 except Exception:
                     logger.debug("sio frame handler raised; dropping", exc_info=True)
         else:

agentix/runtime/server/worker/process.py

@@ -211,12 +211,9 @@ async def _amain() -> None:
 
 
 def main() -> None:
-    level_name = os.environ.get("AGENTIX_LOG_LEVEL", "INFO").upper()
-    level = getattr(logging, level_name, logging.INFO)
-    logging.basicConfig(
-        level=level,
+    _log.configure_logging(
+        default_context="sandbox-{uname}-worker-{id}",
         stream=sys.stderr,
-        format="%(asctime)s [%(name)s] %(message)s",
     )
     try:
         asyncio.run(_amain())

agentix/sio.py

@@ -154,14 +154,7 @@ def _auto_register(self) -> None:
 
     async def emit(self, event: str, data: Any = None) -> None:
         """Emit `event` on this namespace to all connected hosts."""
-        _bridge.send_frame(
-            {
-                "type": "sio_emit",
-                "namespace": self.namespace,
-                "event": event,
-                "data": data,
-            }
-        )
+        _emit_nowait(self.namespace, event, data)
 
     def on(self, event: str, handler: Handler) -> None:
         """Register an additional handler for `event`."""
@@ -291,6 +284,17 @@ def _is_installed() -> bool:
     return _bridge.is_installed()
 
 
+def _emit_nowait(namespace: str, event: str, data: Any = None) -> None:
+    _bridge.send_frame(
+        {
+            "type": "sio_emit",
+            "namespace": namespace,
+            "event": event,
+            "data": data,
+        }
+    )
+
+
 __all__ = [
     "RESERVED_NAMESPACES",
     "Namespace",

agentix/trace/_bridge.py

@@ -11,7 +11,6 @@
 
 from __future__ import annotations
 
-import asyncio
 import contextvars
 from typing import Any
 
@@ -76,7 +75,7 @@ def _emit(self, event: str, payload: dict[str, Any]) -> None:
         if not _sio._is_installed():
             return
         try:
-            asyncio.get_running_loop().create_task(self._ns.emit(event, payload))
+            _sio._emit_nowait(self._ns.namespace, event, payload)
         except RuntimeError:
             pass
 

examples/eval-cc-swe/runner.py

@@ -40,6 +40,7 @@
 
 from agentix import RuntimeClient
 from agentix.deployment.base import SandboxConfig, session
+from agentix.log import configure_logging
 
 WORKDIR = "/testbed"
 logger = logging.getLogger("eval_cc_swe.runner")
@@ -419,10 +420,7 @@ async def main(argv: list[str] | None = None) -> int:
         print("error: --concurrency must be >= 1", file=sys.stderr)
         return 2
 
-    logging.basicConfig(
-        level=logging.INFO,
-        format="%(asctime)s [%(name)s] %(message)s",
-    )
+    configure_logging(default_context="host")
 
     ds = _load_instances_dataset(args.dataset, split=args.split, dataset_file=args.dataset_file)
     instances = _selected_instances(