docs: research unified v1 LMCache observability for #141

cacheroute_observability/README.md

@@ -0,0 +1,48 @@
+# CacheRoute observability v1
+
+`cacheroute_observability` is the dependency-light Phase 1 contract and collection
+boundary for Issue #141. It imports only the Python standard library and Pydantic,
+so Client, Scheduler, Proxy, KDN, Gateway, and Instance code can adopt it later
+without loading the CacheRoute application graph.
+
+## Models and vocabulary
+
+Frozen, extra-forbidding contracts represent trace context, provenance, typed
+measurements, ordered stages, request traces, cache-operation traces, and links
+between one shared operation and multiple waiting requests. Stable enums never
+encode route names, implementation methods, Redis commands, or private LMCache
+events. `TraceStageState` describes lifecycle (`pending`, `running`, `completed`);
+`TraceStageOutcome` independently describes the terminal result.
+
+## Clocks and collector
+
+UTC wall-clock values support cross-process correlation. Durations come only from
+local `monotonic_ns()` readings. `ManualTraceClock` makes both sources deterministic
+in CPU-only tests. `TraceCollector` starts stages with both readings, calculates a
+local monotonic duration on finish, preserves append sequence (including repeated
+stage names), and exports a new immutable snapshot. Disabled and unsampled
+collectors allocate no stages. Optional instrumentation can use `safely()` so an
+observability failure cannot affect caller behavior.
+
+## Legacy projection and security
+
+`project_legacy_proxy_trace()` is a pure allow-list adapter. It classifies known
+Proxy fields, treats ambiguous and overwritten values as `legacy_projected`, uses
+truthful Proxy-boundary names, sanitizes known fallback codes, and omits unknown
+keys and raw error text. It never embeds the source trace or `kv_ack` mapping.
+Contracts reject extras, containers as generic scalar values, secret/physical-KV
+field names, raw content, paths, credentials, Redis keys, KV bytes, tensors, and
+pointers.
+
+## Phase 1 limitations
+
+This package provides contracts, a process-local collector, Legacy projection,
+and a CPU-only demonstration. It does **not** instrument production components,
+propagate context, change `cacheroute_meta`, perform Gateway/network I/O, attribute
+aggregate metrics to requests, or integrate with LMCache/vLLM. Those remain later
+Issue #141 phases after review.
+
+The temporary design record is
+[`doc/research/issue-141-unified-observability.md`](../doc/research/issue-141-unified-observability.md)
+and will be removed after Issue #141 is completed and stable material is moved to
+maintained documentation.

cacheroute_observability/__init__.py

@@ -0,0 +1,20 @@
+"""Dependency-light public exports for CacheRoute observability v1."""
+from .clock import ManualTraceClock, SystemTraceClock, TraceClock
+from .collector import TraceCollector
+from .enums import (
+    OperationWaiterState, TraceComponent, TraceStageName, TraceStageOutcome,
+    TraceStageState, TraceValueKind,
+)
+from .legacy_proxy_trace import project_legacy_proxy_trace
+from .models import (
+    CacheOperationTrace, OperationWaiterLink, RequestTrace, TraceContext,
+    TraceMeasurement, TraceProvenance, TraceStage,
+)
+
+__all__ = [
+    "CacheOperationTrace", "ManualTraceClock", "OperationWaiterLink",
+    "OperationWaiterState", "RequestTrace", "SystemTraceClock", "TraceClock",
+    "TraceCollector", "TraceComponent", "TraceContext", "TraceMeasurement",
+    "TraceProvenance", "TraceStage", "TraceStageName", "TraceStageOutcome",
+    "TraceStageState", "TraceValueKind", "project_legacy_proxy_trace",
+]

cacheroute_observability/clock.py

@@ -0,0 +1,49 @@
+"""Injectable clocks that keep correlation timestamps separate from durations."""
+from __future__ import annotations
+
+import time
+from datetime import datetime, timedelta, timezone
+from typing import Protocol, runtime_checkable
+
+
+@runtime_checkable
+class TraceClock(Protocol):
+    def utc_now(self) -> datetime: ...
+    def monotonic_ns(self) -> int: ...
+
+
+class SystemTraceClock:
+    def utc_now(self) -> datetime:
+        return datetime.now(timezone.utc)
+
+    def monotonic_ns(self) -> int:
+        return time.perf_counter_ns()
+
+
+class ManualTraceClock:
+    def __init__(self, utc_value: datetime, monotonic_value: int = 0):
+        if utc_value.tzinfo is None or utc_value.utcoffset() != timedelta(0):
+            raise ValueError("utc_value must be timezone-aware UTC")
+        if monotonic_value < 0:
+            raise ValueError("monotonic_value must be non-negative")
+        self._utc_value = utc_value
+        self._monotonic_value = monotonic_value
+
+    def utc_now(self) -> datetime:
+        return self._utc_value
+
+    def monotonic_ns(self) -> int:
+        return self._monotonic_value
+
+    def advance_ns(self, value: int) -> None:
+        if value < 0:
+            raise ValueError("monotonic advance must be non-negative")
+        self._monotonic_value += value
+
+    def advance_time(self, value: timedelta) -> None:
+        if value < timedelta(0):
+            raise ValueError("wall-clock advance must be non-negative")
+        self._utc_value += value
+
+
+__all__ = ["ManualTraceClock", "SystemTraceClock", "TraceClock"]

cacheroute_observability/collector.py

@@ -0,0 +1,205 @@
+"""Process-local append-only trace collection with monotonic durations."""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime
+from typing import Callable
+from uuid import uuid4
+
+from .clock import SystemTraceClock, TraceClock
+from .enums import TraceStageName, TraceStageOutcome, TraceStageState
+from .models import RequestTrace, TraceContext, TraceMeasurement, TraceProvenance, TraceStage
+
+
+@dataclass
+class _OpenStage:
+    stage_id: str
+    sequence: int
+    name: TraceStageName
+    provenance: TraceProvenance
+    started_at: datetime
+    started_monotonic_ns: int
+    parent_stage_id: str | None
+    operation_id: str | None
+    artifact_id: str | None
+    measurements: list[TraceMeasurement] = field(default_factory=list)
+
+
+class TraceCollector:
+    """Collect local stages; exported contracts never contain monotonic readings."""
+
+    def __init__(
+        self,
+        context: TraceContext,
+        *,
+        clock: TraceClock | None = None,
+        enabled: bool = True,
+        id_factory: Callable[[], str] | None = None,
+    ):
+        self.context = context
+        self.clock = clock or SystemTraceClock()
+        self.enabled = bool(enabled)
+        self._id_factory = id_factory or (lambda: f"stage_{uuid4().hex}")
+        self._next_sequence = 0
+        self._open: dict[str, _OpenStage] = {}
+        self._timeline: list[str] = []
+        self._completed: dict[str, TraceStage] = {}
+
+    @classmethod
+    def create(
+        cls,
+        *,
+        request_id: str,
+        correlation_id: str,
+        legacy_request_id: int | None = None,
+        sampled: bool = True,
+        clock: TraceClock | None = None,
+        enabled: bool = True,
+        id_factory: Callable[[], str] | None = None,
+    ) -> "TraceCollector":
+        """Create a context at the injected clock and return its collector."""
+        selected_clock = clock or SystemTraceClock()
+        context = TraceContext(
+            trace_id=f"trace_{uuid4().hex}",
+            request_id=request_id,
+            correlation_id=correlation_id,
+            legacy_request_id=legacy_request_id,
+            sampled=sampled,
+            created_at=selected_clock.utc_now(),
+        )
+        return cls(context, clock=selected_clock, enabled=enabled, id_factory=id_factory)
+
+    @property
+    def collecting(self) -> bool:
+        return self.enabled and self.context.sampled
+
+    def start_stage(
+        self,
+        name: TraceStageName,
+        provenance: TraceProvenance,
+        *,
+        parent_stage_id: str | None = None,
+        operation_id: str | None = None,
+        artifact_id: str | None = None,
+    ) -> str | None:
+        if not self.collecting:
+            return None
+        stage_id = self._id_factory()
+        if stage_id in self._open or stage_id in self._completed:
+            raise ValueError("stage_id must be unique within a collector")
+        stage = _OpenStage(
+            stage_id=stage_id,
+            sequence=self._next_sequence,
+            name=TraceStageName(name),
+            provenance=provenance,
+            started_at=self.clock.utc_now(),
+            started_monotonic_ns=self.clock.monotonic_ns(),
+            parent_stage_id=parent_stage_id,
+            operation_id=operation_id,
+            artifact_id=artifact_id,
+        )
+        self._next_sequence += 1
+        self._open[stage_id] = stage
+        self._timeline.append(stage_id)
+        return stage_id
+
+    def append_measurement(self, stage_id: str | None, measurement: TraceMeasurement) -> bool:
+        if not self.collecting or stage_id is None:
+            return False
+        stage = self._open.get(stage_id)
+        if stage is None:
+            if stage_id in self._completed:
+                raise ValueError("cannot append to a completed stage")
+            raise KeyError("unknown stage_id")
+        stage.measurements.append(measurement)
+        return True
+
+    def finish_stage(
+        self,
+        stage_id: str | None,
+        outcome: TraceStageOutcome,
+        *,
+        outcome_code: str | None = None,
+        error_code: str | None = None,
+        error_message: str | None = None,
+        retryable: bool = False,
+        fallback_eligible: bool = False,
+        fallback_stage_id: str | None = None,
+        partial_reason: str | None = None,
+    ) -> TraceStage | None:
+        if not self.collecting or stage_id is None:
+            return None
+        if stage_id in self._completed:
+            raise ValueError("stage is already completed")
+        stage = self._open.get(stage_id)
+        if stage is None:
+            raise KeyError("unknown stage_id")
+        ended_at = self.clock.utc_now()
+        elapsed = self.clock.monotonic_ns() - stage.started_monotonic_ns
+        if elapsed < 0:
+            raise ValueError("monotonic clock moved backwards")
+        completed = TraceStage(
+            stage_id=stage.stage_id,
+            sequence=stage.sequence,
+            name=stage.name,
+            state=TraceStageState.COMPLETED,
+            outcome=outcome,
+            provenance=stage.provenance,
+            measurements=tuple(stage.measurements),
+            started_at=stage.started_at,
+            ended_at=ended_at,
+            duration_ns=elapsed,
+            parent_stage_id=stage.parent_stage_id,
+            operation_id=stage.operation_id,
+            artifact_id=stage.artifact_id,
+            outcome_code=outcome_code,
+            error_code=error_code,
+            error_message=error_message,
+            retryable=retryable,
+            fallback_eligible=fallback_eligible,
+            fallback_stage_id=fallback_stage_id,
+            partial_reason=partial_reason,
+        )
+        del self._open[stage_id]
+        self._completed[stage_id] = completed
+        return completed
+
+    def export(self, *, complete: bool = False, error_code: str | None = None) -> RequestTrace:
+        stages: list[TraceStage] = []
+        for stage_id in self._timeline:
+            completed = self._completed.get(stage_id)
+            if completed is not None:
+                stages.append(completed)
+                continue
+            opened = self._open[stage_id]
+            stages.append(TraceStage(
+                stage_id=opened.stage_id,
+                sequence=opened.sequence,
+                name=opened.name,
+                state=TraceStageState.RUNNING,
+                provenance=opened.provenance,
+                measurements=tuple(opened.measurements),
+                started_at=opened.started_at,
+                parent_stage_id=opened.parent_stage_id,
+                operation_id=opened.operation_id,
+                artifact_id=opened.artifact_id,
+            ))
+        components = tuple(dict.fromkeys(stage.provenance.source_component for stage in stages))
+        return RequestTrace(
+            context=self.context,
+            stages=tuple(stages),
+            exported_at=self.clock.utc_now(),
+            complete=complete,
+            source_components=components,
+            error_code=error_code,
+        )
+
+    def safely(self, operation: Callable[[], object], default=None):
+        """Run optional instrumentation without allowing failures into caller behavior."""
+        try:
+            return operation()
+        except Exception:
+            return default
+
+
+__all__ = ["TraceCollector"]