Class TraceStore

uncommon_route/traces.py:335–575

class TraceStore:
    def __init__(
        self,
        storage: TraceStorage | None = None,
        now_fn: Any = None,
        *,
        hot_days: int | None = None,
    ) -> None:
        if storage is None:
            target_dir = data_dir() / "traces"
            legacy = data_dir() / "traces.json"
            try:
                migrate_legacy_json(legacy, target_dir)
            except Exception:
                pass
            storage = FileTraceStorage(base_dir=target_dir)
        self._storage = storage
        self._now = now_fn or time.time
        if hot_days is None:
            try:
                hot_days = int(os.environ.get("UNCOMMON_ROUTE_TRACE_HOT_DAYS", "2"))
            except Exception:
                hot_days = 2
        self._hot_days = max(1, hot_days)
        self._records: list[RequestTrace] = []
        self._load()

    @property
    def count(self) -> int:
        return len(self._records)

    def record(self, trace: RequestTrace) -> None:
        trace.tier = _normalize_tier_label(trace.tier)
        trace.decision_tier = _normalize_tier_label(trace.decision_tier) if trace.decision_tier else ""
        trace.served_quality = _normalize_served_quality(trace.served_quality)
        trace.served_quality_target = _normalize_served_quality(trace.served_quality_target)
        trace.served_quality_floor = _normalize_served_quality(trace.served_quality_floor)
        trace.capability_lane = str(trace.capability_lane or "").strip().lower()
        trace.feedback_from_tier = _normalize_tier_label(trace.feedback_from_tier) if trace.feedback_from_tier else ""
        trace.feedback_to_tier = _normalize_tier_label(trace.feedback_to_tier) if trace.feedback_to_tier else ""

        # Persist full row (incl. cold fields) to disk.
        payload = _trace_payload(trace)
        try:
            self._storage.append({"type": "trace", **payload})
        except Exception:
            pass

        # Hot copy: drop cold fields before keeping in memory.
        for f in COLD_FIELDS:
            setattr(trace, f, _empty_for(f))
        self._records.append(trace)
        self._cleanup()

    def record_feedback(
        self,
        request_id: str,
        *,