diff --git a/cacheroute_observability/README.md b/cacheroute_observability/README.md
new file mode 100644
index 0000000..3788490
--- /dev/null
+++ b/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.
diff --git a/cacheroute_observability/__init__.py b/cacheroute_observability/__init__.py
new file mode 100644
index 0000000..2ae9381
--- /dev/null
+++ b/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",
+]
diff --git a/cacheroute_observability/clock.py b/cacheroute_observability/clock.py
new file mode 100644
index 0000000..c5bf2b0
--- /dev/null
+++ b/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"]
diff --git a/cacheroute_observability/collector.py b/cacheroute_observability/collector.py
new file mode 100644
index 0000000..55c5166
--- /dev/null
+++ b/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"]
diff --git a/cacheroute_observability/enums.py b/cacheroute_observability/enums.py
new file mode 100644
index 0000000..e0b40b9
--- /dev/null
+++ b/cacheroute_observability/enums.py
@@ -0,0 +1,92 @@
+"""Stable string vocabularies for CacheRoute observability v1."""
+from enum import Enum
+
+
+class _StringEnum(str, Enum):
+    pass
+
+
+class TraceComponent(_StringEnum):
+    CLIENT = "client"
+    SCHEDULER = "scheduler"
+    PROXY = "proxy"
+    KDN = "kdn"
+    GATEWAY = "gateway"
+    INSTANCE = "instance"
+    VLLM = "vllm"
+    LMCACHE = "lmcache"
+    LEGACY_ADAPTER = "legacy_adapter"
+    TEST = "test"
+
+
+class TraceValueKind(_StringEnum):
+    PREDICTED = "predicted"
+    OBSERVED = "observed"
+    ACTUAL = "actual"
+    INFERRED = "inferred"
+    LEGACY_PROJECTED = "legacy_projected"
+
+
+class TraceStageState(_StringEnum):
+    PENDING = "pending"
+    RUNNING = "running"
+    COMPLETED = "completed"
+
+
+class TraceStageOutcome(_StringEnum):
+    SUCCESS = "success"
+    UNSUPPORTED = "unsupported"
+    INCOMPATIBLE = "incompatible"
+    STALE = "stale"
+    PARTIAL = "partial"
+    FAILED = "failed"
+    CANCELLED = "cancelled"
+    SKIPPED = "skipped"
+    TEXT_FALLBACK = "text_fallback"
+    IDEMPOTENCY_CONFLICT = "idempotency_conflict"
+
+
+class TraceStageName(_StringEnum):
+    RUNTIME_PROFILE_RESOLUTION = "runtime_profile_resolution"
+    KNOWLEDGE_LOOKUP = "knowledge_lookup"
+    SEMANTIC_RESOLUTION = "semantic_resolution"
+    ARTIFACT_COMPATIBILITY = "artifact_compatibility"
+    CAPABILITY_SNAPSHOT_DISCOVERY = "capability_snapshot_discovery"
+    TOKEN_LOOKUP = "token_lookup"
+    ARTIFACT_LOOKUP = "artifact_lookup"
+    CACHE_OBSERVATION = "cache_observation"
+    CACHE_OPERATION_QUEUE = "cache_operation_queue"
+    CACHE_PREFETCH_EXECUTION = "cache_prefetch_execution"
+    CACHE_PIN_EXECUTION = "cache_pin_execution"
+    CACHE_UNPIN_EXECUTION = "cache_unpin_execution"
+    CACHE_CLEAR_EXECUTION = "cache_clear_execution"
+    CACHE_REBUILD_EXECUTION = "cache_rebuild_execution"
+    GATEWAY_REQUEST = "gateway_request"
+    GATEWAY_ASYNC_OPERATION = "gateway_async_operation"
+    TIER_ADAPTER_OBSERVATION = "tier_adapter_observation"
+    INSTANCE_LMCACHE_LOAD = "instance_lmcache_load"
+    PROXY_PREPARE_QUEUE = "proxy_prepare_queue"
+    PROXY_READY_QUEUE = "proxy_ready_queue"
+    VLLM_PREFILL = "vllm_prefill"
+    FIRST_TOKEN = "first_token"
+    DECODE = "decode"
+    COMPLETION = "completion"
+    FALLBACK = "fallback"
+    LEGACY_SCAN = "legacy_scan"
+    LEGACY_DUMP = "legacy_dump"
+    LEGACY_RESTORE = "legacy_restore"
+    LEGACY_INJECT = "legacy_inject"
+
+
+class OperationWaiterState(_StringEnum):
+    WAITING = "waiting"
+    COMPLETED = "completed"
+    CANCELLED = "cancelled"
+    DETACHED = "detached"
+    EXPIRED = "expired"
+
+
+__all__ = [
+    "OperationWaiterState", "TraceComponent", "TraceStageName",
+    "TraceStageOutcome", "TraceStageState", "TraceValueKind",
+]
diff --git a/cacheroute_observability/legacy_proxy_trace.py b/cacheroute_observability/legacy_proxy_trace.py
new file mode 100644
index 0000000..3752135
--- /dev/null
+++ b/cacheroute_observability/legacy_proxy_trace.py
@@ -0,0 +1,200 @@
+"""Pure projection from the current free-form Proxy trace into safe v1 contracts."""
+from __future__ import annotations
+
+import hashlib
+from datetime import datetime, timezone
+from typing import Any, Mapping
+
+from .enums import TraceComponent, TraceStageName, TraceStageOutcome, TraceStageState, TraceValueKind
+from .models import RequestTrace, TraceContext, TraceMeasurement, TraceProvenance, TraceStage
+
+
+_STAGE_KEYS: tuple[tuple[TraceStageName, tuple[str, ...]], ...] = (
+    (TraceStageName.RUNTIME_PROFILE_RESOLUTION, ("proxy_recv_ms", "route_select_start_ms", "route_select_end_ms")),
+    (TraceStageName.KNOWLEDGE_LOOKUP, ("kdn_fetch_start_ms", "kdn_fetch_end_ms")),
+    (TraceStageName.PROXY_PREPARE_QUEUE, (
+        "proxy_enqueue_ms", "prepare_queue_enqueue_ms", "prepare_dequeue_ms", "prepare_start_ms",
+        "ready_enqueue_ms", "actual_prepare_ms", "actual_prepare_total_ms", "actual_know_prepare_ms",
+        "prepare_buffer_wait_ms", "predict_prepare_ms", "predict_know_prepare_ms",
+        "predict_prepare_queue_wait_ms", "predict_kv_transfer_ms", "predict_prepare_prefix_ms",
+        "predict_kv_prepare_service_ms", "predict_prepare_initial_ms", "predict_prepare_model_ms",
+        "predict_prepare_corrected_ms", "prepare_error_ms",
+    )),
+    (TraceStageName.INSTANCE_LMCACHE_LOAD, (
+        "kv_inject_queue_enqueue_ms", "kv_inject_reserved_start_ms", "kv_inject_start_ms",
+        "kv_inject_end_ms", "kv_ack_start_ms", "kv_ack_end_ms", "kdn_link_wait_start_ms",
+        "kdn_link_wait_end_ms", "predict_redis_kv_load_ms",
+    )),
+    (TraceStageName.PROXY_READY_QUEUE, (
+        "ready_dequeue_ms", "forward_wait_start_ms", "forward_wait_end_ms", "forward_start_ms",
+        "actual_ready_queue_ms", "predict_queue_wait_ms",
+    )),
+    (TraceStageName.VLLM_PREFILL, (
+        "predict_text_prefill_ms", "predict_residual_prefill_ms", "predict_prefill_service_ms",
+        "predict_vllm_internal_ms", "actual_vllm_internal_ms", "actual_compute_ms",
+    )),
+    (TraceStageName.FIRST_TOKEN, ("pred_first_token_ts_ms", "first_token_ms", "ttft_observable")),
+    (TraceStageName.DECODE, ("decode_start_ms", "decode_end_ms", "predict_decode_ms")),
+    (TraceStageName.COMPLETION, ("forward_end_ms", "actual_total_ms", "predict_total_ms", "predict_error_ms")),
+)
+
+_TIMESTAMP_KEYS = {
+    "proxy_recv_ms", "route_select_start_ms", "route_select_end_ms", "kdn_fetch_start_ms",
+    "kdn_fetch_end_ms", "proxy_enqueue_ms", "prepare_queue_enqueue_ms", "prepare_dequeue_ms",
+    "prepare_start_ms", "ready_enqueue_ms", "prepare_error_ms", "kv_inject_queue_enqueue_ms",
+    "kv_inject_reserved_start_ms", "kv_inject_start_ms", "kv_inject_end_ms", "kv_ack_start_ms",
+    "kv_ack_end_ms", "kdn_link_wait_start_ms", "kdn_link_wait_end_ms", "ready_dequeue_ms",
+    "forward_wait_start_ms", "forward_wait_end_ms", "forward_start_ms", "pred_first_token_ts_ms",
+    "first_token_ms", "decode_start_ms", "decode_end_ms", "forward_end_ms",
+}
+_OVERWRITTEN_PREDICTIONS = {"predict_prepare_ms", "predict_know_prepare_ms", "predict_prepare_corrected_ms"}
+_CLEAR_ACTUAL_NAMES = {
+    "actual_prepare_ms": "proxy_kv_ack_duration",
+    "actual_prepare_total_ms": "proxy_prepare_total_duration",
+    "actual_know_prepare_ms": "proxy_prepare_to_ready_duration",
+    "actual_ready_queue_ms": "proxy_ready_queue_duration",
+    "actual_total_ms": "proxy_enqueue_to_first_chunk_duration",
+    "actual_vllm_internal_ms": "proxy_forward_to_first_chunk_duration",
+    "prepare_buffer_wait_ms": "proxy_prepare_buffer_wait_duration",
+}
+_LEGACY_ALIAS_NAMES = {"actual_compute_ms": "proxy_forward_to_first_chunk_legacy_compute_alias"}
+_KV_ACK_NUMERIC = {"payload_bytes", "network_queue_ms", "network_transfer_ms", "network_total_ms"}
+_PATH_CODES = {
+    "text_inject": (TraceStageOutcome.SUCCESS, "legacy_text_inject"),
+    "no_rag_or_empty_knowledge": (TraceStageOutcome.SKIPPED, "legacy_no_rag_or_empty_knowledge"),
+    "kv_inject": (TraceStageOutcome.SUCCESS, "legacy_kv_inject"),
+    "kv_inject_failed_fallback_text": (TraceStageOutcome.TEXT_FALLBACK, "legacy_kv_inject_failed_text_fallback"),
+    "no_kv_ready_fallback_text": (TraceStageOutcome.TEXT_FALLBACK, "legacy_no_kv_ready_text_fallback"),
+}
+
+
+def _hex_id(prefix: str, *parts: object) -> str:
+    digest = hashlib.sha256("\x1f".join(str(part) for part in parts).encode()).hexdigest()[:32]
+    return f"{prefix}_{digest}"
+
+
+def _millis_timestamp(value: Any) -> datetime | None:
+    if isinstance(value, bool) or not isinstance(value, (int, float)) or value < 0:
+        return None
+    try:
+        return datetime.fromtimestamp(value / 1000, tz=timezone.utc)
+    except (OverflowError, OSError, ValueError):
+        return None
+
+
+def _measurement(key: str, value: Any, provenance: TraceProvenance) -> TraceMeasurement | None:
+    if key in _TIMESTAMP_KEYS:
+        timestamp = _millis_timestamp(value)
+        if timestamp is None:
+            return None
+        kind = TraceValueKind.PREDICTED if key.startswith("pred_") else TraceValueKind.OBSERVED
+        if key in {"first_token_ms", "decode_start_ms", "decode_end_ms", "kv_inject_reserved_start_ms", "kdn_link_wait_end_ms"}:
+            kind = TraceValueKind.LEGACY_PROJECTED
+        return TraceMeasurement(name=key.removesuffix("_ms") + "_timestamp", kind=kind,
+                                provenance=provenance, timestamp=timestamp, legacy_name=key)
+    if isinstance(value, bool) or not isinstance(value, (int, float)) or not __import__("math").isfinite(value):
+        return None
+    if value < 0 and key != "predict_error_ms":
+        return None
+    if key in _CLEAR_ACTUAL_NAMES:
+        return TraceMeasurement(name=_CLEAR_ACTUAL_NAMES[key], kind=TraceValueKind.ACTUAL,
+                                provenance=provenance, duration_ns=int(value * 1_000_000), legacy_name=key)
+    if key in _LEGACY_ALIAS_NAMES:
+        return TraceMeasurement(name=_LEGACY_ALIAS_NAMES[key], kind=TraceValueKind.LEGACY_PROJECTED,
+                                provenance=provenance, duration_ns=max(0, int(value * 1_000_000)), legacy_name=key)
+    kind = TraceValueKind.PREDICTED if key.startswith("predict_") else TraceValueKind.LEGACY_PROJECTED
+    if key in _OVERWRITTEN_PREDICTIONS:
+        kind = TraceValueKind.LEGACY_PROJECTED
+    if key == "ttft_observable":
+        return TraceMeasurement(name=key, kind=TraceValueKind.OBSERVED, provenance=provenance,
+                                boolean=bool(value), legacy_name=key)
+    if key == "predict_error_ms":
+        return TraceMeasurement(name="prediction_error_ms", kind=TraceValueKind.INFERRED,
+                                provenance=provenance, safe_scalar=value, legacy_name=key)
+    if key.endswith("_ms"):
+        return TraceMeasurement(name=key.removesuffix("_ms") + "_duration", kind=kind,
+                                provenance=provenance, duration_ns=int(value * 1_000_000), legacy_name=key)
+    return None
+
+
+def project_legacy_proxy_trace(
+    *,
+    request_id: str,
+    correlation_id: str,
+    legacy_request_id: int | None,
+    trace: Mapping[str, Any],
+    kv_ack: Mapping[str, Any] | None = None,
+    exported_at: datetime,
+    runtime_profile: str,
+) -> RequestTrace:
+    """Project allow-listed scalar fields; raw inputs never enter the result."""
+    trace_id = _hex_id("trace", request_id, correlation_id, legacy_request_id)
+    context = TraceContext(
+        trace_id=trace_id, request_id=request_id, correlation_id=correlation_id,
+        legacy_request_id=legacy_request_id, sampled=True, created_at=exported_at,
+    )
+    provenance = TraceProvenance(
+        source_component=TraceComponent.LEGACY_ADAPTER, runtime_profile=runtime_profile,
+        captured_at=exported_at, legacy=True,
+    )
+    stages: list[TraceStage] = []
+    sequence = 0
+    for stage_name, keys in _STAGE_KEYS:
+        measurements = tuple(
+            measurement for key in keys
+            if key in trace
+            for measurement in [_measurement(key, trace[key], provenance)]
+            if measurement is not None
+        )
+        if not measurements:
+            continue
+        stages.append(TraceStage(
+            stage_id=_hex_id("stage", trace_id, sequence, stage_name.value), sequence=sequence,
+            name=stage_name, state=TraceStageState.COMPLETED, outcome=TraceStageOutcome.SUCCESS,
+            provenance=provenance, measurements=measurements, started_at=exported_at,
+            ended_at=exported_at, duration_ns=0,
+        ))
+        sequence += 1
+
+    if kv_ack:
+        measurements = []
+        for key in sorted(_KV_ACK_NUMERIC):
+            value = kv_ack.get(key)
+            if isinstance(value, (int, float)) and not isinstance(value, bool) and value >= 0:
+                field = "bytes" if key == "payload_bytes" else "duration_ns"
+                converted = int(value) if field == "bytes" else int(value * 1_000_000)
+                measurements.append(TraceMeasurement(
+                    name=f"legacy_kv_ack_{key}", kind=TraceValueKind.LEGACY_PROJECTED,
+                    provenance=provenance, legacy_name=key, **{field: converted},
+                ))
+        if measurements:
+            stages.append(TraceStage(
+                stage_id=_hex_id("stage", trace_id, sequence, "kv_ack"), sequence=sequence,
+                name=TraceStageName.INSTANCE_LMCACHE_LOAD, state=TraceStageState.COMPLETED,
+                outcome=TraceStageOutcome.SUCCESS, provenance=provenance,
+                measurements=tuple(measurements), started_at=exported_at, ended_at=exported_at,
+                duration_ns=0, outcome_code="legacy_kv_ack",
+            ))
+            sequence += 1
+
+    for path_key in ("text_actual_path", "kvcache_actual_path"):
+        path_value = trace.get(path_key)
+        classification = _PATH_CODES.get(path_value) if isinstance(path_value, str) else None
+        if classification is None:
+            continue
+        outcome, code = classification
+        stage_name = TraceStageName.FALLBACK if outcome is TraceStageOutcome.TEXT_FALLBACK else TraceStageName.PROXY_PREPARE_QUEUE
+        stages.append(TraceStage(
+            stage_id=_hex_id("stage", trace_id, sequence, path_key), sequence=sequence,
+            name=stage_name, state=TraceStageState.COMPLETED, outcome=outcome,
+            provenance=provenance, started_at=exported_at, ended_at=exported_at,
+            duration_ns=0, outcome_code=code,
+        ))
+        sequence += 1
+
+    components = (TraceComponent.LEGACY_ADAPTER,) if stages else ()
+    return RequestTrace(context=context, stages=tuple(stages), exported_at=exported_at,
+                        complete=True, source_components=components)
+
+
+__all__ = ["project_legacy_proxy_trace"]
diff --git a/cacheroute_observability/models.py b/cacheroute_observability/models.py
new file mode 100644
index 0000000..36e8237
--- /dev/null
+++ b/cacheroute_observability/models.py
@@ -0,0 +1,415 @@
+"""Frozen, dependency-light CacheRoute observability contracts."""
+from __future__ import annotations
+
+import math
+import re
+from datetime import datetime, timedelta
+from typing import Any, ClassVar
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
+
+from .enums import (
+    OperationWaiterState, TraceComponent, TraceStageName, TraceStageOutcome,
+    TraceStageState, TraceValueKind,
+)
+
+_TRACE_RE = re.compile(r"^trace_[0-9a-f]{32}$")
+_STAGE_RE = re.compile(r"^stage_[0-9a-f]{32}$")
+_SAFE_CODE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,127}$")
+_FORBIDDEN_KEYS = {
+    "authorization", "cookie", "credentials", "device_address", "device_pointer",
+    "generated_content", "http_headers", "kv_bytes", "password", "physical_kv",
+    "physical_path", "prompt", "raw_exception", "redis_key", "request_body",
+    "secret", "tensor", "token_value",
+}
+
+
+def _utc(value: datetime, name: str) -> datetime:
+    if value.tzinfo is None or value.utcoffset() is None or value.utcoffset() != timedelta(0):
+        raise ValueError(f"{name} must be timezone-aware UTC")
+    return value
+
+
+def _nonempty(value: str, name: str, maximum: int = 256) -> str:
+    if not value.strip() or len(value) > maximum:
+        raise ValueError(f"{name} must be non-empty and at most {maximum} characters")
+    return value
+
+
+def _safe_code(value: str, name: str) -> str:
+    _nonempty(value, name, 128)
+    if not _SAFE_CODE_RE.fullmatch(value):
+        raise ValueError(f"{name} must be a safe logical code")
+    return value
+
+
+def _scan_forbidden(value: Any) -> None:
+    if isinstance(value, dict):
+        for key, nested in value.items():
+            normalized = str(key).strip().lower().replace("-", "_")
+            if normalized in _FORBIDDEN_KEYS:
+                raise ValueError(f"forbidden observability field: {key}")
+            _scan_forbidden(nested)
+    elif isinstance(value, (list, tuple, set)):
+        for nested in value:
+            _scan_forbidden(nested)
+
+
+class ContractModel(BaseModel):
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    @model_validator(mode="before")
+    @classmethod
+    def reject_forbidden_fields(cls, value: Any) -> Any:
+        _scan_forbidden(value)
+        return value
+
+    def model_copy(self, *, update=None, deep: bool = False):
+        """Copy through validation so frozen-contract invariants cannot be bypassed."""
+        values = self.model_dump(mode="python")
+        values.update(update or {})
+        return type(self).model_validate(values)
+
+
+class TraceContext(ContractModel):
+    schema_version: str = "cacheroute.trace-context.v1"
+    trace_id: str
+    request_id: str
+    correlation_id: str
+    legacy_request_id: int | None = None
+    sampled: bool = True
+    created_at: datetime
+    expires_at: datetime | None = None
+
+    @field_validator("schema_version")
+    @classmethod
+    def exact_version(cls, value: str) -> str:
+        if value != "cacheroute.trace-context.v1":
+            raise ValueError("unsupported trace context schema version")
+        return value
+
+    @field_validator("trace_id")
+    @classmethod
+    def trace_id_format(cls, value: str) -> str:
+        if not _TRACE_RE.fullmatch(value):
+            raise ValueError("trace_id must use trace_<32 lowercase hex>")
+        return value
+
+    @field_validator("request_id", "correlation_id")
+    @classmethod
+    def bounded_ids(cls, value: str, info) -> str:
+        return _nonempty(value, info.field_name, 256)
+
+    @field_validator("created_at", "expires_at")
+    @classmethod
+    def utc_datetimes(cls, value: datetime | None, info):
+        return None if value is None else _utc(value, info.field_name)
+
+    @model_validator(mode="after")
+    def expiry_order(self):
+        if self.expires_at is not None and self.expires_at <= self.created_at:
+            raise ValueError("expires_at must be later than created_at")
+        return self
+
+
+class TraceProvenance(ContractModel):
+    source_component: TraceComponent
+    runtime_profile: str
+    captured_at: datetime
+    source_endpoint: str | None = None
+    endpoint_id: str | None = None
+    endpoint_generation: int | None = None
+    compatibility_profile_id: str | None = None
+    gateway_profile: str | None = None
+    adapter: str | None = None
+    tier: str | None = None
+    source_version: str | None = None
+    fresh_until: datetime | None = None
+    legacy: bool = False
+
+    @field_validator("runtime_profile")
+    @classmethod
+    def profile(cls, value: str) -> str:
+        return _safe_code(value, "runtime_profile")
+
+    @field_validator(
+        "source_endpoint", "endpoint_id", "compatibility_profile_id", "gateway_profile",
+        "adapter", "tier", "source_version",
+    )
+    @classmethod
+    def safe_labels(cls, value: str | None, info):
+        return None if value is None else _safe_code(value, info.field_name)
+
+    @field_validator("captured_at", "fresh_until")
+    @classmethod
+    def utc_datetimes(cls, value: datetime | None, info):
+        return None if value is None else _utc(value, info.field_name)
+
+    @model_validator(mode="after")
+    def endpoint_and_freshness(self):
+        if (self.endpoint_id is None) != (self.endpoint_generation is None):
+            raise ValueError("endpoint_id and endpoint_generation must appear together")
+        if self.endpoint_generation is not None:
+            minimum = 0 if self.legacy else 1
+            if self.endpoint_generation < minimum:
+                raise ValueError("endpoint_generation is invalid for provenance")
+            if not self.legacy and self.endpoint_generation == 0:
+                raise ValueError("endpoint_generation=0 is Legacy-only")
+        if self.fresh_until is not None and self.fresh_until < self.captured_at:
+            raise ValueError("fresh_until cannot be earlier than captured_at")
+        return self
+
+
+Scalar = str | int | float | bool
+
+
+class TraceMeasurement(ContractModel):
+    _VALUE_FIELDS: ClassVar[tuple[str, ...]] = (
+        "duration_ns", "count", "bytes", "tokens", "ratio", "boolean", "timestamp", "safe_scalar",
+    )
+    name: str
+    kind: TraceValueKind
+    provenance: TraceProvenance
+    duration_ns: int | None = Field(default=None, ge=0)
+    count: int | None = Field(default=None, ge=0)
+    bytes: int | None = Field(default=None, ge=0)
+    tokens: int | None = Field(default=None, ge=0)
+    ratio: float | None = None
+    boolean: bool | None = None
+    timestamp: datetime | None = None
+    safe_scalar: Scalar | None = None
+    observed_at: datetime | None = None
+    expires_at: datetime | None = None
+    uncertainty: float | None = Field(default=None, ge=0)
+    sample_count: int | None = Field(default=None, ge=0)
+    legacy_name: str | None = None
+
+    @field_validator("name", "legacy_name")
+    @classmethod
+    def safe_names(cls, value: str | None, info):
+        return None if value is None else _safe_code(value, info.field_name)
+
+    @field_validator("timestamp", "observed_at", "expires_at")
+    @classmethod
+    def utc_datetimes(cls, value: datetime | None, info):
+        return None if value is None else _utc(value, info.field_name)
+
+    @field_validator("ratio")
+    @classmethod
+    def valid_ratio(cls, value: float | None):
+        if value is not None and (not math.isfinite(value) or not 0 <= value <= 1):
+            raise ValueError("ratio must be finite and between 0 and 1")
+        return value
+
+    @field_validator("safe_scalar", mode="before")
+    @classmethod
+    def scalar_only(cls, value: Any):
+        if value is not None and (isinstance(value, (dict, list, set, tuple)) or not isinstance(value, (str, int, float, bool))):
+            raise ValueError("safe_scalar must be a scalar")
+        if isinstance(value, float) and not math.isfinite(value):
+            raise ValueError("safe_scalar float must be finite")
+        if isinstance(value, str):
+            _nonempty(value, "safe_scalar", 256)
+        return value
+
+    @model_validator(mode="after")
+    def exactly_one_value(self):
+        if sum(getattr(self, field) is not None for field in self._VALUE_FIELDS) != 1:
+            raise ValueError("exactly one typed measurement value is required")
+        if self.expires_at is not None:
+            reference = self.observed_at or self.timestamp or self.provenance.captured_at
+            if self.expires_at <= reference:
+                raise ValueError("measurement expires_at must be later than its observation")
+        return self
+
+
+class TraceStage(ContractModel):
+    stage_id: str
+    sequence: int = Field(ge=0)
+    name: TraceStageName
+    state: TraceStageState
+    outcome: TraceStageOutcome | None = None
+    provenance: TraceProvenance
+    measurements: tuple[TraceMeasurement, ...] = ()
+    started_at: datetime | None = None
+    ended_at: datetime | None = None
+    duration_ns: int | None = Field(default=None, ge=0)
+    parent_stage_id: str | None = None
+    operation_id: str | None = None
+    artifact_id: str | None = None
+    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
+
+    @field_validator("stage_id", "parent_stage_id", "fallback_stage_id")
+    @classmethod
+    def stage_ids(cls, value: str | None):
+        if value is not None and not _STAGE_RE.fullmatch(value):
+            raise ValueError("stage IDs must use stage_<32 lowercase hex>")
+        return value
+
+    @field_validator("operation_id", "artifact_id", "outcome_code", "error_code", "partial_reason")
+    @classmethod
+    def safe_codes(cls, value: str | None, info):
+        return None if value is None else _safe_code(value, info.field_name)
+
+    @field_validator("error_message")
+    @classmethod
+    def sanitized_message(cls, value: str | None):
+        if value is None:
+            return None
+        _nonempty(value, "error_message", 256)
+        if any(char in value for char in ("\n", "\r", "\x00")):
+            raise ValueError("error_message must be a sanitized single line")
+        return value
+
+    @field_validator("started_at", "ended_at")
+    @classmethod
+    def utc_datetimes(cls, value: datetime | None, info):
+        return None if value is None else _utc(value, info.field_name)
+
+    @model_validator(mode="after")
+    def lifecycle(self):
+        if self.state is TraceStageState.COMPLETED and self.outcome is None:
+            raise ValueError("completed stages require an outcome")
+        if self.state is not TraceStageState.COMPLETED and self.outcome is not None:
+            raise ValueError("pending/running stages cannot claim a terminal outcome")
+        if self.ended_at is not None and self.started_at is None:
+            raise ValueError("ended_at requires started_at")
+        if self.started_at is not None and self.ended_at is not None and self.ended_at < self.started_at:
+            raise ValueError("ended_at cannot precede started_at")
+        if self.state is TraceStageState.COMPLETED and (self.ended_at is None or self.duration_ns is None):
+            raise ValueError("completed stages require ended_at and duration_ns")
+        return self
+
+
+class RequestTrace(ContractModel):
+    schema_version: str = "cacheroute.trace.v1"
+    context: TraceContext
+    stages: tuple[TraceStage, ...] = ()
+    exported_at: datetime
+    complete: bool = False
+    source_components: tuple[TraceComponent, ...] = ()
+    error_code: str | None = None
+
+    @field_validator("schema_version")
+    @classmethod
+    def exact_version(cls, value: str) -> str:
+        if value != "cacheroute.trace.v1":
+            raise ValueError("unsupported request trace schema version")
+        return value
+
+    @field_validator("exported_at")
+    @classmethod
+    def exported_utc(cls, value: datetime):
+        return _utc(value, "exported_at")
+
+    @field_validator("error_code")
+    @classmethod
+    def safe_error(cls, value: str | None):
+        return None if value is None else _safe_code(value, "error_code")
+
+    @model_validator(mode="after")
+    def sequence_order(self):
+        sequences = [stage.sequence for stage in self.stages]
+        if sequences != sorted(sequences) or len(sequences) != len(set(sequences)):
+            raise ValueError("stages must have unique ascending sequence values")
+        expected = tuple(dict.fromkeys(stage.provenance.source_component for stage in self.stages))
+        if self.source_components and self.source_components != expected:
+            raise ValueError("source_components must follow first stage occurrence")
+        return self
+
+
+class CacheOperationTrace(ContractModel):
+    schema_version: str = "cacheroute.cache-operation-trace.v1"
+    operation_id: str
+    trace_id: str
+    operation: str
+    state: str
+    stages: tuple[TraceStage, ...]
+    provenance: TraceProvenance
+    artifact_id: str | None = None
+    endpoint_id: str | None = None
+    endpoint_generation: int | None = None
+    created_at: datetime
+    updated_at: datetime
+    completed_at: datetime | None = None
+    error_code: str | None = None
+    fallback_code: str | None = None
+
+    @field_validator("operation_id", "operation", "state", "artifact_id", "endpoint_id", "error_code", "fallback_code")
+    @classmethod
+    def safe_values(cls, value: str | None, info):
+        return None if value is None else _safe_code(value, info.field_name)
+
+    @field_validator("trace_id")
+    @classmethod
+    def trace_id_format(cls, value: str):
+        if not _TRACE_RE.fullmatch(value):
+            raise ValueError("trace_id must use trace_<32 lowercase hex>")
+        return value
+
+    @field_validator("created_at", "updated_at", "completed_at")
+    @classmethod
+    def utc_datetimes(cls, value: datetime | None, info):
+        return None if value is None else _utc(value, info.field_name)
+
+    @model_validator(mode="after")
+    def consistency(self):
+        if self.updated_at < self.created_at or (self.completed_at is not None and self.completed_at < self.created_at):
+            raise ValueError("operation timestamps are out of order")
+        if (self.endpoint_id is None) != (self.endpoint_generation is None):
+            raise ValueError("endpoint_id and endpoint_generation must appear together")
+        if self.endpoint_generation is not None:
+            minimum = 0 if self.provenance.legacy else 1
+            if self.endpoint_generation < minimum or (not self.provenance.legacy and self.endpoint_generation == 0):
+                raise ValueError("endpoint_generation is invalid")
+        sequences = [stage.sequence for stage in self.stages]
+        if sequences != sorted(sequences) or len(sequences) != len(set(sequences)):
+            raise ValueError("operation stages must have unique ascending sequence values")
+        return self
+
+
+class OperationWaiterLink(ContractModel):
+    operation_id: str
+    trace_id: str
+    request_id: str
+    correlation_id: str
+    attach_sequence: int = Field(ge=0)
+    attached_at: datetime
+    state: OperationWaiterState
+    detached_at: datetime | None = None
+    reason: str | None = None
+
+    @field_validator("operation_id", "request_id", "correlation_id", "reason")
+    @classmethod
+    def safe_values(cls, value: str | None, info):
+        return None if value is None else _safe_code(value, info.field_name) if info.field_name in {"operation_id", "reason"} else _nonempty(value, info.field_name, 256)
+
+    @field_validator("trace_id")
+    @classmethod
+    def trace_id_format(cls, value: str):
+        if not _TRACE_RE.fullmatch(value):
+            raise ValueError("trace_id must use trace_<32 lowercase hex>")
+        return value
+
+    @field_validator("attached_at", "detached_at")
+    @classmethod
+    def utc_datetimes(cls, value: datetime | None, info):
+        return None if value is None else _utc(value, info.field_name)
+
+    @model_validator(mode="after")
+    def detach_order(self):
+        if self.detached_at is not None and self.detached_at < self.attached_at:
+            raise ValueError("detached_at cannot precede attached_at")
+        return self
+
+
+__all__ = [
+    "CacheOperationTrace", "OperationWaiterLink", "RequestTrace", "TraceContext",
+    "TraceMeasurement", "TraceProvenance", "TraceStage",
+]
diff --git a/doc/research/issue-141-unified-observability.md b/doc/research/issue-141-unified-observability.md
new file mode 100644
index 0000000..195f48b
--- /dev/null
+++ b/doc/research/issue-141-unified-observability.md
@@ -0,0 +1,590 @@
+# Issue #141 research: unified v1 LMCache observability
+
+Status: design proposal only<br>
+Baseline audited: `ad087d1` (current `main` lineage, including merged PR #154)<br>
+Lifecycle: temporary design record for Issue #141. Remove this file after the Issue is completed and its stable content has been moved into maintained package and component documentation.
+
+Non-goals: production instrumentation, Gateway I/O, routing/queue changes, or a tracing backend.
+
+## 1. Executive decision record
+
+1. Put the dependency-light contracts in **`cacheroute_observability/`**. Scheduler,
+   Proxy, KDN, Gateway, and Instance can import this standalone package without
+   loading `core` or another component; placing it under `kdn_server` would invert
+   dependencies for Scheduler/Proxy/Instance.
+2. Make an exported trace an immutable, versioned snapshot. Build it by appending
+   immutable stages/measurements to a process-local collector; never mutate a
+   measurement from `predicted` into `actual`.
+3. Use UTC timestamps for correlation and per-process monotonic nanoseconds for
+   durations. Never subtract monotonic readings produced by different processes.
+4. Propagate a small internal trace envelope alongside the current serialized
+   `Request`, stripping it before the OpenAI/vLLM body. During migration, retain
+   `Request_ID`, `scheduler-request-id`, the free-form `ProxyTask.trace`, and
+   `cacheroute_meta` unchanged.
+5. Treat Prometheus before/after deltas as endpoint aggregates. Only an isolated,
+   single-request validation run may label them `inferred`; concurrent deltas are
+   never request-correlated actuals.
+
+## 2. Current path and identity inventory
+
+```text
+client payload
+  -> Scheduler allocates cyclic integer Request_ID (1..65535)
+     -> serialized core.request.Request + scheduler-request-id HTTP header
+        -> Proxy restores Request_ID; ProxyTask.request_id duplicates it
+           -> OpenAI-style instance_body (identity is dropped)
+              -> Instance forwards the body unchanged (identity remains absent)
+                 -> vLLM/LMCache
+
+Proxy prepare path
+  -> KDN text lookup (no trace/correlation envelope)
+  -> Instance /v1/kv/inject_ready {request_id, ...}
+     -> KDN /knowledge/inject_ready_kv {request_id, ...}
+
+Versioned KDN/Gateway contracts (currently separate from serving path)
+  -> string request_id (default req_<uuid>) + optional correlation_id
+  -> runtime_profile + created_at
+  -> targeted calls add compatibility_profile_id, endpoint_id, generation
+  -> Gateway responses preserve those fields
+```
+
+### Existing correlation fields
+
+| Field | Location/semantics | Finding |
+|---|---|---|
+| `Request.Request_ID` | Scheduler-assigned cyclic `int` | Not globally unique; reuse after 65,535 and across Scheduler processes/restarts. |
+| `ProxyTask.request_id` | Optional `int`, copied from `Request_ID` | Useful local alias, but annotation conflicts with KDN v1 string IDs. |
+| `scheduler-request-id` | Scheduler-to-Proxy HTTP header | Redundant with payload; not forwarded beyond Proxy. |
+| KDN v1 `request_id` | `VersionedMessage`, non-empty string, defaults `req_<uuid>` | Strong message identity, but the production serving path does not use these contracts yet. |
+| KDN v1 `correlation_id` | Optional string | Correct place to retain an end-to-end correlation, but currently absent from `core.Request`. |
+| Cache operation `task_id` | `cacheop_<uuid>` | Operation identity, not a request identity. |
+| `idempotency_key` | Cache operation and queue work | Logical deduplication identity; must not be overloaded as trace correlation. |
+| artifact/observation/endpoint/work IDs | Canonical or UUID-backed domain IDs | Resource identity only. |
+
+`TraceContext` should therefore carry both an opaque globally unique `request_id`
+and a `correlation_id`; the legacy Scheduler integer belongs in
+`legacy_request_id`. A generated trace ID must not replace either.
+
+### Current `cacheroute_meta`
+
+Both streaming and non-streaming Proxy responses expose exactly: `request_id`,
+the entire mutable `trace` dictionary, `kv_ack`, `kv_ready_kids`,
+`text_only_kids`, `miss_kids`, and `error`. Streaming inserts an
+`event: cacheroute_meta` immediately before `[DONE]`; completions add
+`_cacheroute_meta` to parsed JSON (or the raw-response wrapper). This behavior is
+a compatibility contract for the rollout.
+
+## 3. Complete `ProxyTask.trace` audit and classification
+
+Kinds below mean: **P** predicted, **O** observed snapshot/event, **A** actual
+duration measured at the named boundary, **I** inferred/derived, **L**
+legacy-projected compatibility, and **X** ambiguous or incorrectly named.
+All `*_ms` timestamps are wall-clock epoch milliseconds today unless explicitly
+described as durations.
+
+### Wall-clock event markers
+
+| Keys | Kind | Meaning / issue |
+|---|---|---|
+| `proxy_recv_ms`, `route_select_start_ms`, `route_select_end_ms` | O | Proxy receipt and local selection events. |
+| `proxy_enqueue_ms`, `prepare_queue_enqueue_ms`, `prepare_dequeue_ms`, `prepare_start_ms` | O | Prepare queue lifecycle; first two currently receive the same reading. |
+| `kdn_fetch_start_ms`, `kdn_fetch_end_ms` | O | Proxy-observed KDN fetch boundary. |
+| `text_prefill_build_start_ms`, `text_prefill_build_end_ms`, `prompt_injected_ms` | O | Local body construction/injection markers; “prefill” is a misnomer because no vLLM prefill occurs here. |
+| `kdn_link_wait_start_ms`, `kdn_link_wait_end_ms` | X | Wall-clock reservation model points, not necessarily observed wait start/end; `end` can be a predicted future time. |
+| `kv_inject_queue_enqueue_ms`, `kv_inject_reserved_start_ms` | X | Queue marker plus predicted reservation time; names do not disclose differing value kinds. |
+| `kv_inject_start_ms`, `kv_inject_end_ms`, `kv_ack_start_ms`, `kv_ack_end_ms` | O | Proxy/Instance-control-call boundary; injection completion is inferred from the acknowledgement, not LMCache consumption. |
+| `prepare_self_done_ms`, `ready_enqueue_ms`, `ready_dequeue_ms` | O | Prepare completion/buffer release/ready dequeue. |
+| `forward_wait_start_ms`, `forward_wait_end_ms`, `forward_start_ms`, `forward_end_ms` | O | Dispatch wait and Proxy-to-Instance response boundary. |
+| `first_token_ms` | X | First non-empty downstream chunk, which may be SSE metadata rather than a decoded token. |
+| `decode_start_ms`, `decode_end_ms` | X | Locally inferred state transition and forward end; neither directly measures vLLM decode. |
+| `prepare_error_ms`, `ready_failed_before_forward_ms`, `stream_exception_ms` | O | Failure event timestamps. |
+| `pred_forward_start_ts_ms`, `pred_first_token_ts_ms`, `pred_forward_end_ts_ms`, `pred_worker_free_ts_ms` | P | Predicted epoch times. They must remain predictions even after recomputation. |
+| `kdn_link_free_before`, `kdn_link_free_after` | X | Predicted/shared model wall-clock values; units are hidden and “free” is not an observation of the remote link. |
+
+### Durations, counts, decisions, and outcomes
+
+| Keys | Kind | Meaning / issue |
+|---|---|---|
+| `predict_text_prefill_ms`, `predict_redis_kv_load_ms`, `predict_residual_prefill_ms` | P | Model estimates, not runtime measurements. |
+| `predict_prepare_initial_ms`, `predict_prepare_queue_wait_ms`, `predict_kv_transfer_ms`, `predict_prepare_prefix_ms`, `predict_kv_prepare_service_ms`, `predict_prepare_model_ms`, `predict_prepare_corrected_ms` | P | Evolving prepare estimates. “corrected” is derived after completion and is no longer a prediction. |
+| `predict_know_prepare_ms`, `predict_prepare_ms` | **X** | Initially predicted, then overwritten with wall-clock-derived actual prepare duration at ready release/success. This is the clearest kind violation. |
+| `predict_queue_wait_ms`, `predict_compute_ms`, `predict_prefill_service_ms`, `predict_vllm_internal_ms`, `predict_decode_ms`, `predict_cold_start_extra_ms`, `predict_total_ms`, `predict_wait_ms` | P | Reservation/model outputs; `predict_wait_ms` is a legacy alias for knowledge preparation, not queue wait. |
+| `predict_error_ms` | I | Signed residual `actual_total_ms - predict_total_ms`, not a prediction. |
+| `actual_prepare_ms`, `actual_prepare_total_ms`, `prepare_buffer_wait_ms` | A | Wall-clock-derived local/control-call durations; susceptible to wall-clock adjustment. “prepare” has call-only and end-to-end meanings. |
+| `actual_total_ms`, `actual_know_prepare_ms`, `actual_ready_queue_ms`, `actual_vllm_internal_ms` | A | Wall-clock-derived Proxy-observed durations to first non-empty chunk. `actual_vllm_internal_ms` includes network, Instance, vLLM queue, and prefill. |
+| `actual_compute_ms` | L | Alias of `actual_vllm_internal_ms`; it is not compute-only. |
+| `prepare_seq`, `ready_release_seq`, `ready_worker_idx`, `prepared_buffer_size`, `recompute_generation` (task field) | O | Local ordering/state observations. `recompute_generation` is not exported in `trace`. |
+| `ready_release_blocked_seq` | O | Optional observed sequence; dictionary annotation does not allow `None`. |
+| `ready_release_bypass`, `kv_link_reserved`, `ttft_observable` | O | Integer booleans. Typed v1 should serialize booleans. |
+| `predict_length_tokens`, `predict_bs`, `predict_total_tokens`, `predict_reused_tokens`, `predict_residual_tokens` | P | Inputs/outputs of the model; token reuse is predicted rather than observed. |
+| `predict_kv_bandwidth_mbps` | P/I | Link-snapshot/env/default input used by prediction; provenance determines whether observed or inferred. |
+| `predict_kv_bandwidth_source` | P | Provenance string; violates `Dict[str, int]`. |
+| `predict_stage` | P | Free-form reservation phase (`prefill`/`decode`), not an observed execution stage. |
+| `injection_mode` | O | Chosen request mode, not proof that KV was used. |
+| `text_actual_path` | X | Free-form outcome (`text_inject`, `no_rag_or_empty_knowledge`). |
+| `kvcache_actual_path` | X | Free-form outcome (`kv_inject`, `kv_inject_failed_fallback_text`, `no_kv_ready_fallback_text`). It combines attempted path, failure, and fallback. |
+| `ready_release_policy` | O | Policy label, violates `Dict[str, int]`; contextual attribute rather than measurement. |
+| `prepare_failed_injection_mode` | O | Mode at failure, violates dictionary annotation. |
+| `first_token_missing_reason` | X | Free-form outcome string, violates dictionary annotation. |
+
+The declared `trace: Dict[str, int]` is incorrect: current values include floats
+(`predict_kv_bandwidth_mbps`), strings, `None`, and integer booleans. `mark()`
+also forces arbitrary values to `int`, although most call sites bypass it.
+
+### Separately stored ProxyTask prediction/state fields
+
+`predict_stage`, `pred_slot_idx`, `pred_slot_ready_ts_ms`,
+`pred_forward_start_ts_ms`, `pred_prefill_start_ts_ms`,
+`pred_first_token_ts_ms`, `pred_decode_ms`, `pred_forward_end_ts_ms`,
+`pred_worker_free_ts_ms`, and `pred_service_ms` are prediction state outside the
+dictionary. `reservation_seq` and `recompute_generation` order reservation
+updates; `has_started_forward` and `has_seen_first_token` are observed local
+booleans. Several values are copied into `trace`, producing two mutable sources
+of truth. `created_at` is a wall-clock float and is not currently exported.
+
+### Knowledge, injection, failure, and completion markers outside `trace`
+
+* `kv_ready_kids`, `text_only_kids`, and `miss_kids` are KDN-result
+  classifications, not proof of cache consumption.
+* `kv_ready_meta` holds untyped KDN metadata and is not in `cacheroute_meta`.
+* `kv_ack` exposes the Instance/KDN acknowledgement, including current
+  `payload_bytes`, `network_queue_ms`, `network_transfer_ms`, and
+  `network_total_ms` when supplied. Their producer clock/boundary is not encoded.
+* `error` is a free-form `prepare_failed: ...`, `ready_failed: ...`, or
+  `stream_wrap_failed: ...` string. Exceptions can leak implementation detail.
+* There is no direct vLLM prefill-start/end, decoded-token event, LMCache
+  load-start/end, or server completion event. Proxy `forward_end_ms` is response
+  consumption completion; non-streaming `first_token_ms` is only first body
+  chunk and explicitly sets `ttft_observable=0`.
+
+## 4. Other current metrics and attribution
+
+Proxy metrics are predictors rather than per-request telemetry:
+`queue_predictor`/TTFT regressors estimate prefill time,
+`decode_tpot_predictor` estimates decode service, and the Redis regressor
+estimates pull time. Queue snapshots report aggregate instantaneous queue/active
+counts plus reservation-derived backlog. Instance registration reports a frozen
+capability object (model/tokenizer/adapters/KV layout/parallelism and detected
+vLLM/LMCache versions), while heartbeats usually send only its fingerprint.
+Topology reporting contains endpoint-level measured/inferred latency and link
+bandwidth. None is a request-correlated execution measurement.
+
+### Validator metric-attribution table
+
+The validator sums every Prometheus series with each metric name, losing label
+dimensions, then subtracts before from after. Classification assumes the current
+unlabelled request path.
+
+| Metric | Request actual | Endpoint aggregate | Isolated-test inference | Concurrent attribution |
+|---|---:|---:|---:|---:|
+| `vllm:external_prefix_cache_queries_total` | No | Yes | Yes: query occurred | Unsupported |
+| `vllm:external_prefix_cache_hits_total` | No | Yes | Yes: some external hit occurred | Unsupported |
+| `vllm:prompt_tokens_cached_total` | No | Yes | Yes: cached-token delta | Unsupported |
+| `vllm:prompt_tokens_total` | No | Yes | Yes: prompt-token delta | Unsupported |
+| `lmcache_mp_lookup_requested_tokens_total` | No | Yes | Yes: requested tokens | Unsupported |
+| `lmcache_mp_lookup_hit_tokens_total` | No | Yes | Yes: hit tokens/rate | Unsupported |
+| `lmcache_mp_l2_prefetch_lookup_requests_total` | No | Yes | Yes | Unsupported |
+| `lmcache_mp_l2_prefetch_lookup_objects_chunks_total` | No | Yes | Yes | Unsupported |
+| `lmcache_mp_l2_prefetch_hit_chunks_total` | No | Yes | Yes | Unsupported |
+| `lmcache_mp_l2_prefetch_load_submitted_requests_total` | No | Yes | Yes | Unsupported |
+| `lmcache_mp_l2_prefetch_load_submitted_objects_chunks_total` | No | Yes | Yes | Unsupported |
+| `lmcache_mp_l2_prefetch_load_completed_chunks_total` | No | Yes | Yes | Unsupported |
+| `lmcache_mp_l2_load_completed_requests_total` | No | Yes | Yes | Unsupported |
+| `lmcache_mp_l2_prefetch_failure_chunks_total` | No | Yes | Yes: absence/presence of failures | Unsupported |
+| `lmcache_mp_l1_read_chunks_total` | No | Yes | Yes | Unsupported |
+| `lmcache_mp_l1_write_chunks_total` | No | Yes | Yes | Unsupported |
+
+Even in an isolated test, record `value_kind=inferred`, the before/after scrape
+times, endpoint, full label set (if retained), and the exclusivity assumption.
+Counter reset, delayed emission, background work, and scrape races can invalidate
+the inference. Future public LMCache events or request-labelled metrics can be
+ingested through a small optional observer adapter and emitted as `observed` or
+`actual` only when the upstream event specifies the request/operation identity
+and measurement boundary. The core schema must not import or depend on a private
+LMCache API.
+
+## 5. Where context is lost
+
+1. Client-to-Scheduler has no canonical correlation ID or trace controls.
+2. The Scheduler integer ID is cyclic and process-local; it has no boot/session
+   namespace.
+3. Proxy reconstructs `Request` but does not receive a typed trace context;
+   route-selection time begins only at Proxy receipt.
+4. `build_body_for_instance()` intentionally produces an OpenAI body and drops
+   `Request_ID`; Proxy-to-Instance forwarding sends no trace headers.
+5. Instance forwards the same body to vLLM, so it cannot correlate vLLM/LMCache
+   observation with the Scheduler request.
+6. Text lookup calls do not propagate identity. KV injection propagates only the
+   legacy integer `request_id`, with no operation/trace/correlation ID.
+7. Gateway v1 contracts have good request/correlation/target provenance but are
+   not connected to the production serving path.
+8. Streaming byte forwarding identifies the first non-empty chunk, not the first
+   decoded token, and does not parse upstream IDs/events.
+9. `cacheroute_meta` is assembled only at Proxy response export; stages from
+   Scheduler, KDN, Instance, vLLM, and LMCache cannot currently join it.
+
+## 6. Package-boundary evaluation
+
+| Candidate | Benefits | Problems | Decision |
+|---|---|---|---|
+| `kdn_server/observability/` | Close to #139/#140 types | Forces Proxy, Scheduler, Instance, and `core` to depend on a server package; likely dependency inversion and future import cycles. | Reject. |
+| `cacheroute_observability/` | Standalone dependency-light boundary; simple imports; same repository/version | Must enforce that it imports only the standard library and Pydantic. | **Implemented.** |
+| separately released distribution | Strong release isolation | Independent packaging/versioning overhead is unnecessary in Phase 1. | Revisit only if contracts need an independent release cycle. |
+
+`cacheroute_observability/` imports only Python stdlib and Pydantic. It defines
+trace-only enums and uses validated opaque resource IDs instead of importing or
+duplicating #139 domain payloads, #140 contracts, or Gateway implementations. It
+must never import FastAPI, Redis, vLLM, LMCache, GPU libraries, `httpx`, network
+clients, application components, configuration, or forwarding modules.
+
+## 7. Stable v1 schemas
+
+All models use Pydantic `ConfigDict(frozen=True, extra="forbid")`, stable
+`schema_version="cacheroute.trace.v1"`, JSON-mode ISO-8601 UTC timestamps, enum
+wire values, arrays (not sets), and deterministic ordering. Export returns a new
+frozen snapshot; collection is append-only and process-local. Unknown future
+enum values require a version change rather than silently accepting route names.
+
+### `TraceContext`
+
+Required: `trace_id` (`trace_<32 lowercase hex>`), opaque non-empty `request_id`,
+opaque non-empty `correlation_id`, `sampled: bool`, and `created_at` UTC. Optional:
+`parent_trace_id`, `legacy_request_id: int`, and `expires_at` UTC. `request_id`
+identifies one end-user generation request; `correlation_id` groups retries or
+related requests; `trace_id` identifies one exported timeline. Validate expiry
+after creation. Context contains no prompt, token content, credentials, Redis
+keys, physical KV bytes/pointers/paths, or arbitrary baggage.
+
+### `TraceStageName`
+
+Stable values, ordered by a separate integer rank (not enum declaration order):
+
+```text
+runtime_profile_resolution
+knowledge_lookup
+semantic_resolution
+artifact_compatibility
+capability_snapshot_discovery
+token_lookup
+artifact_lookup
+cache_observation
+cache_operation_queue
+cache_prefetch_execution
+cache_pin_execution
+cache_unpin_execution
+cache_clear_execution
+cache_rebuild_execution
+gateway_request
+gateway_async_operation
+tier_adapter_observation
+instance_lmcache_load
+proxy_prepare_queue
+proxy_ready_queue
+vllm_prefill
+first_token
+decode
+completion
+fallback
+legacy_scan
+legacy_dump
+legacy_restore
+legacy_inject
+```
+
+Queue operation is generic and carries `operation` from existing
+`CacheOperationType`; execution stages are explicit because they have materially
+different safety/latency semantics. Do not add `/route` strings, Python method
+names, Redis calls, or LMCache private event names as stage values.
+
+### `TraceValueKind`
+
+Stable values: `predicted`, `observed`, `actual`, `inferred`,
+`legacy_projected`. “Ambiguous” is an audit classification, not a valid new
+measurement kind; adapters must choose a truthful kind or omit the measurement.
+
+### `TraceStageOutcome`
+
+Stable values: `pending`, `running`, `succeeded`, `failed`, `cancelled`,
+`unsupported`, `stale`, `partial`, `skipped`, `fallback`. Map #140 outcomes
+directly: `success -> succeeded`, all identical values remain identical,
+`incompatible` is represented as `failed` with `outcome_code="incompatible"`,
+`text_fallback -> fallback`, and `idempotency_conflict -> failed`. Maintainer
+approval is needed on whether to add `incompatible` and `idempotency_conflict`
+directly instead of keeping the #140 code alongside the stage outcome.
+
+### `TraceProvenance`
+
+Required: `source_component` (`client|scheduler|proxy|kdn|gateway|instance|vllm|lmcache|legacy_adapter|test`),
+`runtime_profile`, and `captured_at` UTC. Optional: `source_endpoint` (logical
+endpoint/route label, never credentials or URL query), `endpoint_id`,
+`endpoint_generation`, `gateway_profile`, `compatibility_profile_id`,
+`observation_source`, `profile_id`, `adapter`, `tier`, `source_version`,
+`fresh_until`, and `legacy: bool`. Targeted Gateway observations require endpoint
+ID and generation together; non-Legacy generation is positive, Legacy unknown is
+zero. Freshness follows #139 observation semantics and cannot precede capture.
+
+### `TraceMeasurement`
+
+Required: `name` (a versioned allow-listed measurement name), `kind`, one and
+only one of `duration_ns`, `count`, `bytes`, `tokens`, `ratio`, `boolean`,
+`timestamp`, or safe scalar `value`, plus `provenance`. Optional: `unit` only for
+generic scalar values, `uncertainty`, `sample_count`, `observed_at`, `expires_at`,
+and `legacy_name`. Duration unit is structurally fixed to nanoseconds and must be
+non-negative. Ratios are finite and `[0,1]`; counts/bytes/tokens are non-negative.
+Predicted and actual values are separate objects and may share a semantic name.
+Generic values reject mappings/lists to prevent arbitrary payload smuggling.
+
+### `TraceStage`
+
+Required: `stage_id` (`stage_<32 hex>`), `name`, `sequence` (non-negative,
+process-assigned append sequence), `outcome`, `provenance`, and tuple
+`measurements`. Optional: UTC `started_at`/`ended_at`, `duration_ns`,
+`operation_id`, `artifact_id`, safe `outcome_code`, safe `error_code`, sanitized
+`error_message`, `retryable`, `fallback_eligible`, `fallback_stage_id`,
+`parent_stage_id`, `partial_reason`, and `legacy`. Start/end are correlation
+markers only. If both exist, order them by UTC but do not derive the duration from
+them. Duration is local monotonic elapsed time. A terminal stage is immutable;
+an in-progress collector state is not exported as the same object.
+
+### `RequestTrace`
+
+Required: `context`, ordered tuple `stages`, `exported_at` UTC, `complete: bool`,
+and schema version. Optional: `source_components`, sanitized top-level
+`error_code`, `legacy_trace` (JSON-safe copy of current dictionary), and
+`legacy_cacheroute_meta` during migration. Canonical ordering is
+`(stage-rank, started_at-or-max, source_component, sequence, stage_id)`; sequence
+breaks ties but does not imply cross-process causality. JSON uses enum strings,
+UTC `Z` timestamps, integer nanoseconds, and stable tuple-to-array conversion.
+
+### `CacheOperationTrace`
+
+Required: existing `CacheOperationTask.task_id` as `operation_id`, operation
+type, `trace_id`, ordered stages, current existing task state, and provenance.
+Optional: artifact/endpoint IDs and generation, idempotency key hash (never the
+raw secret-like key), created/updated/expiry UTC timestamps, sanitized error and
+fallback fields, and legacy projection. It references rather than duplicates
+`CacheArtifact`, `CacheReplicaObservation`, `LMCacheEndpoint`, or
+`CacheOperationTask`.
+
+### `OperationWaiterLink`
+
+Required: `operation_id`, request `trace_id`, request/correlation IDs,
+`attached_at`, monotonically allocated `attach_sequence`, and state
+`waiting|completed|cancelled|detached|expired`. Optional: `detached_at`, safe
+reason, and operation outcome observed at attachment. This is a correlation
+record, not scheduler state and cannot influence dispatch.
+
+### Secret and physical-KV rejection
+
+All models use `extra="forbid"` plus recursive validators rejecting case-folded
+field names such as `password`, `secret`, `authorization`, `cookie`, `token_value`,
+`redis_key`, `kv_bytes`, `physical_kv`, `tensor`, `pointer`, `device_address`, and
+absolute/file paths. `tokens` means a numeric count only. Prompts, generated
+content, raw exceptions, HTTP bodies/headers, Redis keys, manifests, physical KV
+locations, and credentials are never valid trace values. Legacy projection
+allow-lists known scalar keys rather than embedding the whole acknowledgement.
+
+## 8. Clock design
+
+Inject a `TraceClock` with `utc_now()` and `monotonic_ns()`; production uses
+timezone-aware `datetime.now(timezone.utc)` and `time.perf_counter_ns()`.
+Tests use a deterministic fake. On local stage start, store both readings in the
+mutable collector. On finish, calculate `duration_ns = end_mono - start_mono`,
+reject negatives, and export UTC start/end plus the local duration. Never export
+monotonic readings, compare them across processes, or derive durations by
+subtracting epoch milliseconds.
+
+A stage spanning Proxy and Instance is represented as either (a) two linked
+local stages with independently measured durations, or (b) one parent stage with
+UTC correlation markers and child stages. Its end-to-end duration is measured by
+one process that owns both send and response receipt (for example, Proxy Gateway
+request duration). Remote work has its own Instance duration. Network/clock
+offset must not be inferred as the difference between unrelated monotonic clocks.
+
+## 9. Compatibility-first propagation
+
+| Mechanism | Use | Decision |
+|---|---|---|
+| Add fields directly to current Request | Easy serialization, but changes a widely used dataclass/wire body | Do not use in Phase 1/2. |
+| Dedicated internal envelope | Separates routing body and trace context; explicit filtering | Primary mechanism from Phase 3. |
+| HTTP headers | Works at hops and avoids OpenAI JSON pollution; intermediaries may strip them | Mirror only allow-listed context IDs per internal hop. |
+| Process-local collector reference | Efficient within one process, impossible across processes | Use only behind adapters; never serialize references. |
+
+Recommended envelope:
+
+```json
+{
+  "request": {"Request_ID": 42, "Prompt": {}, "Service": {}, "Task": {}},
+  "_cacheroute_internal": {
+    "trace_context": {
+      "schema_version": "cacheroute.trace-context.v1",
+      "trace_id": "trace_...",
+      "request_id": "req_...",
+      "correlation_id": "corr_...",
+      "legacy_request_id": 42,
+      "sampled": true
+    }
+  }
+}
+```
+
+Proxy accepts both bare legacy Request and envelope. It strips the internal
+section before constructing `instance_body`, then sends only allow-listed
+`x-cacheroute-trace-id`, `x-cacheroute-request-id`, and
+`x-cacheroute-correlation-id` to Instance. Instance strips these before vLLM
+unless a public adapter explicitly supports correlation. KDN/Gateway versioned
+calls place IDs in their existing request fields; headers are hop diagnostics,
+not competing identity. Streaming merges immutable snapshots at the final meta
+event; non-streaming does so in `_cacheroute_meta`. Existing meta fields and
+dictionary remain unchanged while an optional `request_trace_v1` is added.
+
+If tracing is disabled, no collector/stages are allocated and propagation may be
+limited to existing IDs. If not sampled, propagate context IDs with
+`sampled=false` for correlation but collect/export no detailed measurements.
+Sampling is decided once at ingress using a deterministic policy and never
+changed downstream. Any trace validation/export error is swallowed at the
+observability boundary with a rate-limited diagnostic; it must never change
+routing, forwarding, fallback, reuse, queue ordering, or responses.
+
+## 10. Shared-operation correlation
+
+Maintain a bounded `operation_id -> ordered waiter links` registry in the KDN or
+Gateway orchestration layer, separate from `CacheOperationTask` and scheduling.
+
+* **Attach:** idempotent on `(operation_id, trace_id)`. Duplicate legacy request
+  IDs are allowed because trace ID is authoritative. Return the existing link on
+  retry; allocate a stable sequence only once.
+* **Detach/cancel:** request cancellation transitions only its link; it does not
+  cancel shared work unless the existing operation owner independently decides
+  to do so. No scheduling-policy change is implied.
+* **Completion races:** retain terminal operation summaries until TTL. A waiter
+  registering after completion receives a completed link and the same immutable
+  summary. Completion snapshots the then-current waiters; later links remain
+  deterministically ordered after them.
+* **Expiry:** terminal task/link TTL is explicit. Active-task entries have a
+  maximum lifetime and emit `expired`; purge uses injected UTC/monotonic clocks.
+* **Bounded memory:** configure maximum operations, waiters per operation, and
+  terminal TTL; evict expired terminal records first. If capacity remains full,
+  omit correlation with a diagnostic rather than affecting the operation.
+* **Idempotency:** the existing operation idempotency key controls operation
+  deduplication; waiter idempotency is the pair above. Never use request ID alone.
+* **Ordering:** export links by `(attach_sequence, trace_id)` and operation stages
+  by the trace canonical order, never hash/set iteration order.
+
+## 11. Migration map (no silent renames)
+
+| Current representation | v1 projection |
+|---|---|
+| `Request_ID` / `ProxyTask.request_id` | `TraceContext.legacy_request_id`; generate opaque request/trace IDs. |
+| `*_start_ms`, `*_end_ms` | UTC `started_at`/`ended_at` observations plus separately measured local `duration_ns`. |
+| `predict_*` | `TraceMeasurement(kind="predicted")`; never overwrite. |
+| overwritten `predict_prepare_ms` / `predict_know_prepare_ms` | Preserve original dictionary; adapter emits the final value as `legacy_projected` because original kind is unknowable, and emits unambiguous `actual_*` separately when present. |
+| `actual_compute_ms` | Legacy projection of Proxy-observed Instance round trip to first chunk, not compute. |
+| `actual_vllm_internal_ms` | Actual at Proxy boundary with name `proxy_forward_to_first_chunk`; not vLLM-internal. |
+| `first_token_ms` | `first_token` only when token parsing proves it; otherwise `gateway_request`/forward first-chunk observation. |
+| `decode_start_ms/end_ms` | Legacy-projected decode approximation until vLLM events exist. |
+| `text_actual_path`, `kvcache_actual_path` | Explicit stage outcome + error/fallback code; retain strings in legacy dictionary. |
+| `error` / `first_token_missing_reason` | Sanitized stable `error_code` and optional safe message; retain original only in existing meta. |
+| integer boolean flags | Typed booleans in stage fields. |
+| `kv_ack.network_*` | Measurement with declared producer/boundary and kind; otherwise legacy-projected. |
+| queue snapshots / Prometheus deltas | Endpoint-level `observed`, or `inferred` only under documented isolated-test assumptions. |
+
+## 12. Rollout and exact file plan
+
+### Phase 1 — dependency-light vertical slice
+
+Add:
+
+* `cacheroute_observability/__init__.py` — public, dependency-light exports.
+* `cacheroute_observability/enums.py` — stage, kind, outcome, component vocabularies and
+  stable rank table.
+* `cacheroute_observability/models.py` — frozen schemas and validators.
+* `cacheroute_observability/clock.py` — clock protocol, production clock, test seam.
+* `cacheroute_observability/collector.py` — process-local append-only collection.
+* `cacheroute_observability/legacy_proxy_trace.py` — pure Legacy projection.
+* `test/observability/` — CPU-only contract, collector, isolation, adapter, and demo tests.
+* `scripts/demo_observability_v1.py` — deterministic executable demonstration.
+* `cacheroute_observability/README.md` — maintained package documentation.
+
+Do not modify production component files. This is the independently reviewable
+first PR: contracts, collector, Legacy adapter, deterministic CPU-only tests, and
+package documentation. It includes no production instrumentation, propagation,
+Gateway I/O, or external tracing backend.
+
+### Phase 2 — production compatibility integration
+
+Modify `proxy/proxy.py` at `build_cacheroute_meta` to optionally append
+`request_trace_v1`, and possibly correct `ProxyTask.trace`'s annotation to
+`Dict[str, Any]` without changing values. Retain the dictionary and meta shape.
+
+### Phase 3 — component instrumentation/propagation
+
+Add internal-envelope helpers and modify
+`scheduler/scheduler.py`, `core/fwd.py`, `proxy/proxy.py`,
+`proxy/queue/task.py`, `proxy/queue/manager.py`, `proxy/queue/instance_queues.py`,
+`proxy/queue/knowledge.py`, KDN cache-service facade call sites,
+`kdn_server/gateway/base.py`, `instance/instance_api.py`,
+`instance/control_plane.py`, and `instance/kv_service.py`. Add component and Mock
+Gateway timeline tests. Changes remain observational and additive.
+
+### Phase 4 — optional runtime observations
+
+Add optional adapter modules under `instance/observability/` (public event or
+labelled-metric adapters only) and validator reporting changes. No core model
+imports an LMCache/vLLM client. Keep unlabelled Prometheus delta classification
+as endpoint aggregate/isolated inference.
+
+## 13. CPU-only test matrix
+
+| Test | Assertions | External requirements |
+|---|---|---|
+| Serialization | Exact enum strings, UTC JSON, ns integers, deterministic stage/measurement order, round trip | None |
+| Frozen export | Exported context/stage/trace cannot mutate; collector append creates a new snapshot | None |
+| Clock | Injected monotonic duration, zero allowed, negative rejected, no epoch-derived duration | None |
+| Value kinds | Same semantic metric can coexist as predicted/observed/actual/inferred; no overwrite | None |
+| Provenance | Runtime/Profile/endpoint pair and positive generation validation; Legacy generation zero only | None |
+| Freshness | valid expiry; stale observation retained with stale outcome; bad ordering rejected | None |
+| Outcomes | unsupported/partial/failed/skipped/cancelled/fallback serialization and safe errors | None |
+| Shared task | multiple waiters, duplicates, late registration, cancellation, expiry, bounded capacity, stable order | None |
+| Legacy projection | Every current trace key maps or remains allow-listed legacy; overwritten predictions marked ambiguous projection | None |
+| Free-form compatibility | Current mixed int/float/string/None dictionary is preserved byte-for-value in existing meta | None |
+| Disabled/sampled | disabled allocates no stages; unsampled propagates only context; neither changes callback results/order | None |
+| Security | reject secrets, auth headers, raw tokens/prompts, Redis keys, file paths, tensors, KV bytes/pointers, extras | None |
+| Mock Gateway timeline | request, async operation, observation, shared waiter, Instance load, completion combine deterministically | In-memory Mock Gateway only |
+| Import isolation | importing `cacheroute_observability` does not load FastAPI/httpx/redis/vllm/lmcache/torch/CUDA modules | None |
+| Environment independence | full suite runs with network disabled and without Redis, LMCache, vLLM, GPU, or tracing backend | None |
+
+Validation for the first PR should run from the repository root with
+`python3 -m pytest -q test/observability` and an import-isolation subprocess test.
+Failure is any nondeterministic order, mutable exported object, negative duration
+acceptance, kind collapse, forbidden physical/secret field acceptance, external
+service access, or behavior difference when tracing is disabled.
+
+## 14. Maintainer decisions and unresolved questions
+
+1. Is `cacheroute_observability/` accepted, or should #139/#140 common primitives first
+   move inward to avoid `core -> kdn_server.domain/contracts` imports?
+2. Is the canonical request ID generated at Client when supplied or always at
+   Scheduler? What retry semantics should preserve it?
+3. Should `correlation_id` default to `request_id`, or be mandatory and distinct?
+4. Are `incompatible` and `idempotency_conflict` first-class stage outcomes, or
+   should the trace retain the smaller stage vocabulary plus #140 `outcome_code`?
+5. Which measurement names warrant a closed v1 allow-list in the first PR?
+6. May safe `source_endpoint` contain normalized route templates such as
+   `/v1/chat/completions`, or only logical endpoint IDs? Arbitrary method names
+   must not become stage names either way.
+7. What are the default sampling rate, terminal operation TTL, maximum tracked
+   operations, and maximum waiters per operation?
+8. Should an unsampled request export a minimal `RequestTrace`, or only preserve
+   context in existing `cacheroute_meta`?
+9. Can upstream streaming adapters reliably distinguish first decoded token from
+   role/metadata chunks across supported vLLM versions? Until confirmed, keep the
+   existing marker legacy-projected.
+10. Which `kv_ack` network durations are measured with monotonic clocks at their
+    producer, and which are estimates? They must remain Legacy/ambiguous until
+    provenance is confirmed.
+11. Is preserving raw free-form `error` in legacy meta acceptable during the
+    transition, given that v1 exports will sanitize it?
+
+These decisions do not block a contract-only first PR if the initial schema
+marks disputed fields optional and the PR contains no production instrumentation.
diff --git a/scripts/demo_observability_v1.py b/scripts/demo_observability_v1.py
new file mode 100644
index 0000000..087de6b
--- /dev/null
+++ b/scripts/demo_observability_v1.py
@@ -0,0 +1,100 @@
+#!/usr/bin/env python3
+"""Deterministic, CPU-only demonstration of CacheRoute observability v1."""
+from __future__ import annotations
+
+import json
+import sys
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from cacheroute_observability import (  # noqa: E402
+    CacheOperationTrace, ManualTraceClock, OperationWaiterLink, OperationWaiterState,
+    TraceCollector, TraceComponent, TraceContext, TraceMeasurement, TraceProvenance,
+    TraceStageName, TraceStageOutcome, TraceValueKind, project_legacy_proxy_trace,
+)
+
+
+def main() -> int:
+    clock = ManualTraceClock(datetime(2026, 1, 2, 3, 4, 5, tzinfo=timezone.utc), 1_000)
+    context = TraceContext(
+        trace_id="trace_00000000000000000000000000000001", request_id="req-demo-1",
+        correlation_id="corr-demo", legacy_request_id=42, sampled=True,
+        created_at=clock.utc_now(), expires_at=clock.utc_now() + timedelta(minutes=5),
+    )
+    provenance = TraceProvenance(source_component=TraceComponent.TEST,
+                                 runtime_profile="test.mock", captured_at=clock.utc_now())
+    stage_ids = iter(f"stage_{number:032x}" for number in range(1, 20))
+    collector = TraceCollector(context, clock=clock, id_factory=lambda: next(stage_ids))
+
+    ordered = [
+        (TraceStageName.RUNTIME_PROFILE_RESOLUTION, TraceStageOutcome.SUCCESS),
+        (TraceStageName.GATEWAY_REQUEST, TraceStageOutcome.UNSUPPORTED),
+        (TraceStageName.FALLBACK, TraceStageOutcome.TEXT_FALLBACK),
+        (TraceStageName.GATEWAY_REQUEST, TraceStageOutcome.SUCCESS),
+        (TraceStageName.PROXY_PREPARE_QUEUE, TraceStageOutcome.SUCCESS),
+        (TraceStageName.PROXY_READY_QUEUE, TraceStageOutcome.SUCCESS),
+        (TraceStageName.FIRST_TOKEN, TraceStageOutcome.SUCCESS),
+        (TraceStageName.DECODE, TraceStageOutcome.SUCCESS),
+        (TraceStageName.COMPLETION, TraceStageOutcome.SUCCESS),
+    ]
+    for index, (name, outcome) in enumerate(ordered):
+        stage_id = collector.start_stage(name, provenance)
+        if name is TraceStageName.PROXY_READY_QUEUE:
+            collector.append_measurement(stage_id, TraceMeasurement(
+                name="queue_wait", kind=TraceValueKind.PREDICTED, provenance=provenance,
+                duration_ns=2_000_000,
+            ))
+            collector.append_measurement(stage_id, TraceMeasurement(
+                name="queue_wait", kind=TraceValueKind.ACTUAL, provenance=provenance,
+                duration_ns=3_000_000,
+            ))
+        clock.advance_ns(100 + index)
+        clock.advance_time(timedelta(microseconds=1))
+        collector.finish_stage(stage_id, outcome)
+
+    request_trace = collector.export(complete=True)
+    operation = CacheOperationTrace(
+        operation_id="cacheop-demo", trace_id=context.trace_id, operation="prefetch",
+        state="succeeded", stages=request_trace.stages[1:4], provenance=provenance,
+        artifact_id="artifact-demo", endpoint_id="endpoint-demo", endpoint_generation=1,
+        created_at=context.created_at, updated_at=clock.utc_now(), completed_at=clock.utc_now(),
+    )
+    waiters = tuple(OperationWaiterLink(
+        operation_id=operation.operation_id, trace_id=f"trace_{number:032x}",
+        request_id=f"req-waiter-{number}", correlation_id="corr-demo",
+        attach_sequence=number - 2, attached_at=clock.utc_now(),
+        state=OperationWaiterState.COMPLETED,
+    ) for number in (2, 3))
+    legacy = project_legacy_proxy_trace(
+        request_id="req-legacy", correlation_id="corr-legacy", legacy_request_id=7,
+        trace={
+            "predict_prepare_ms": 12, "actual_vllm_internal_ms": 8,
+            "first_token_ms": 1760000000000, "kvcache_actual_path": "kv_inject_failed_fallback_text",
+            "error": "raw exception must be omitted", "unknown_secretish_result": "omitted",
+        },
+        kv_ack={"payload_bytes": 1024, "raw": "omitted"}, exported_at=clock.utc_now(),
+        runtime_profile="legacy",
+    )
+
+    output = {
+        "request_trace": request_trace.model_dump(mode="json"),
+        "stage_sequence": [stage.name.value for stage in request_trace.stages],
+        "value_kinds": [
+            measurement.kind.value for stage in request_trace.stages
+            for measurement in stage.measurements
+        ],
+        "cache_operation": operation.model_dump(mode="json"),
+        "shared_operation_waiter_ids": [waiter.request_id for waiter in waiters],
+        "legacy_projection": legacy.model_dump(mode="json"),
+    }
+    print(json.dumps(output, indent=2, sort_keys=True))
+    print("observability v1 demo: passed")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/test/observability/conftest.py b/test/observability/conftest.py
new file mode 100644
index 0000000..836cf50
--- /dev/null
+++ b/test/observability/conftest.py
@@ -0,0 +1,22 @@
+from datetime import datetime, timezone
+
+import pytest
+
+from cacheroute_observability import TraceComponent, TraceContext, TraceProvenance
+
+
+@pytest.fixture
+def now():
+    return datetime(2026, 1, 1, tzinfo=timezone.utc)
+
+
+@pytest.fixture
+def context(now):
+    return TraceContext(trace_id="trace_" + "1" * 32, request_id="req-1",
+                        correlation_id="corr-1", sampled=True, created_at=now)
+
+
+@pytest.fixture
+def provenance(now):
+    return TraceProvenance(source_component=TraceComponent.TEST,
+                           runtime_profile="test.mock", captured_at=now)
diff --git a/test/observability/test_clock.py b/test/observability/test_clock.py
new file mode 100644
index 0000000..bd5937f
--- /dev/null
+++ b/test/observability/test_clock.py
@@ -0,0 +1,24 @@
+from datetime import timedelta
+
+import pytest
+
+from cacheroute_observability import ManualTraceClock, SystemTraceClock
+
+
+def test_manual_clock_advances_independently(now):
+    clock = ManualTraceClock(now, 10)
+    clock.advance_ns(0)
+    assert clock.monotonic_ns() == 10
+    clock.advance_ns(5)
+    clock.advance_time(timedelta(seconds=2))
+    assert clock.monotonic_ns() == 15
+    assert clock.utc_now() == now + timedelta(seconds=2)
+
+
+def test_clock_rejects_invalid_values(now):
+    with pytest.raises(ValueError):
+        ManualTraceClock(now.replace(tzinfo=None))
+    clock = ManualTraceClock(now)
+    with pytest.raises(ValueError):
+        clock.advance_ns(-1)
+    assert SystemTraceClock().monotonic_ns() >= 0
diff --git a/test/observability/test_collector.py b/test/observability/test_collector.py
new file mode 100644
index 0000000..1ccd218
--- /dev/null
+++ b/test/observability/test_collector.py
@@ -0,0 +1,72 @@
+from datetime import timedelta
+
+import pytest
+from pydantic import ValidationError
+
+from cacheroute_observability import (
+    ManualTraceClock, TraceCollector, TraceMeasurement, TraceStageName,
+)
+
+
+def _ids():
+    return iter(f"stage_{i:032x}" for i in range(1, 20))
+
+
+def test_order_repetition_monotonic_and_snapshot(context, provenance, now):
+    clock = ManualTraceClock(now, 100)
+    ids = _ids()
+    collector = TraceCollector(context, clock=clock, id_factory=lambda: next(ids))
+    names = ["gateway_request", "fallback", "gateway_request", "completion"]
+    first_snapshot = None
+    for index, name in enumerate(names):
+        stage_id = collector.start_stage(name, provenance)
+        if index == 0:
+            collector.append_measurement(stage_id, TraceMeasurement(
+                name="latency", kind="predicted", provenance=provenance, duration_ns=4))
+            collector.append_measurement(stage_id, TraceMeasurement(
+                name="latency", kind="actual", provenance=provenance, duration_ns=5))
+        clock.advance_ns(index)
+        clock.advance_time(timedelta(seconds=20 if index == 0 else 0))
+        collector.finish_stage(stage_id, "text_fallback" if name == "fallback" else "success")
+        if index == 0:
+            first_snapshot = collector.export()
+    final = collector.export(complete=True)
+    assert [stage.name.value for stage in final.stages] == names
+    assert [stage.duration_ns for stage in final.stages] == [0, 1, 2, 3]
+    assert len(first_snapshot.stages) == 1
+    assert [m.kind.value for m in final.stages[0].measurements] == ["predicted", "actual"]
+    with pytest.raises(ValidationError):
+        final.stages += ()
+
+
+def test_finish_errors(context, provenance, now):
+    clock = ManualTraceClock(now, 10)
+    ids = _ids()
+    collector = TraceCollector(context, clock=clock, id_factory=lambda: next(ids))
+    with pytest.raises(KeyError):
+        collector.finish_stage("stage_" + "9" * 32, "success")
+    stage_id = collector.start_stage(TraceStageName.COMPLETION, provenance)
+    collector.finish_stage(stage_id, "success")
+    with pytest.raises(ValueError):
+        collector.finish_stage(stage_id, "success")
+
+
+def test_disabled_and_unsampled(context, provenance, now):
+    disabled = TraceCollector(context, enabled=False)
+    assert disabled.start_stage("completion", provenance) is None
+    assert disabled.export().stages == ()
+    unsampled_context = context.model_copy(update={"sampled": False})
+    unsampled = TraceCollector(unsampled_context)
+    assert unsampled.start_stage("completion", provenance) is None
+    assert unsampled.export().context.sampled is False
+    assert unsampled.safely(lambda: 1 / 0, "unchanged") == "unchanged"
+
+
+def test_negative_monotonic_duration_rejected(context, provenance, now):
+    clock = ManualTraceClock(now, 10)
+    ids = _ids()
+    collector = TraceCollector(context, clock=clock, id_factory=lambda: next(ids))
+    stage_id = collector.start_stage("completion", provenance)
+    clock._monotonic_value = 9
+    with pytest.raises(ValueError):
+        collector.finish_stage(stage_id, "success")
diff --git a/test/observability/test_demo.py b/test/observability/test_demo.py
new file mode 100644
index 0000000..82f534f
--- /dev/null
+++ b/test/observability/test_demo.py
@@ -0,0 +1,23 @@
+import json
+import subprocess
+import sys
+from pathlib import Path
+
+
+def test_demo_succeeds_and_shows_required_data():
+    root = Path(__file__).resolve().parents[2]
+    result = subprocess.run([sys.executable, str(root / "scripts/demo_observability_v1.py")],
+                            cwd=root, check=True, text=True, capture_output=True)
+    lines = result.stdout.splitlines()
+    assert lines[-1] == "observability v1 demo: passed"
+    payload = json.loads("\n".join(lines[:-1]))
+    assert payload["request_trace"]["schema_version"] == "cacheroute.trace.v1"
+    assert payload["request_trace"]["context"]["schema_version"] == "cacheroute.trace-context.v1"
+    assert payload["stage_sequence"][:4] == [
+        "runtime_profile_resolution", "gateway_request", "fallback", "gateway_request",
+    ]
+    assert payload["value_kinds"] == ["predicted", "actual"]
+    assert payload["shared_operation_waiter_ids"] == ["req-waiter-2", "req-waiter-3"]
+    legacy_json = json.dumps(payload["legacy_projection"])
+    assert "raw exception" not in legacy_json
+    assert "unknown_secretish_result" not in legacy_json
diff --git a/test/observability/test_enums.py b/test/observability/test_enums.py
new file mode 100644
index 0000000..5858433
--- /dev/null
+++ b/test/observability/test_enums.py
@@ -0,0 +1,29 @@
+from cacheroute_observability import (
+    TraceComponent, TraceStageName, TraceStageOutcome, TraceStageState, TraceValueKind,
+)
+
+
+def test_exact_enum_wire_values():
+    assert [x.value for x in TraceComponent] == [
+        "client", "scheduler", "proxy", "kdn", "gateway", "instance", "vllm",
+        "lmcache", "legacy_adapter", "test",
+    ]
+    assert [x.value for x in TraceValueKind] == [
+        "predicted", "observed", "actual", "inferred", "legacy_projected",
+    ]
+    assert [x.value for x in TraceStageState] == ["pending", "running", "completed"]
+    assert [x.value for x in TraceStageOutcome] == [
+        "success", "unsupported", "incompatible", "stale", "partial", "failed",
+        "cancelled", "skipped", "text_fallback", "idempotency_conflict",
+    ]
+    assert [x.value for x in TraceStageName] == [
+        "runtime_profile_resolution", "knowledge_lookup", "semantic_resolution",
+        "artifact_compatibility", "capability_snapshot_discovery", "token_lookup",
+        "artifact_lookup", "cache_observation", "cache_operation_queue",
+        "cache_prefetch_execution", "cache_pin_execution", "cache_unpin_execution",
+        "cache_clear_execution", "cache_rebuild_execution", "gateway_request",
+        "gateway_async_operation", "tier_adapter_observation", "instance_lmcache_load",
+        "proxy_prepare_queue", "proxy_ready_queue", "vllm_prefill", "first_token",
+        "decode", "completion", "fallback", "legacy_scan", "legacy_dump",
+        "legacy_restore", "legacy_inject",
+    ]
diff --git a/test/observability/test_import_isolation.py b/test/observability/test_import_isolation.py
new file mode 100644
index 0000000..bf57a36
--- /dev/null
+++ b/test/observability/test_import_isolation.py
@@ -0,0 +1,18 @@
+import subprocess
+import sys
+
+
+def test_import_isolation_in_fresh_process():
+    code = r'''
+import sys
+import cacheroute_observability
+forbidden = {"fastapi", "httpx", "aiohttp", "redis", "numpy", "torch", "vllm",
+             "lmcache", "core", "scheduler", "proxy", "instance", "kdn_server"}
+loaded = {name.split(".", 1)[0] for name in sys.modules}
+unexpected = forbidden & loaded
+assert not unexpected, sorted(unexpected)
+print("observability import isolation: passed")
+'''
+    result = subprocess.run([sys.executable, "-c", code], check=True, text=True,
+                            capture_output=True)
+    assert result.stdout.strip() == "observability import isolation: passed"
diff --git a/test/observability/test_legacy_proxy_trace.py b/test/observability/test_legacy_proxy_trace.py
new file mode 100644
index 0000000..118dd0f
--- /dev/null
+++ b/test/observability/test_legacy_proxy_trace.py
@@ -0,0 +1,48 @@
+from cacheroute_observability import TraceStageName, TraceValueKind, project_legacy_proxy_trace
+
+
+def _measurements(result):
+    return {m.legacy_name: m for stage in result.stages for m in stage.measurements}
+
+
+def test_legacy_classification_omission_and_sanitization(now):
+    result = project_legacy_proxy_trace(
+        request_id="req", correlation_id="corr", legacy_request_id=9,
+        trace={
+            "predict_total_ms": 10,
+            "predict_prepare_ms": 11,
+            "predict_know_prepare_ms": 12,
+            "actual_vllm_internal_ms": 13,
+            "actual_compute_ms": 14,
+            "first_token_ms": 1760000000000,
+            "decode_start_ms": 1760000000001,
+            "decode_end_ms": 1760000000002,
+            "kvcache_actual_path": "kv_inject_failed_fallback_text",
+            "error": "Traceback secret raw exception",
+            "unknown_key": {"anything": "must not copy"},
+        },
+        kv_ack={"payload_bytes": 42, "password": "must not copy"},
+        exported_at=now, runtime_profile="legacy",
+    )
+    measurements = _measurements(result)
+    assert measurements["predict_total_ms"].kind is TraceValueKind.PREDICTED
+    assert measurements["predict_prepare_ms"].kind is TraceValueKind.LEGACY_PROJECTED
+    assert measurements["predict_know_prepare_ms"].kind is TraceValueKind.LEGACY_PROJECTED
+    assert measurements["actual_vllm_internal_ms"].name == "proxy_forward_to_first_chunk_duration"
+    assert measurements["actual_vllm_internal_ms"].kind is TraceValueKind.ACTUAL
+    assert measurements["actual_compute_ms"].kind is TraceValueKind.LEGACY_PROJECTED
+    assert measurements["first_token_ms"].kind is TraceValueKind.LEGACY_PROJECTED
+    assert measurements["decode_start_ms"].kind is TraceValueKind.LEGACY_PROJECTED
+    assert "unknown_key" not in result.model_dump_json()
+    assert "Traceback" not in result.model_dump_json()
+    assert "password" not in result.model_dump_json()
+    fallback = [stage for stage in result.stages if stage.name is TraceStageName.FALLBACK]
+    assert fallback[0].outcome.value == "text_fallback"
+    assert [stage.sequence for stage in result.stages] == list(range(len(result.stages)))
+
+
+def test_projection_is_deterministic(now):
+    kwargs = dict(request_id="req", correlation_id="corr", legacy_request_id=None,
+                  trace={"actual_total_ms": 2, "unknown": 3}, exported_at=now,
+                  runtime_profile="legacy")
+    assert project_legacy_proxy_trace(**kwargs) == project_legacy_proxy_trace(**kwargs)
diff --git a/test/observability/test_models.py b/test/observability/test_models.py
new file mode 100644
index 0000000..98e37f3
--- /dev/null
+++ b/test/observability/test_models.py
@@ -0,0 +1,93 @@
+from datetime import timedelta
+
+import pytest
+from pydantic import ValidationError
+
+from cacheroute_observability import (
+    CacheOperationTrace, OperationWaiterLink, OperationWaiterState, RequestTrace,
+    TraceComponent, TraceContext, TraceMeasurement, TraceProvenance, TraceStage,
+    TraceStageOutcome, TraceValueKind,
+)
+
+
+def test_context_is_frozen_utc_and_round_trips(context):
+    with pytest.raises(ValidationError):
+        context.request_id = "changed"
+    encoded = context.model_dump_json()
+    assert '"created_at":"2026-01-01T00:00:00Z"' in encoded
+    assert TraceContext.model_validate_json(encoded) == context
+
+
+def test_context_validation(now):
+    with pytest.raises(ValidationError):
+        TraceContext(trace_id="bad", request_id="r", correlation_id="c", created_at=now)
+    with pytest.raises(ValidationError):
+        TraceContext(trace_id="trace_" + "0" * 32, request_id="r", correlation_id="c",
+                     created_at=now.replace(tzinfo=None))
+    with pytest.raises(ValidationError):
+        TraceContext(trace_id="trace_" + "0" * 32, request_id="r", correlation_id="c",
+                     created_at=now, expires_at=now)
+
+
+def test_endpoint_generation_and_freshness(now):
+    current = TraceProvenance(source_component="gateway", runtime_profile="v1",
+                              captured_at=now, endpoint_id="endpoint-1", endpoint_generation=1,
+                              fresh_until=now + timedelta(seconds=1))
+    assert current.endpoint_generation == 1
+    legacy = TraceProvenance(source_component="legacy_adapter", runtime_profile="legacy",
+                             captured_at=now, endpoint_id="endpoint-old", endpoint_generation=0,
+                             legacy=True)
+    assert legacy.endpoint_generation == 0
+    for values in (
+        {"endpoint_id": "endpoint-1"},
+        {"endpoint_id": "endpoint-1", "endpoint_generation": 0},
+        {"fresh_until": now - timedelta(microseconds=1)},
+    ):
+        with pytest.raises(ValidationError):
+            TraceProvenance(source_component="gateway", runtime_profile="v1", captured_at=now, **values)
+
+
+def test_measurement_exact_value_and_ranges(provenance):
+    predicted = TraceMeasurement(name="queue_wait", kind="predicted", provenance=provenance,
+                                 duration_ns=0)
+    actual = TraceMeasurement(name="queue_wait", kind="actual", provenance=provenance,
+                              duration_ns=4)
+    assert (predicted.kind, actual.kind) == (TraceValueKind.PREDICTED, TraceValueKind.ACTUAL)
+    for values in ({}, {"count": 1, "tokens": 2}, {"duration_ns": -1}, {"ratio": float("nan")}, {"ratio": 1.1}):
+        with pytest.raises(ValidationError):
+            TraceMeasurement(name="bad", kind="observed", provenance=provenance, **values)
+
+
+def test_stage_state_outcome_and_time_rules(now, provenance):
+    base = dict(stage_id="stage_" + "2" * 32, sequence=0, name="completion",
+                provenance=provenance, started_at=now)
+    with pytest.raises(ValidationError):
+        TraceStage(**base, state="completed", ended_at=now, duration_ns=0)
+    with pytest.raises(ValidationError):
+        TraceStage(**base, state="running", outcome="success")
+    with pytest.raises(ValidationError):
+        TraceStage(**base, state="completed", outcome="success",
+                   ended_at=now - timedelta(seconds=1), duration_ns=1)
+    complete = TraceStage(**base, state="completed", outcome="success", ended_at=now, duration_ns=0)
+    assert complete.outcome is TraceStageOutcome.SUCCESS
+
+
+def test_request_and_shared_operation_models(now, context, provenance):
+    stage = TraceStage(stage_id="stage_" + "3" * 32, sequence=0, name="cache_operation_queue",
+                       state="completed", outcome="success", provenance=provenance,
+                       started_at=now, ended_at=now, duration_ns=0)
+    trace = RequestTrace(context=context, stages=(stage,), exported_at=now, complete=True,
+                         source_components=(TraceComponent.TEST,))
+    assert RequestTrace.model_validate_json(trace.model_dump_json()) == trace
+    operation = CacheOperationTrace(
+        operation_id="cacheop-1", trace_id=context.trace_id, operation="prefetch",
+        state="succeeded", stages=(stage,), provenance=provenance, artifact_id="artifact-1",
+        endpoint_id="endpoint-1", endpoint_generation=1, created_at=now, updated_at=now,
+        completed_at=now,
+    )
+    waiters = tuple(OperationWaiterLink(
+        operation_id=operation.operation_id, trace_id="trace_" + str(i) * 32,
+        request_id=f"req-{i}", correlation_id="corr", attach_sequence=i,
+        attached_at=now, state=OperationWaiterState.WAITING,
+    ) for i in (1, 2))
+    assert [waiter.attach_sequence for waiter in waiters] == [1, 2]
diff --git a/test/observability/test_security.py b/test/observability/test_security.py
new file mode 100644
index 0000000..8b0fc35
--- /dev/null
+++ b/test/observability/test_security.py
@@ -0,0 +1,26 @@
+import pytest
+from pydantic import ValidationError
+
+from cacheroute_observability import TraceContext, TraceMeasurement
+
+
+@pytest.mark.parametrize("key", [
+    "password", "authorization", "redis_key", "kv_bytes", "tensor",
+    "device_pointer", "physical_path", "request_body", "raw_exception",
+])
+def test_recursive_secret_and_physical_field_rejection(now, key):
+    data = {
+        "trace_id": "trace_" + "1" * 32, "request_id": "r", "correlation_id": "c",
+        "created_at": now, "nested": {key: "bad"},
+    }
+    with pytest.raises(ValidationError, match="forbidden observability field"):
+        TraceContext.model_validate(data)
+
+
+def test_generic_container_and_extra_rejected(provenance):
+    with pytest.raises(ValidationError):
+        TraceMeasurement(name="unsafe", kind="observed", provenance=provenance,
+                         safe_scalar={"safe": "looking"})
+    with pytest.raises(ValidationError):
+        TraceMeasurement(name="unsafe", kind="observed", provenance=provenance,
+                         safe_scalar="ok", prompt="secret")