| 1 | """Structured runtime logging for diagnosing agent behavior.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import json |
| 6 | import time |
| 7 | from dataclasses import dataclass, field |
| 8 | from pathlib import Path |
| 9 | from typing import Any |
| 10 | |
| 11 | _LOG_PATH = Path("/tmp/loader_runtime.log") |
| 12 | |
| 13 | |
| 14 | @dataclass(slots=True) |
| 15 | class LogEntry: |
| 16 | """One structured log entry.""" |
| 17 | |
| 18 | ts: float |
| 19 | event: str |
| 20 | data: dict[str, Any] = field(default_factory=dict) |
| 21 | |
| 22 | def to_line(self) -> str: |
| 23 | compact = {k: v for k, v in self.data.items() if v is not None} |
| 24 | return json.dumps({"t": round(self.ts, 2), "e": self.event, **compact}) |
| 25 | |
| 26 | |
| 27 | class RuntimeLogger: |
| 28 | """Append-only structured logger for one session.""" |
| 29 | |
| 30 | def __init__(self, path: Path | str | None = None) -> None: |
| 31 | self._path = Path(path) if path else _LOG_PATH |
| 32 | self._start = time.monotonic() |
| 33 | self._path.write_text("") # truncate on session start |
| 34 | |
| 35 | def _elapsed(self) -> float: |
| 36 | return time.monotonic() - self._start |
| 37 | |
| 38 | def log(self, event: str, **data: Any) -> None: |
| 39 | entry = LogEntry(ts=self._elapsed(), event=event, data=data) |
| 40 | try: |
| 41 | with self._path.open("a") as f: |
| 42 | f.write(entry.to_line() + "\n") |
| 43 | except Exception: |
| 44 | pass |
| 45 | |
| 46 | # ── convenience methods ────────────────────────────────────────── |
| 47 | |
| 48 | def turn_start(self, iteration: int, message_count: int, task: str) -> None: |
| 49 | self.log( |
| 50 | "turn.start", |
| 51 | iteration=iteration, |
| 52 | messages=message_count, |
| 53 | task=task[:120], |
| 54 | ) |
| 55 | |
| 56 | def turn_response( |
| 57 | self, |
| 58 | iteration: int, |
| 59 | content_len: int, |
| 60 | tool_call_count: int, |
| 61 | tool_names: list[str], |
| 62 | usage: dict[str, int] | None = None, |
| 63 | ) -> None: |
| 64 | self.log( |
| 65 | "turn.response", |
| 66 | iteration=iteration, |
| 67 | content_len=content_len, |
| 68 | tool_calls=tool_call_count, |
| 69 | tools=tool_names or None, |
| 70 | usage=usage or None, |
| 71 | ) |
| 72 | |
| 73 | def turn_decision( |
| 74 | self, |
| 75 | iteration: int, |
| 76 | action: str, |
| 77 | continuation_count: int, |
| 78 | consecutive_errors: int, |
| 79 | reason: str | None = None, |
| 80 | ) -> None: |
| 81 | self.log( |
| 82 | "turn.decision", |
| 83 | iteration=iteration, |
| 84 | action=action, |
| 85 | continuations=continuation_count, |
| 86 | errors=consecutive_errors, |
| 87 | reason=reason, |
| 88 | ) |
| 89 | |
| 90 | def tool_exec( |
| 91 | self, |
| 92 | name: str, |
| 93 | state: str, |
| 94 | is_error: bool, |
| 95 | result_preview: str, |
| 96 | appended_to_session: bool, |
| 97 | ) -> None: |
| 98 | self.log( |
| 99 | "tool.exec", |
| 100 | name=name, |
| 101 | state=state, |
| 102 | error=is_error or None, |
| 103 | result=result_preview[:200], |
| 104 | in_session=appended_to_session, |
| 105 | ) |
| 106 | |
| 107 | def verification_gate(self, tool_name: str, should_continue: bool) -> None: |
| 108 | self.log( |
| 109 | "tool.verify_gate", |
| 110 | tool=tool_name, |
| 111 | continue_loop=should_continue, |
| 112 | ) |
| 113 | |
| 114 | def completion_check( |
| 115 | self, |
| 116 | stage: str, |
| 117 | outcome: str, |
| 118 | reason: str | None = None, |
| 119 | ) -> None: |
| 120 | self.log("completion", stage=stage, outcome=outcome, reason=reason) |
| 121 | |
| 122 | def session_context(self, message_count: int, roles: dict[str, int]) -> None: |
| 123 | self.log("session.context", messages=message_count, roles=roles) |
| 124 | |
| 125 | def loop_exit(self, iterations: int, reason_code: str, reason: str) -> None: |
| 126 | self.log( |
| 127 | "loop.exit", |
| 128 | iterations=iterations, |
| 129 | reason_code=reason_code, |
| 130 | reason=reason[:200], |
| 131 | ) |
| 132 | |
| 133 | |
| 134 | # Module-level singleton, initialized lazily by the runtime. |
| 135 | _logger: RuntimeLogger | None = None |
| 136 | |
| 137 | |
| 138 | def get_runtime_logger() -> RuntimeLogger: |
| 139 | """Return the active runtime logger, creating one if needed.""" |
| 140 | global _logger |
| 141 | if _logger is None: |
| 142 | _logger = RuntimeLogger() |
| 143 | return _logger |
| 144 | |
| 145 | |
| 146 | def reset_runtime_logger(path: Path | str | None = None) -> RuntimeLogger: |
| 147 | """Create a fresh logger (call once per session start).""" |
| 148 | global _logger |
| 149 | _logger = RuntimeLogger(path) |
| 150 | return _logger |