eval_agent/logging.py

import importlib.util
import json
import math
import os
import sysconfig
import time
from pathlib import Path
from typing import Any, Dict, Optional

# ---------------------------------------------------------------------------
# Stdlib logging compatibility
# ---------------------------------------------------------------------------
# This file is intentionally named logging.py.
# When scripts under eval_agent/ are executed directly, `import logging` may
# resolve to this file instead of Python stdlib logging.
# To avoid breaking stdlib users, load stdlib logging and re-export its API.
_stdlib_logging_path = os.path.join(sysconfig.get_path("stdlib"), "logging", "__init__.py")
_stdlib_spec = importlib.util.spec_from_file_location("_stdlib_logging", _stdlib_logging_path)
_stdlib_logging = importlib.util.module_from_spec(_stdlib_spec)
assert _stdlib_spec is not None and _stdlib_spec.loader is not None
_stdlib_spec.loader.exec_module(_stdlib_logging)

for _name, _value in vars(_stdlib_logging).items():
    if _name not in globals():
        globals()[_name] = _value


def _sanitize_for_json(value: Any) -> Any:
    if isinstance(value, float):
        return value if math.isfinite(value) else None
    if isinstance(value, dict):
        return {k: _sanitize_for_json(v) for k, v in value.items()}
    if isinstance(value, list):
        return [_sanitize_for_json(v) for v in value]
    if isinstance(value, tuple):
        return [_sanitize_for_json(v) for v in value]
    if hasattr(value, "item") and callable(getattr(value, "item")):
        try:
            return _sanitize_for_json(value.item())
        except Exception:
            return str(value)
    return value


def _normalize_experiment_root(results_dir: str) -> Path:
    root = Path(results_dir).resolve()
    if root.name == "results":
        return root.parent.parent
    if root.name.startswith("gen_"):
        return root.parent
    return root


def _ensure_memory_dir(results_dir: str) -> Path:
    memory_dir = _normalize_experiment_root(results_dir) / "eval_agent_memory"
    memory_dir.mkdir(parents=True, exist_ok=True)
    return memory_dir


def log_event(results_dir: str, event_type: str, payload: Optional[Dict[str, Any]] = None) -> None:
    """Append one structured event to eval_agent_memory/agent_behavior.jsonl."""
    if not results_dir:
        return
    try:
        memory_dir = _ensure_memory_dir(results_dir)
        event = {
            "timestamp": time.time(),
            "event_type": event_type,
            **(payload or {}),
        }
        event = _sanitize_for_json(event)
        with open(memory_dir / "agent_behavior.jsonl", "a", encoding="utf-8") as f:
            f.write(json.dumps(event, ensure_ascii=False) + "\n")
    except Exception:
        # Never fail caller because of logging.
        pass


def save_text_artifact(results_dir: str, relative_path: str, content: str) -> None:
    """Save text artifact under eval_agent_memory/<relative_path>."""
    if not results_dir:
        return
    try:
        memory_dir = _ensure_memory_dir(results_dir)
        artifact_path = memory_dir / relative_path
        artifact_path.parent.mkdir(parents=True, exist_ok=True)
        artifact_path.write_text(content, encoding="utf-8")
    except Exception:
        pass