Spaces:
Sleeping
Sleeping
File size: 3,123 Bytes
2e818da | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 | """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()
@property
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
|