"""Local session telemetry: payload-free per-call trajectory records.
ADR R7's operational half: every console session leaves a local, per-call
trajectory record so the harness is measurable and a live failure can be
traced and promoted into a golden scenario. The records are deliberately
METADATA-ONLY — tool name, command key, confirmation route, error flag,
duration, and content HASHES of the arguments and result — never the payloads
themselves: a tool result carries the taxpayer's figures, and
`sensitive-financial-data-secure-storage-only` forbids persisting those
anywhere outside encrypted secure storage. A hash lets two records be compared
for identity (the flywheel's dedup needs that) without storing a single
figure; the full payloads exist only inside the eval harness's own in-memory
:class:`~agent.eval.LiveTrajectory` during a measurement run.
Records append as JSON lines to ``<aeat_local_storage_root>/telemetry/
<session_id>.jsonl``, following the same state-root derivation the diagnostic
log uses, so each workspace's telemetry stays isolated.
The directory is bounded, not unbounded: a :class:`SessionTelemetryWriter`
sweeps its directory once at construction (server start) via
:func:`prune_telemetry`, dropping trajectory files past an age or count bound
while ALWAYS preserving the newest N sessions. Retention only ever removes whole
per-session files — it never touches the payload-free posture of the rows that
remain, and telemetry is a rebuildable derived surface, so a pruned file is a
lost measurement sample, never a correctness dependency.
"""
from __future__ import annotations
import contextlib
import hashlib
import json
import time
from pathlib import Path
from pydantic import BaseModel, ConfigDict, Field
from ...core.config import load_settings
from ...core.external_constants import UTF_8_ENCODING as _UTF_8
_STRICT_FROZEN = ConfigDict(frozen=True, strict=True, validate_assignment=True, extra="forbid")
_TELEMETRY_DIRNAME = "telemetry"
_SECONDS_PER_DAY = 86_400
[docs]
def content_sha256(text: str) -> str:
"""The one-way content reference telemetry stores instead of a payload."""
return hashlib.sha256(text.encode(_UTF_8)).hexdigest()
[docs]
def telemetry_dir() -> Path:
"""The workspace-scoped telemetry directory under the local storage root."""
return load_settings().aeat_local_storage_root / _TELEMETRY_DIRNAME
[docs]
class TelemetryRetention(BaseModel):
"""The bounds the startup trajectory-file sweep enforces.
Attributes:
max_age_days: Sessions whose file is older than this are pruned.
max_sessions: The maximum number of session files kept; the oldest
beyond this count are pruned.
keep_newest: The newest N sessions are ALWAYS retained, whatever their
age or the count bound — the sweep never removes them.
"""
model_config = _STRICT_FROZEN
max_age_days: float = Field(gt=0, default=30.0)
max_sessions: int = Field(ge=1, default=200)
keep_newest: int = Field(ge=0, default=20)
[docs]
def prune_telemetry(
directory: Path,
*,
max_age_days: float,
max_sessions: int,
keep_newest: int,
now: float | None = None,
) -> tuple[Path, ...]:
"""Prune per-session trajectory files by age and count, never the newest N.
Session files are ranked newest-first by modification time (ties broken by
filename for determinism). The newest ``keep_newest`` are retained
unconditionally. Of the remainder, a file is removed when it falls beyond
the ``max_sessions`` count bound OR is older than ``max_age_days``.
Args:
directory: The telemetry directory to sweep. A missing directory is a
no-op (returns an empty tuple).
max_age_days: Age bound in days; files with an older mtime are pruned.
max_sessions: Count bound; files ranked at or beyond this position
(newest-first, zero-based) are pruned.
keep_newest: The number of newest sessions to retain unconditionally.
now: Reference epoch seconds for the age comparison; defaults to the
wall clock. Injected by tests for deterministic ages.
Returns:
The tuple of file paths removed, in the order they were pruned.
"""
if not directory.exists():
return ()
reference = time.time() if now is None else now
cutoff = reference - max_age_days * _SECONDS_PER_DAY
entries = [(path, path.stat().st_mtime) for path in directory.glob("*.jsonl")]
# Newest first; the filename tie-break keeps the ranking deterministic when
# two sessions share an mtime (common in a fast test that writes in a burst).
entries.sort(key=lambda item: (item[1], item[0].name), reverse=True)
removed: list[Path] = []
for position, (path, mtime) in enumerate(entries):
if position < keep_newest:
continue
if position >= max_sessions or mtime < cutoff:
path.unlink()
removed.append(path)
return tuple(removed)
[docs]
class SessionTelemetryWriter:
"""Appends one session's records to its JSONL file, creating lazily.
The writer is append-only for its own session's rows. At construction
(server start) it runs a single best-effort :func:`prune_telemetry` sweep
over its directory so a long-lived installation's telemetry stays bounded;
the sweep is best-effort because a locked or vanished peer file must never
abort a new session's telemetry. The read path is :func:`read_session_records`;
the flywheel and any operator inspection read the files back through it, and
a rebuildable derived surface must never become a correctness dependency.
"""
def __init__(
self,
*,
session_id: str,
directory: Path | None = None,
retention: TelemetryRetention | None = None,
) -> None:
self._session_id = session_id
self._directory = directory if directory is not None else telemetry_dir()
self._retention = retention if retention is not None else TelemetryRetention()
self._sequence = 0
# Best-effort retention: a locked/racing peer file on a shared host must
# not stop this session from recording, so a sweep error is suppressed
# and the next server start retries the prune.
with contextlib.suppress(OSError):
prune_telemetry(
self._directory,
max_age_days=self._retention.max_age_days,
max_sessions=self._retention.max_sessions,
keep_newest=self._retention.keep_newest,
)
@property
def session_id(self) -> str:
"""The session identity every record of this writer carries."""
return self._session_id
@property
def path(self) -> Path:
"""The JSONL file this session appends to."""
return self._directory / f"{self._session_id}.jsonl"
[docs]
def record(
self,
*,
tool_name: str,
command_key: str = "",
route: str = "",
is_error: bool = False,
duration_ms: int = 0,
arguments_text: str = "",
result_text: str = "",
) -> ToolCallTelemetryRecord:
"""Append one payload-free record and return it.
Returns:
A :class:`ToolCallTelemetryRecord`.
"""
row = ToolCallTelemetryRecord(
session_id=self._session_id,
sequence=self._sequence,
tool_name=tool_name,
command_key=command_key,
route=route,
is_error=is_error,
duration_ms=duration_ms,
arguments_sha256=content_sha256(arguments_text) if arguments_text else "",
result_sha256=content_sha256(result_text) if result_text else "",
)
self._sequence += 1
self._directory.mkdir(parents=True, exist_ok=True)
with self.path.open("a", encoding=_UTF_8) as sink:
sink.write(json.dumps(row.model_dump(mode="json"), ensure_ascii=False, sort_keys=True))
sink.write("\n")
return row
[docs]
def read_session_records(path: Path) -> tuple[ToolCallTelemetryRecord, ...]:
"""Load one session file back into typed records (a strict roundtrip surface).
Returns:
A :class:`ToolCallTelemetryRecord`.
"""
rows: list[ToolCallTelemetryRecord] = []
for line in path.read_text(encoding=_UTF_8).splitlines():
if line.strip():
rows.append(ToolCallTelemetryRecord.model_validate(json.loads(line)))
return tuple(rows)
__all__ = [
"SessionTelemetryWriter",
"TelemetryRetention",
"ToolCallTelemetryRecord",
"content_sha256",
"prune_telemetry",
"read_session_records",
"telemetry_dir",
]