| """Structured step logging for deepagents-rl pipelines.
|
|
|
| Emits:
|
| - Console (stderr) with timestamps + step tags
|
| - Optional JSONL event file for full replay (`events.jsonl`)
|
|
|
| Env:
|
| DARL_LOG_LEVEL DEBUG|INFO|WARNING (default INFO)
|
| DARL_LOG_FILE path to text log (optional)
|
| DARL_EVENTS_JSONL path to JSONL events (optional; auto under artifacts if set via configure)
|
| """
|
| from __future__ import annotations
|
|
|
| import json
|
| import logging
|
| import os
|
| import sys
|
| import threading
|
| import time
|
| import traceback
|
| from contextlib import contextmanager
|
| from datetime import datetime, timezone
|
| from pathlib import Path
|
| from typing import Any, Iterator
|
|
|
| _LOCK = threading.Lock()
|
| _EVENTS_PATH: Path | None = None
|
| _CONFIGURED = False
|
|
|
| LOGGER_NAME = "deepagents_rl"
|
|
|
|
|
| def _utc() -> str:
|
| return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
| def get_logger(name: str | None = None) -> logging.Logger:
|
| return logging.getLogger(name or LOGGER_NAME)
|
|
|
|
|
| def configure_logging(
|
| *,
|
| level: str | None = None,
|
| log_file: str | Path | None = None,
|
| events_jsonl: str | Path | None = None,
|
| force: bool = False,
|
| ) -> logging.Logger:
|
| """Idempotent logging setup for CLI scripts and harness modules."""
|
| global _CONFIGURED, _EVENTS_PATH
|
| log = get_logger()
|
|
|
| lvl_name = (level or os.getenv("DARL_LOG_LEVEL") or "INFO").upper()
|
| lvl = getattr(logging, lvl_name, logging.INFO)
|
|
|
| if _CONFIGURED and not force:
|
| log.setLevel(lvl)
|
| if events_jsonl:
|
| _EVENTS_PATH = Path(events_jsonl)
|
| _EVENTS_PATH.parent.mkdir(parents=True, exist_ok=True)
|
| return log
|
|
|
| log.setLevel(lvl)
|
| log.handlers.clear()
|
| log.propagate = False
|
|
|
| fmt = logging.Formatter(
|
| fmt="%(asctime)s | %(levelname)-5s | %(name)s | %(message)s",
|
| datefmt="%H:%M:%S",
|
| )
|
| sh = logging.StreamHandler(sys.stderr)
|
| sh.setLevel(lvl)
|
| sh.setFormatter(fmt)
|
| log.addHandler(sh)
|
|
|
| file_path = log_file or os.getenv("DARL_LOG_FILE")
|
| if file_path:
|
| fp = Path(file_path)
|
| fp.parent.mkdir(parents=True, exist_ok=True)
|
| fh = logging.FileHandler(fp, encoding="utf-8")
|
| fh.setLevel(lvl)
|
| fh.setFormatter(fmt)
|
| log.addHandler(fh)
|
|
|
| ev = events_jsonl or os.getenv("DARL_EVENTS_JSONL")
|
| if ev:
|
| _EVENTS_PATH = Path(ev)
|
| _EVENTS_PATH.parent.mkdir(parents=True, exist_ok=True)
|
| else:
|
| _EVENTS_PATH = None
|
|
|
| _CONFIGURED = True
|
| log.debug("logging configured level=%s file=%s events=%s", lvl_name, file_path, _EVENTS_PATH)
|
| return log
|
|
|
|
|
| def _write_event(event: dict[str, Any]) -> None:
|
| if _EVENTS_PATH is None:
|
| return
|
| with _LOCK:
|
| with _EVENTS_PATH.open("a", encoding="utf-8") as f:
|
| f.write(json.dumps(event, default=str) + "\n")
|
|
|
|
|
| def log_event(
|
| event_type: str,
|
| message: str,
|
| *,
|
| level: int = logging.INFO,
|
| **fields: Any,
|
| ) -> None:
|
| """Log a structured event to console + JSONL."""
|
| log = get_logger()
|
| payload = {
|
| "ts": _utc(),
|
| "event": event_type,
|
| "message": message,
|
| **fields,
|
| }
|
|
|
| extras = {k: v for k, v in fields.items() if k not in ("traceback",)}
|
| extra_s = ""
|
| if extras:
|
| try:
|
| extra_s = " | " + json.dumps(extras, default=str)[:1500]
|
| except Exception:
|
| extra_s = f" | {extras!r}"[:1500]
|
| log.log(level, f"[{event_type}] {message}{extra_s}")
|
| _write_event(payload)
|
|
|
|
|
| def step(name: str, message: str = "", **fields: Any) -> None:
|
| log_event("step", message or name, step=name, **fields)
|
|
|
|
|
| def call(name: str, message: str = "", **fields: Any) -> None:
|
| log_event("call", message or name, call=name, **fields)
|
|
|
|
|
| def result(name: str, message: str = "", **fields: Any) -> None:
|
| log_event("result", message or name, call=name, **fields)
|
|
|
|
|
| def error(name: str, message: str, **fields: Any) -> None:
|
| log_event("error", message, level=logging.ERROR, call=name, **fields)
|
|
|
|
|
| @contextmanager
|
| def span(name: str, message: str = "", **fields: Any) -> Iterator[dict[str, Any]]:
|
| """Timed span: logs start + end with duration_ms."""
|
| t0 = time.perf_counter()
|
| ctx: dict[str, Any] = {"span": name, **fields}
|
| log_event("span_start", message or name, **ctx)
|
| try:
|
| yield ctx
|
| except Exception as e:
|
| dt = int((time.perf_counter() - t0) * 1000)
|
| error(
|
| name,
|
| f"{type(e).__name__}: {e}",
|
| duration_ms=dt,
|
| traceback=traceback.format_exc()[-3000:],
|
| **{k: v for k, v in ctx.items() if k != "span"},
|
| )
|
| raise
|
| else:
|
| dt = int((time.perf_counter() - t0) * 1000)
|
| log_event(
|
| "span_end",
|
| message or name,
|
| duration_ms=dt,
|
| ok=True,
|
| **ctx,
|
| )
|
|
|
|
|
| def truncate(s: Any, n: int = 800) -> str:
|
| text = s if isinstance(s, str) else json.dumps(s, default=str)
|
| if len(text) <= n:
|
| return text
|
| return text[:n] + f"...(+{len(text)-n} chars)"
|
|
|
