Spaces:
Sleeping
Sleeping
| """Compact JSONL storage for completed operation observations. | |
| Per the signal-responsibility table, the local observation exists to "Feed | |
| Control Room and preserve compact experiment evidence: latest stage durations, | |
| counts, statuses, and trace ID". This is deliberately the *smallest* useful | |
| persistence: one JSON line per completed operation, append-only, no index, no | |
| rotation policy (a later Control Room task owns summaries/aggregation). | |
| The rows are already-sanitized ``OperationObservation`` values; this module | |
| does not re-sanitize -- redaction happens once, in the operation boundary, | |
| before an observation is handed here. | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import threading | |
| from pathlib import Path | |
| from typing import Any | |
| from app.observability.contracts import OperationObservation | |
| _FILENAME = "observations.jsonl" | |
| class LocalObservationStore: | |
| """Append-only JSONL sink for :class:`OperationObservation` rows. | |
| Construction ensures the target directory exists. ``record`` may raise | |
| (disk full, permissions) -- the operation boundary is responsible for | |
| suppressing and counting such failures so a telemetry-write problem never | |
| fails a product request. | |
| """ | |
| def __init__(self, root: Path, filename: str = _FILENAME) -> None: | |
| self._root = Path(root) | |
| self._root.mkdir(parents=True, exist_ok=True) | |
| self._path = self._root / filename | |
| self._lock = threading.Lock() | |
| def path(self) -> Path: | |
| return self._path | |
| def record(self, observation: OperationObservation) -> None: | |
| """Append one observation as a single JSON line.""" | |
| line = json.dumps(observation.model_dump(mode="json"), ensure_ascii=False) | |
| with self._lock: | |
| with self._path.open("a", encoding="utf-8") as handle: | |
| handle.write(line + "\n") | |
| def latest(self) -> dict[str, Any]: | |
| """Return the most recently recorded row as a dict. | |
| Raises ``ValueError`` if the store file exists but has no recorded | |
| rows yet (an empty or all-blank-lines file). Note this does not | |
| guard against a missing file at all -- opening a store whose file has | |
| never been created raises the underlying ``FileNotFoundError`` from | |
| ``Path.open`` instead, since construction only ensures the *directory* | |
| exists, not the file itself. | |
| """ | |
| last: str | None = None | |
| with self._path.open("r", encoding="utf-8") as handle: | |
| for line in handle: | |
| if line.strip(): | |
| last = line | |
| if last is None: | |
| raise ValueError("LocalObservationStore is empty") | |
| return json.loads(last) | |
| def all(self) -> list[dict[str, Any]]: | |
| """Return every recorded row (small-scale evidence reads only).""" | |
| if not self._path.exists(): | |
| return [] | |
| rows: list[dict[str, Any]] = [] | |
| with self._path.open("r", encoding="utf-8") as handle: | |
| for line in handle: | |
| if line.strip(): | |
| rows.append(json.loads(line)) | |
| return rows | |