diff --git a/kdn_server/README.md b/kdn_server/README.md
index f40b63f..a755143 100644
--- a/kdn_server/README.md
+++ b/kdn_server/README.md
@@ -426,3 +426,22 @@ The batch script is useful for preparing knowledge-intensive workloads used in C
 - The standalone `kv_injector.py` path is mainly for debugging. In the full CacheRoute workflow, KVCache injection should be triggered by the scheduling and KDN matching process.
 - Some paths and model names in the examples should be adjusted according to your local deployment.
 - For full deployment with vLLM, LMCache, Redis, Scheduler, Proxy, and Instance, see the main `README.md`.
+
+## Versioned control-plane contracts
+
+The v0.1.10 versioned control-plane layer supplements the operational KDN
+workflows documented above. It defines storage-neutral domain objects,
+versioned Knowledge and Cache Service messages, explicit
+Runtime/Profile/Endpoint negotiation, immutable capability discovery, a
+CPU-only Mock Gateway, and a read-only Legacy adapter. It does not replace the
+existing KDN HTTP service, CLI, text database, KV construction, injection
+workflow, Redis integration, or LMCache data path. Production LMCache HTTP,
+Coordinator, SDK, Redis, and KV data movement are not implemented by this
+contract layer.
+
+Detailed references:
+
+- [Domain vocabulary](domain/README.md)
+- [Service contracts](contracts/README.md)
+- [Gateway architecture](gateway/README.md)
+- [CPU-only tests and workflows](../test/kdn/README.md)
diff --git a/kdn_server/contracts/README.md b/kdn_server/contracts/README.md
new file mode 100644
index 0000000..ed7b5a5
--- /dev/null
+++ b/kdn_server/contracts/README.md
@@ -0,0 +1,73 @@
+# Versioned KDN service contracts
+
+[Back to the KDN architecture](../README.md) · [Shared domain vocabulary](../domain/README.md) · [Gateway architecture](../gateway/README.md) · [Executable workflows](../../test/kdn/README.md)
+
+The contracts package is the storage-neutral public wire layer. Definitions live in [`common.py`](common.py), [`knowledge.py`](knowledge.py), [`cache_service.py`](cache_service.py), and [`errors.py`](errors.py); the stable import surface is [`__init__.py`](__init__.py).
+
+## Contract versions and evolution
+
+- `KDN_CONTRACT_VERSION = "kdn.v1"`
+- `GATEWAY_CONTRACT_VERSION = "lmcache-gateway.v1"`
+
+Enum wire values and existing field meanings are stable. Unsupported versions are rejected explicitly. Additive evolution should use compatible optional fields when semantics remain unchanged; otherwise introduce a new contract version. Never silently reinterpret a persisted field.
+
+## Common envelope
+
+Every versioned message carries a resolved `runtime_profile`, `request_id`, optional `correlation_id`, and timezone-aware UTC timestamp. Models are frozen Pydantic models with `extra="forbid"`, and validation rejects startup-only `auto`. Unknown fields therefore cannot smuggle backend-specific or sensitive state across the boundary.
+
+Gateway-targeted requests and responses additionally require a non-empty `compatibility_profile_id`, canonical `endpoint_id`, and endpoint generation. The target makes negotiation reproducible:
+
+- Runtime/Profile/Endpoint mismatch → `incompatible`;
+- generation mismatch → `stale`;
+- unknown or unsupported capability → `unsupported`.
+
+Legacy may use generation `0` for unknown; v1 and test/mock require a positive generation.
+
+## Knowledge Service families
+
+[`knowledge.py`](knowledge.py) defines storage-neutral contracts to register, update, and resolve knowledge; list compatible artifacts; query artifact compatibility; and report request outcomes. These are contracts only—this package does not register HTTP routes or prescribe a database.
+
+## Cache Service families
+
+[`cache_service.py`](cache_service.py) defines dedicated request and response families for artifact lookup, cache observation, token lookup, prefetch, pin/unpin, clear, rebuild, operation status and cancellation, endpoint discovery, tier/adapter summaries, and maintenance status.
+
+Dedicated responses validate their expected success payload, target provenance, observation freshness, operation provenance, intent-specific operation type, and terminal cancellation no-op semantics. The outcome vocabulary is `success`, `unsupported`, `incompatible`, `stale`, `partial`, `failed`, `cancelled`, `text_fallback`, and `idempotency_conflict`. Every non-success response carries a matching safe `ContractErrorDetail`.
+
+## Token contracts
+
+`TokenInput` accepts exactly one of an immutable tuple of token IDs or an opaque `TokenReference`. `TokenCoverage.whole_request_hit` represents whole-request hit/miss independently from `covered_ranges`; exact ranges are exposed only when `range_coverage` is explicitly supported. This layer does not implement LMCache hashing, chunk keys, token databases, serialization, or block allocation.
+
+## Security and storage neutrality
+
+Contracts prohibit credentials, API keys, authorization headers, raw Redis keys, KV bytes, tensors, device pointers, physical chunk/block indexes, and private serialized LMCache objects. Structured errors contain stable codes and safe messages—not backend exceptions.
+
+## Minimal examples
+
+```python
+from kdn_server.contracts import LookupArtifactRequest, LookupTokensRequest, TokenInput
+
+TARGET = {
+    "runtime_profile": "test/mock",
+    "compatibility_profile_id": "example-compatible-v1",
+    "endpoint_id": "endpoint_" + "1" * 32,
+    "endpoint_generation": 1,
+}
+artifact_request = LookupArtifactRequest(
+    **TARGET,
+    artifact_id="artifact_" + "2" * 32,
+)
+token_request = LookupTokensRequest(
+    **TARGET,
+    tokens=TokenInput(token_ids=(11, 12, 13)),
+)
+```
+
+Inspect outcomes without parsing backend-specific payloads:
+
+```python
+response = gateway.lookup_tokens(token_request)
+if response.outcome.value == "text_fallback":
+    assert response.error.fallback_eligible
+```
+
+For complete, executable Mock and Legacy sequences, continue to the [CPU-only workflow guide](../../test/kdn/README.md).
diff --git a/kdn_server/contracts/__init__.py b/kdn_server/contracts/__init__.py
new file mode 100644
index 0000000..ab2eccd
--- /dev/null
+++ b/kdn_server/contracts/__init__.py
@@ -0,0 +1,44 @@
+"""Stable public KDN versioned service contract API."""
+from .common import (ContractModel, GATEWAY_CONTRACT_VERSION, GatewayTargetedRequest,
+    KDN_CONTRACT_VERSION, SupportState, TokenInput, TokenReference, VersionedMessage)
+from .errors import ContractError, ContractErrorDetail, OutcomeCode
+from .knowledge import (KnowledgeDescriptor, KnowledgeResponse, ListCompatibleArtifactsRequest,
+    ListCompatibleArtifactsResponse, QueryArtifactCompatibilityRequest,
+    QueryArtifactCompatibilityResponse, RegisterKnowledgeRequest, RegisterKnowledgeResponse,
+    ReportRequestOutcomeRequest, ReportRequestOutcomeResponse, ResolveKnowledgeRequest,
+    ResolveKnowledgeResponse, UpdateKnowledgeRequest, UpdateKnowledgeResponse)
+from .cache_service import (AdapterSummary, ArtifactRequest,
+    CancelOperationRequest, CancelOperationResponse, CapacityUsageObservation,
+    CreateClearIntentRequest, CreateClearIntentResponse, CreatePinIntentRequest,
+    CreatePinIntentResponse, CreatePrefetchIntentRequest, CreatePrefetchIntentResponse,
+    CreateRebuildIntentRequest, CreateRebuildIntentResponse, CreateUnpinIntentRequest,
+    CreateUnpinIntentResponse, GetCacheObservationRequest, GetCacheObservationResponse,
+    GetLMCacheEndpointsRequest, GetLMCacheEndpointsResponse, GetMaintenanceStatusRequest,
+    GetMaintenanceStatusResponse, GetOperationStatusRequest, GetOperationStatusResponse,
+    GetTierAndAdapterSummaryRequest, GetTierAndAdapterSummaryResponse,
+    LookupArtifactRequest, LookupArtifactResponse, LookupTokensRequest, LookupTokensResponse,
+    MaintenanceSummary, OperationIntentRequest, SummaryBase, TierLevel,
+    TierSummary, TokenCoverage)
+
+__all__ = [
+    "KDN_CONTRACT_VERSION", "GATEWAY_CONTRACT_VERSION", "ContractModel", "VersionedMessage",
+    "GatewayTargetedRequest", "SupportState", "TokenReference", "TokenInput", "OutcomeCode",
+    "ContractErrorDetail", "ContractError", "KnowledgeDescriptor",
+    "KnowledgeResponse", "RegisterKnowledgeRequest", "RegisterKnowledgeResponse",
+    "UpdateKnowledgeRequest", "UpdateKnowledgeResponse", "ResolveKnowledgeRequest",
+    "ResolveKnowledgeResponse", "ListCompatibleArtifactsRequest", "ListCompatibleArtifactsResponse",
+    "QueryArtifactCompatibilityRequest", "QueryArtifactCompatibilityResponse",
+    "ReportRequestOutcomeRequest", "ReportRequestOutcomeResponse", "ArtifactRequest",
+    "GetCacheObservationRequest", "LookupArtifactRequest", "LookupTokensRequest",
+    "OperationIntentRequest", "CreatePrefetchIntentRequest", "CreatePinIntentRequest",
+    "CreateUnpinIntentRequest", "CreateClearIntentRequest", "CreateRebuildIntentRequest",
+    "GetOperationStatusRequest", "CancelOperationRequest", "GetLMCacheEndpointsRequest",
+    "GetTierAndAdapterSummaryRequest", "GetMaintenanceStatusRequest", "TokenCoverage",
+    "SummaryBase", "AdapterSummary", "TierLevel", "CapacityUsageObservation", "TierSummary",
+    "MaintenanceSummary", "GetCacheObservationResponse",
+    "LookupArtifactResponse", "LookupTokensResponse",
+    "CreatePrefetchIntentResponse", "CreatePinIntentResponse", "CreateUnpinIntentResponse",
+    "CreateClearIntentResponse", "CreateRebuildIntentResponse", "GetOperationStatusResponse",
+    "CancelOperationResponse", "GetLMCacheEndpointsResponse", "GetTierAndAdapterSummaryResponse",
+    "GetMaintenanceStatusResponse",
+]
diff --git a/kdn_server/contracts/cache_service.py b/kdn_server/contracts/cache_service.py
new file mode 100644
index 0000000..b38aa35
--- /dev/null
+++ b/kdn_server/contracts/cache_service.py
@@ -0,0 +1,302 @@
+"""Versioned Cache Service Facade contracts; no physical cache vocabulary."""
+from __future__ import annotations
+
+from datetime import datetime, timedelta
+from enum import Enum
+from typing import ClassVar
+
+from pydantic import AwareDatetime, Field, field_validator, model_validator
+
+from kdn_server.domain import RuntimeProfile, CacheArtifact, CacheOperationState, CacheOperationTask, CacheReplicaObservation, LMCacheEndpoint, CacheOperationType
+from .common import ContractModel, GatewayTargetedRequest, SupportState, TokenInput, VersionedMessage, utc_now
+from .errors import ContractErrorDetail, OutcomeCode
+
+
+class ArtifactRequest(GatewayTargetedRequest):
+    artifact_id: str = Field(pattern=r"^artifact_[0-9a-f]{32}$")
+
+class GetCacheObservationRequest(ArtifactRequest): pass
+class LookupArtifactRequest(ArtifactRequest): pass
+class LookupTokensRequest(GatewayTargetedRequest): tokens: TokenInput
+
+class OperationIntentRequest(ArtifactRequest):
+    idempotency_key: str = Field(min_length=1)
+
+class CreatePrefetchIntentRequest(OperationIntentRequest): pass
+class CreatePinIntentRequest(OperationIntentRequest): pass
+class CreateUnpinIntentRequest(OperationIntentRequest): pass
+class CreateClearIntentRequest(OperationIntentRequest): pass
+class CreateRebuildIntentRequest(OperationIntentRequest): pass
+class GetOperationStatusRequest(GatewayTargetedRequest): task_id: str = Field(pattern=r"^cacheop_[0-9a-f]{32}$")
+class CancelOperationRequest(GetOperationStatusRequest): pass
+class GetLMCacheEndpointsRequest(VersionedMessage): pass
+class GetTierAndAdapterSummaryRequest(GatewayTargetedRequest): pass
+class GetMaintenanceStatusRequest(GatewayTargetedRequest): pass
+
+
+class TokenCoverage(ContractModel):
+    """Separates whole-request hit state from capability-gated range detail."""
+    whole_request_hit: bool
+    covered_ranges: tuple[tuple[int, int], ...] = ()
+    total_tokens: int = Field(ge=0)
+
+    @model_validator(mode="after")
+    def valid_ranges(self):
+        previous_end = 0
+        for index, (start, end) in enumerate(self.covered_ranges):
+            if start < 0 or start >= end or end > self.total_tokens:
+                raise ValueError("coverage ranges require 0 <= start < end <= total_tokens")
+            if index and start < previous_end:
+                raise ValueError("coverage ranges must be ordered and non-overlapping")
+            previous_end = end
+        return self
+
+
+class SummaryBase(ContractModel):
+    """Common provenance that binds summaries to an endpoint generation."""
+    source: str = Field(min_length=1)
+    observed_at: AwareDatetime = Field(default_factory=utc_now)
+    runtime_profile: RuntimeProfile
+    compatibility_profile_id: str = Field(min_length=1)
+    endpoint_id: str = Field(pattern=r"^endpoint_[0-9a-f]{32}$")
+    endpoint_generation: int = Field(ge=0)
+    support: SupportState
+    partial: bool = False
+
+    @field_validator("observed_at")
+    @classmethod
+    def utc_only(cls, value: datetime):
+        if value.utcoffset() != timedelta(0): raise ValueError("observed_at must use UTC")
+        return value
+
+    @model_validator(mode="after")
+    def data_semantics(self):
+        if self.runtime_profile is not RuntimeProfile.LEGACY and self.endpoint_generation == 0:
+            raise ValueError("unknown generation is Legacy-only")
+        return self
+
+
+class AdapterSummary(SummaryBase):
+    loaded_adapters: tuple[str, ...] = ()
+
+    @field_validator("loaded_adapters")
+    @classmethod
+    def valid_adapters(cls, value):
+        if any(not item.strip() for item in value) or len(set(value)) != len(value):
+            raise ValueError("loaded adapters must be non-empty and unique")
+        return value
+
+    @model_validator(mode="after")
+    def supported_has_data(self):
+        if self.support is not SupportState.SUPPORTED and self.loaded_adapters:
+            raise ValueError("unsupported or unknown summaries cannot carry adapter data")
+        return self
+
+
+class TierLevel(str, Enum):
+    L1 = "l1"
+    L2 = "l2"
+
+
+class CapacityUsageObservation(SummaryBase):
+    tier_name: str = Field(min_length=1)
+    tier_level: TierLevel
+    capacity_bytes: int | None = Field(default=None, ge=0)
+    used_bytes: int | None = Field(default=None, ge=0)
+
+    @model_validator(mode="after")
+    def bounded(self):
+        if self.capacity_bytes is not None and self.used_bytes is not None and self.used_bytes > self.capacity_bytes:
+            raise ValueError("used_bytes cannot exceed capacity_bytes")
+        if self.support is not SupportState.SUPPORTED and (self.capacity_bytes is not None or self.used_bytes is not None):
+            raise ValueError("unsupported or unknown capacity cannot carry measurements")
+        return self
+
+
+class TierSummary(SummaryBase):
+    l1_tiers: tuple[str, ...] = ()
+    l2_tiers: tuple[str, ...] = ()
+    capacity: tuple[CapacityUsageObservation, ...] = ()
+
+    @field_validator("l1_tiers", "l2_tiers")
+    @classmethod
+    def valid_tiers(cls, value):
+        if any(not item.strip() for item in value) or len(set(value)) != len(value):
+            raise ValueError("tier names must be non-empty and unique")
+        return value
+
+    @model_validator(mode="after")
+    def tier_consistency(self):
+        if self.support is not SupportState.SUPPORTED and (self.l1_tiers or self.l2_tiers or self.capacity):
+            raise ValueError("unsupported or unknown summaries cannot carry tier data")
+        known = {(name, TierLevel.L1) for name in self.l1_tiers} | {(name, TierLevel.L2) for name in self.l2_tiers}
+        if any((item.tier_name, item.tier_level) not in known for item in self.capacity):
+            raise ValueError("capacity observations must identify a listed tier and level")
+        for item in self.capacity:
+            if any((item.runtime_profile is not self.runtime_profile,
+                    item.compatibility_profile_id != self.compatibility_profile_id,
+                    item.endpoint_id != self.endpoint_id,
+                    item.endpoint_generation != self.endpoint_generation)):
+                raise ValueError("capacity provenance must match its tier summary")
+        return self
+
+
+class MaintenanceSummary(SummaryBase):
+    active: bool | None = None
+    eviction_observable: bool | None = None
+    detail: str | None = None
+
+    @model_validator(mode="after")
+    def maintenance_semantics(self):
+        if self.support is not SupportState.SUPPORTED and (self.active is not None or self.eviction_observable is not None):
+            raise ValueError("unsupported or unknown maintenance cannot carry observations")
+        return self
+
+
+class CacheServiceResponse(VersionedMessage):
+    """Internal response base enforcing outcome and nested provenance consistency."""
+    compatibility_profile_id: str | None = None
+    endpoint_id: str | None = None
+    endpoint_generation: int | None = Field(default=None, ge=0)
+    outcome: OutcomeCode = OutcomeCode.SUCCESS
+    artifact: CacheArtifact | None = None
+    observation: CacheReplicaObservation | None = None
+    operation: CacheOperationTask | None = None
+    endpoints: tuple[LMCacheEndpoint, ...] | None = None
+    token_coverage: TokenCoverage | None = None
+    adapter_summary: AdapterSummary | None = None
+    tier_summary: TierSummary | None = None
+    maintenance_summary: MaintenanceSummary | None = None
+    error: ContractErrorDetail | None = None
+
+    @model_validator(mode="after")
+    def consistent_outcome(self):
+        # Generic construction must not bypass the targeted response envelope.
+        targeted_payload = any(value is not None for value in (
+            self.artifact, self.observation, self.operation, self.token_coverage,
+            self.adapter_summary, self.tier_summary, self.maintenance_summary))
+        if targeted_payload and any(value is None for value in (
+                self.compatibility_profile_id, self.endpoint_id, self.endpoint_generation)):
+            raise ValueError("targeted payloads require complete target metadata")
+        if self.runtime_profile is not RuntimeProfile.LEGACY and self.endpoint_generation == 0:
+            raise ValueError("endpoint_generation=0 is only valid for Legacy responses")
+        if self.outcome is OutcomeCode.SUCCESS and self.error is not None:
+            raise ValueError("successful responses cannot carry an error")
+        if self.outcome is not OutcomeCode.SUCCESS and (self.error is None or self.error.code is not self.outcome):
+            raise ValueError("non-success responses require a matching error detail")
+        if self.outcome is OutcomeCode.CANCELLED and (self.operation is None or self.operation.state is not CacheOperationState.CANCELLED):
+            raise ValueError("cancelled responses require a cancelled operation")
+        if self.outcome is OutcomeCode.STALE and self.observation is not None and self.observation.is_fresh(at=self.timestamp):
+            raise ValueError("stale responses cannot carry an observation fresh at response timestamp")
+        if self.outcome is OutcomeCode.TEXT_FALLBACK and not self.error.fallback_eligible:
+            raise ValueError("text fallback must be explicitly fallback eligible")
+        for summary in (self.adapter_summary, self.tier_summary, self.maintenance_summary):
+            if summary is not None and any((summary.runtime_profile is not self.runtime_profile,
+                    summary.compatibility_profile_id != self.compatibility_profile_id,
+                    summary.endpoint_id != self.endpoint_id, summary.endpoint_generation != self.endpoint_generation)):
+                raise ValueError("summary provenance must match response envelope")
+        if self.observation is not None and self.observation.endpoint_generation != self.endpoint_generation:
+            raise ValueError("observation generation must match response envelope")
+        if self.artifact is not None and self.compatibility_profile_id is not None and any((
+                self.artifact.runtime_profile is not self.runtime_profile,
+                self.artifact.compatibility_profile_id != self.compatibility_profile_id)):
+            raise ValueError("artifact provenance must match response envelope")
+        if self.observation is not None and self.compatibility_profile_id is not None and any((
+                self.observation.runtime_profile is not self.runtime_profile,
+                self.observation.compatibility_profile_id != self.compatibility_profile_id,
+                self.observation.endpoint_id != self.endpoint_id,
+                self.observation.endpoint_generation != self.endpoint_generation)):
+            raise ValueError("observation provenance must match response envelope")
+        if self.operation is not None and self.compatibility_profile_id is not None and any((
+                self.operation.runtime_profile is not self.runtime_profile,
+                self.operation.compatibility_profile_id != self.compatibility_profile_id,
+                self.operation.endpoint_id != self.endpoint_id,
+                self.operation.endpoint_generation != self.endpoint_generation)):
+            raise ValueError("operation provenance must match response envelope")
+        if self.endpoints is not None and any(endpoint.runtime_profile is not self.runtime_profile for endpoint in self.endpoints):
+            raise ValueError("endpoint runtime profile must match response envelope")
+        return self
+
+
+class GatewayTargetedResponse(CacheServiceResponse):
+    compatibility_profile_id: str = Field(min_length=1)
+    endpoint_id: str = Field(pattern=r"^endpoint_[0-9a-f]{32}$")
+    endpoint_generation: int = Field(ge=0)
+
+
+class GetCacheObservationResponse(GatewayTargetedResponse):
+    """Observation result whose successful payload is fresh at response time."""
+    @model_validator(mode="after")
+    def success_payload(self):
+        if self.outcome is OutcomeCode.SUCCESS and (
+                self.observation is None or not self.observation.is_fresh(at=self.timestamp)):
+            raise ValueError("successful observation response requires a fresh observation")
+        return self
+
+class LookupArtifactResponse(GatewayTargetedResponse):
+    @model_validator(mode="after")
+    def success_payload(self):
+        if self.outcome is OutcomeCode.SUCCESS and self.artifact is None: raise ValueError("successful artifact response requires artifact")
+        return self
+
+class LookupTokensResponse(GatewayTargetedResponse):
+    @model_validator(mode="after")
+    def success_payload(self):
+        if self.outcome is OutcomeCode.SUCCESS and self.token_coverage is None: raise ValueError("successful token response requires coverage")
+        return self
+
+class OperationResponse(GatewayTargetedResponse):
+    @model_validator(mode="after")
+    def success_payload(self):
+        if self.outcome is OutcomeCode.SUCCESS and self.operation is None: raise ValueError("successful operation response requires operation")
+        return self
+
+class _TypedOperationResponse(OperationResponse):
+    """Prevents a successful intent response from carrying another operation kind."""
+    expected_operation: ClassVar[CacheOperationType]
+
+    @model_validator(mode="after")
+    def correct_operation(self):
+        if self.outcome is OutcomeCode.SUCCESS and self.operation.operation is not self.expected_operation:
+            raise ValueError(f"successful response requires {self.expected_operation.value} operation")
+        return self
+
+class CreatePrefetchIntentResponse(_TypedOperationResponse): expected_operation = CacheOperationType.PREFETCH
+class CreatePinIntentResponse(_TypedOperationResponse): expected_operation = CacheOperationType.PIN
+class CreateUnpinIntentResponse(_TypedOperationResponse): expected_operation = CacheOperationType.UNPIN
+class CreateClearIntentResponse(_TypedOperationResponse): expected_operation = CacheOperationType.CLEAR
+class CreateRebuildIntentResponse(_TypedOperationResponse): expected_operation = CacheOperationType.REBUILD
+class GetOperationStatusResponse(OperationResponse): pass
+class CancelOperationResponse(OperationResponse):
+    """Allows success only when cancellation is already a terminal no-op."""
+    @model_validator(mode="after")
+    def cancellation_state(self):
+        if self.outcome is OutcomeCode.SUCCESS and not self.operation.terminal:
+            raise ValueError("successful cancellation no-op requires an already-terminal operation")
+        return self
+
+class GetLMCacheEndpointsResponse(CacheServiceResponse):
+    @model_validator(mode="after")
+    def success_payload(self):
+        if self.outcome is OutcomeCode.SUCCESS and self.endpoints is None: raise ValueError("successful endpoint response requires endpoints")
+        return self
+
+class GetTierAndAdapterSummaryResponse(GatewayTargetedResponse):
+    @model_validator(mode="after")
+    def success_payload(self):
+        if self.outcome is OutcomeCode.SUCCESS and (self.adapter_summary is None or self.tier_summary is None): raise ValueError("successful tier response requires both summaries")
+        return self
+
+class GetMaintenanceStatusResponse(GatewayTargetedResponse):
+    @model_validator(mode="after")
+    def success_payload(self):
+        if self.outcome is OutcomeCode.SUCCESS and self.maintenance_summary is None: raise ValueError("successful maintenance response requires summary")
+        return self
+
+INTENT_OPERATION_TYPES = {
+    CreatePrefetchIntentRequest: CacheOperationType.PREFETCH,
+    CreatePinIntentRequest: CacheOperationType.PIN,
+    CreateUnpinIntentRequest: CacheOperationType.UNPIN,
+    CreateClearIntentRequest: CacheOperationType.CLEAR,
+    CreateRebuildIntentRequest: CacheOperationType.REBUILD,
+}
diff --git a/kdn_server/contracts/common.py b/kdn_server/contracts/common.py
new file mode 100644
index 0000000..ef7985c
--- /dev/null
+++ b/kdn_server/contracts/common.py
@@ -0,0 +1,107 @@
+"""Common, versioned wire vocabulary for KDN service contracts."""
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+from enum import Enum
+from typing import Any
+from uuid import uuid4
+
+from pydantic import AwareDatetime, BaseModel, ConfigDict, Field, field_validator, model_validator
+
+from kdn_server.domain import RuntimeProfile
+
+KDN_CONTRACT_VERSION = "kdn.v1"
+GATEWAY_CONTRACT_VERSION = "lmcache-gateway.v1"
+ENDPOINT_ID_PATTERN = r"^endpoint_[0-9a-f]{32}$"
+
+
+class SupportState(str, Enum):
+    """Tri-state capability result; uncertainty is deliberately falsey."""
+    SUPPORTED = "supported"
+    UNSUPPORTED = "unsupported"
+    UNKNOWN = "unknown"
+
+    def __bool__(self):
+        return self is SupportState.SUPPORTED
+
+
+def utc_now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+class ContractModel(BaseModel):
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    def model_copy(self, *, update=None, deep=False):
+        data = self.model_dump(mode="python")
+        data.update(update or {})
+        return type(self).model_validate(data)
+
+
+class VersionedMessage(ContractModel):
+    """Immutable persisted envelope with resolved runtime and UTC identity."""
+    contract_version: str = KDN_CONTRACT_VERSION
+    runtime_profile: RuntimeProfile
+    request_id: str = Field(default_factory=lambda: f"req_{uuid4().hex}", min_length=1)
+    correlation_id: str | None = None
+    timestamp: AwareDatetime = Field(default_factory=utc_now)
+
+    @field_validator("contract_version")
+    @classmethod
+    def supported_version(cls, value: str) -> str:
+        if value != KDN_CONTRACT_VERSION:
+            raise ValueError(f"unsupported contract version: {value}")
+        return value
+
+    @field_validator("runtime_profile", mode="before")
+    @classmethod
+    def resolved_profile(cls, value: Any) -> RuntimeProfile:
+        profile = RuntimeProfile.normalize(value)
+        if profile is RuntimeProfile.AUTO:
+            raise ValueError("runtime_profile 'auto' is startup-only")
+        return profile
+
+    @field_validator("timestamp")
+    @classmethod
+    def utc_timestamp(cls, value: datetime) -> datetime:
+        if value.utcoffset() != timedelta(0):
+            raise ValueError("timestamp must use UTC")
+        return value
+
+
+class GatewayTargetedRequest(VersionedMessage):
+    """Pins work to one negotiated compatibility endpoint incarnation."""
+    compatibility_profile_id: str = Field(min_length=1)
+    endpoint_id: str = Field(pattern=ENDPOINT_ID_PATTERN)
+    endpoint_generation: int = Field(ge=0)
+
+    @model_validator(mode="after")
+    def valid_generation(self):
+        if self.runtime_profile is not RuntimeProfile.LEGACY and self.endpoint_generation == 0:
+            raise ValueError("endpoint_generation=0 is only valid for Legacy targets")
+        return self
+
+
+class TokenReference(ContractModel):
+    reference_id: str = Field(min_length=1)
+    token_count: int | None = Field(default=None, ge=0)
+
+
+class TokenInput(ContractModel):
+    """Storage-neutral token input: inline IDs or one opaque reference, never both."""
+    token_ids: tuple[int, ...] | None = None
+    token_reference: TokenReference | None = None
+
+    @field_validator("token_ids")
+    @classmethod
+    def valid_ids(cls, value):
+        if value is not None and (not value or any(x < 0 for x in value)):
+            raise ValueError("token_ids must be a non-empty tuple of non-negative IDs")
+        return value
+
+    @model_validator(mode="after")
+    def exactly_one(self):
+        # Ambiguous input would let adapters choose different lookup identities.
+        if (self.token_ids is None) == (self.token_reference is None):
+            raise ValueError("provide exactly one of token_ids or token_reference")
+        return self
diff --git a/kdn_server/contracts/errors.py b/kdn_server/contracts/errors.py
new file mode 100644
index 0000000..56764c2
--- /dev/null
+++ b/kdn_server/contracts/errors.py
@@ -0,0 +1,37 @@
+"""Safe structured outcomes shared by KDN facade and gateways."""
+from enum import Enum
+
+from pydantic import Field, field_validator
+
+from .common import ContractModel, KDN_CONTRACT_VERSION
+
+
+class OutcomeCode(str, Enum):
+    SUCCESS = "success"
+    UNSUPPORTED = "unsupported"
+    INCOMPATIBLE = "incompatible"
+    STALE = "stale"
+    PARTIAL = "partial"
+    FAILED = "failed"
+    CANCELLED = "cancelled"
+    TEXT_FALLBACK = "text_fallback"
+    IDEMPOTENCY_CONFLICT = "idempotency_conflict"
+
+
+class ContractErrorDetail(ContractModel):
+    """Stable safe failure detail; backend exceptions never cross the wire."""
+    code: OutcomeCode
+    message: str = Field(min_length=1)
+    contract_version: str = KDN_CONTRACT_VERSION
+    retryable: bool = False
+    fallback_eligible: bool = False
+
+    @field_validator("contract_version")
+    @classmethod
+    def exact_version(cls, value):
+        if value != KDN_CONTRACT_VERSION:
+            raise ValueError("unsupported error contract version")
+        return value
+
+
+ContractError = ContractErrorDetail
diff --git a/kdn_server/contracts/knowledge.py b/kdn_server/contracts/knowledge.py
new file mode 100644
index 0000000..6b1295c
--- /dev/null
+++ b/kdn_server/contracts/knowledge.py
@@ -0,0 +1,51 @@
+"""Storage-neutral Knowledge Service request and response contracts."""
+from typing import Literal
+
+from pydantic import Field, model_validator
+
+from kdn_server.domain import CacheArtifact
+from .common import VersionedMessage
+from .errors import ContractError, OutcomeCode
+
+
+class KnowledgeDescriptor(VersionedMessage):
+    knowledge_id: str = Field(min_length=1)
+    revision: str = Field(default="1", min_length=1)
+    content_reference: str | None = None
+
+class RegisterKnowledgeRequest(KnowledgeDescriptor): pass
+class UpdateKnowledgeRequest(KnowledgeDescriptor): pass
+class ResolveKnowledgeRequest(VersionedMessage): knowledge_id: str = Field(min_length=1)
+class ListCompatibleArtifactsRequest(VersionedMessage): knowledge_id: str = Field(min_length=1)
+class QueryArtifactCompatibilityRequest(VersionedMessage): artifact: CacheArtifact
+class ReportRequestOutcomeRequest(VersionedMessage):
+    knowledge_id: str = Field(min_length=1)
+    outcome: OutcomeCode
+    operation_id: str | None = None
+
+class KnowledgeResponse(VersionedMessage):
+    outcome: OutcomeCode = OutcomeCode.SUCCESS
+    knowledge_id: str | None = None
+    artifact: CacheArtifact | None = None
+    artifacts: tuple[CacheArtifact, ...] = ()
+    compatible: bool | None = None
+    error: ContractError | None = None
+
+    @model_validator(mode="after")
+    def consistent_outcome(self):
+        if self.outcome is OutcomeCode.SUCCESS and self.error is not None:
+            raise ValueError("successful responses cannot carry an error")
+        if self.outcome is not OutcomeCode.SUCCESS and (
+            self.error is None or self.error.code is not self.outcome
+        ):
+            raise ValueError("non-success responses require a matching error detail")
+        if self.outcome is OutcomeCode.TEXT_FALLBACK and not self.error.fallback_eligible:
+            raise ValueError("text fallback must be explicitly fallback eligible")
+        return self
+
+RegisterKnowledgeResponse = KnowledgeResponse
+UpdateKnowledgeResponse = KnowledgeResponse
+ResolveKnowledgeResponse = KnowledgeResponse
+ListCompatibleArtifactsResponse = KnowledgeResponse
+QueryArtifactCompatibilityResponse = KnowledgeResponse
+ReportRequestOutcomeResponse = KnowledgeResponse
diff --git a/kdn_server/domain/README.md b/kdn_server/domain/README.md
new file mode 100644
index 0000000..f2e6f01
--- /dev/null
+++ b/kdn_server/domain/README.md
@@ -0,0 +1,50 @@
+# KDN shared domain vocabulary
+
+[Back to the KDN architecture](../README.md) · [Source: `models.py`](models.py) · [Wire contracts](../contracts/README.md)
+
+The domain package defines immutable logical identities and lifecycle state reused by the Cache Service contracts and Gateway responses. These objects describe control-plane truth; they never contain physical KV payloads.
+
+## RuntimeProfile
+
+`v1`, `legacy`, and `test/mock` are resolved execution profiles. `auto` exists only for startup selection and must be resolved before any model is serialized or persisted. This prevents an ambiguous runtime choice from crossing a process boundary.
+
+## CacheArtifact
+
+A `CacheArtifact` identifies a logical cache materialization. Its canonical `artifact_id` is deterministically derived from knowledge version, model, tokenizer, ordered adapters, cache-data profile, compatibility profile, runtime profile, and schema version. It is an identity for compatibility and lookup—not a physical chunk, block, key, or storage location.
+
+## LMCacheEndpoint
+
+An `LMCacheEndpoint` has a stable logical endpoint identity. Its generation increases when compatibility-relevant endpoint state changes, allowing callers to distinguish current observations from facts about an older incarnation.
+
+## CacheReplicaObservation
+
+A replica observation is a time-bounded logical report, not an assertion that KDN owns a physical replica. It records source, confidence, Runtime Profile, compatibility profile, endpoint identity and generation, observation/projected timestamps, and expiry.
+
+Freshness uses a closed-open interval:
+
+```text
+source_observed_at <= at < expires_at
+```
+
+Legacy projections may use endpoint generation `0` to mean that the generation is unknown. Non-Legacy observations require a positive generation.
+
+## CacheOperationTask
+
+A task represents logical lookup, prefetch, pin, unpin, clear, rebuild, or observation work. The main lifecycle is:
+
+```text
+pending -> running -> succeeded
+                   -> retry_wait -> running
+                   -> failed
+pending/running/retry_wait -> cancelled
+```
+
+Terminal states do not transition further; same-state transitions are idempotent. Retry state is explicit, and v1 write operations cannot target the Legacy Gateway implicitly.
+
+## QueueWork
+
+`QueueWork` supplies queue lifecycle vocabulary—queued, claimed, executing, retry wait, completed, failed, and cancelled—without implementing admission, placement, or scheduling policy. It links queue work to a logical cache task while preserving independent transition validation.
+
+## Reuse by higher layers
+
+[Cache Service responses](../contracts/README.md) return these objects directly where appropriate. [Gateway adapters](../gateway/README.md) validate their Runtime/Profile/Endpoint/Generation provenance before returning success, so callers share one domain vocabulary instead of adapter-specific state shapes.
diff --git a/kdn_server/gateway/README.md b/kdn_server/gateway/README.md
new file mode 100644
index 0000000..fa4b6c6
--- /dev/null
+++ b/kdn_server/gateway/README.md
@@ -0,0 +1,66 @@
+# LMCache Gateway architecture
+
+[Back to the KDN architecture](../README.md) · [Wire contracts](../contracts/README.md) · [Shared domain](../domain/README.md) · [CPU-only workflows](../../test/kdn/README.md)
+
+The Gateway package is the orchestration boundary between stable CacheRoute contracts and transport-specific LMCache capability surfaces. It does not move physical KV payloads.
+
+## Gateway vocabulary
+
+[`profiles.py`](profiles.py) defines adapter kinds: `mp_http_api`, `mp_coordinator`, `mp_sdk`, `mp_metrics_events`, optional `mp_l2_plugin`, `legacy_redis`, `mock`, and `unknown_future`. A transport kind says *how an adapter is reached*; it is separate from `LMCacheCompatibilityProfile`, which says *whether logical cache artifacts and operations are compatible*.
+
+## LMCacheCompatibilityProfile
+
+A compatibility profile may identify LMCache version/build, configuration, key/hash scheme, layout, serde, chunk size, and connector profile. It never contains credentials, authorization data, or connection secrets.
+
+## CapabilitySnapshot
+
+[`capabilities.py`](capabilities.py) defines an immutable startup/discovery snapshot containing the Runtime Profile, ordered composed adapter bindings, compatibility profile, endpoint generation, loaded adapters and tiers, provenance, and tri-state capabilities.
+
+> **`UNKNOWN` is never treated as `SUPPORTED`.**
+
+Capability groups cover logical lookup/observation, range coverage, object listing/deletion, prefetch and operation lifecycle, pin/unpin, tier capacity and maintenance, metrics/events, batching, leases, asynchronous completion, and cancellation. Runtime isolation permits MP/future bindings for v1, `legacy_redis` only for Legacy, and `mock` only for test/mock.
+
+## Negotiation and capability gating
+
+[`GatewayAdapterBase`](base.py) applies the shared flow:
+
+```text
+request
+  -> Runtime/Profile/Endpoint validation
+  -> endpoint-generation validation
+  -> capability gate
+  -> dedicated structured response
+```
+
+An incompatible target returns `INCOMPATIBLE`; an old generation returns `STALE`; an unknown or unsupported capability returns `UNSUPPORTED`. Response construction preserves request and correlation identity.
+
+## LMCacheGateway Protocol
+
+[`protocol.py`](protocol.py) is the stable boundary future adapters implement. It covers capability discovery, artifact and token lookup, cache observation, logical operation submission, status, cancellation, endpoint discovery, tier/adapter summary, and maintenance status. Upper layers depend on this Protocol and dedicated response contracts—not transport routes, module paths, or private response shapes.
+
+## Current adapters
+
+### MockGateway
+
+[`mock.py`](mock.py) is deterministic and CPU-only. It provides fixture-based artifact, observation, token, endpoint, and summary results; independent token/range capabilities; idempotent task identity; asynchronous state simulation; cancellation; and structured stale, incompatible, unsupported, conflict, and fallback outcomes. It performs no network, Redis, filesystem, GPU, vLLM, or LMCache I/O.
+
+### LegacyCacheAdapter
+
+[`legacy.py`](legacy.py) is an explicit read-only boundary. Generation `0` represents unknown Legacy generation. It adds no v1 functionality: Legacy writes are unsupported and v1 calls are incompatible. The adapter performs no real Redis or filesystem I/O.
+
+## Factory
+
+[`factory.py`](factory.py) first verifies that the requested transport appears in the snapshot's adapter bindings. It constructs only Mock and Legacy adapters today. Production transport kinds are representable for negotiation but intentionally raise `NotImplementedError` at construction.
+
+## Adding a production adapter later
+
+1. Implement `LMCacheGateway`.
+2. Keep route names, module paths, and private API differences inside the adapter.
+3. Produce an immutable `CapabilitySnapshot` at startup.
+4. Preserve request and correlation metadata.
+5. Negotiate Runtime/Profile/Endpoint/Generation.
+6. Gate every operation by an explicitly supported capability.
+7. Return dedicated response types.
+8. Never expose credentials or physical KV payloads.
+9. Add CPU-only contract tests and separate integration tests.
+10. Version any change to existing wire values or semantics.
diff --git a/kdn_server/gateway/__init__.py b/kdn_server/gateway/__init__.py
new file mode 100644
index 0000000..82b143c
--- /dev/null
+++ b/kdn_server/gateway/__init__.py
@@ -0,0 +1,11 @@
+"""Stable public LMCache gateway contract API."""
+from .profiles import GatewayAdapterBinding, GatewayTransportKind, LMCacheCompatibilityProfile
+from .capabilities import CapabilitySnapshot, SupportState
+from .protocol import LMCacheGateway
+from .mock import MockGateway
+from .legacy import LegacyCacheAdapter
+from .factory import create_gateway
+
+__all__ = ["GatewayTransportKind", "GatewayAdapterBinding", "LMCacheCompatibilityProfile",
+           "SupportState", "CapabilitySnapshot", "LMCacheGateway", "MockGateway",
+           "LegacyCacheAdapter", "create_gateway"]
diff --git a/kdn_server/gateway/base.py b/kdn_server/gateway/base.py
new file mode 100644
index 0000000..b08eeba
--- /dev/null
+++ b/kdn_server/gateway/base.py
@@ -0,0 +1,54 @@
+"""Shared adapter negotiation and safe response construction."""
+from kdn_server.contracts.cache_service import CacheServiceResponse
+from kdn_server.contracts.errors import ContractErrorDetail, OutcomeCode
+from .capabilities import SupportState
+
+
+class GatewayAdapterBase:
+    capabilities = None
+
+    def discover_capabilities(self):
+        return self.capabilities
+
+    def _response(self, request, *, response_type=CacheServiceResponse, outcome=OutcomeCode.SUCCESS, message=None,
+                  retryable=False, fallback_eligible=False, **values):
+        """Build a safe dedicated response while preserving caller correlation."""
+        metadata = dict(runtime_profile=request.runtime_profile, request_id=request.request_id,
+                        correlation_id=request.correlation_id)
+        for field in ("compatibility_profile_id", "endpoint_id", "endpoint_generation"):
+            if hasattr(request, field): metadata[field] = getattr(request, field)
+        error = None
+        if outcome is not OutcomeCode.SUCCESS:
+            error = ContractErrorDetail(code=outcome, message=message or outcome.value.replace("_", " "),
+                                        retryable=retryable, fallback_eligible=fallback_eligible)
+        return response_type(**metadata, outcome=outcome, error=error, **values)
+
+    def _negotiate(self, request):
+        """Map target disagreement to INCOMPATIBLE and old generation to STALE."""
+        cap = self.capabilities
+        if (request.runtime_profile is not cap.runtime_profile or
+                request.compatibility_profile_id != cap.compatibility_profile.compatibility_profile_id or
+                request.endpoint_id != cap.endpoint_id):
+            return self._response(request, outcome=OutcomeCode.INCOMPATIBLE,
+                                  message="request target is incompatible with gateway capabilities")
+        if request.endpoint_generation != cap.endpoint_generation:
+            return self._response(request, outcome=OutcomeCode.STALE,
+                                  message="request endpoint generation is stale", retryable=True)
+        return None
+
+    def _gate(self, request, capability):
+        """Negotiate first, then map unknown/unsupported capability to UNSUPPORTED."""
+        mismatch = self._negotiate(request)
+        if mismatch is not None: return mismatch
+        if capability is not SupportState.SUPPORTED:
+            return self._response(request, outcome=OutcomeCode.UNSUPPORTED,
+                                  message="gateway capability is unsupported or unknown")
+        return None
+
+    def _discovery_negotiate(self, request, response_type):
+        """Keep untargeted discovery inside its requested Runtime Profile."""
+        if request.runtime_profile is not self.capabilities.runtime_profile:
+            return self._response(request, response_type=response_type,
+                                  outcome=OutcomeCode.INCOMPATIBLE,
+                                  message="discovery runtime profile is incompatible")
+        return None
diff --git a/kdn_server/gateway/capabilities.py b/kdn_server/gateway/capabilities.py
new file mode 100644
index 0000000..528a7f2
--- /dev/null
+++ b/kdn_server/gateway/capabilities.py
@@ -0,0 +1,88 @@
+"""Immutable, provenance-bearing gateway capability discovery."""
+from datetime import datetime, timedelta
+from pydantic import AwareDatetime, Field, field_validator, model_validator
+
+from kdn_server.contracts.common import ContractModel, ENDPOINT_ID_PATTERN, GATEWAY_CONTRACT_VERSION, SupportState, utc_now
+from kdn_server.domain import RuntimeProfile
+from .profiles import GatewayAdapterBinding, GatewayTransportKind, LMCacheCompatibilityProfile
+
+
+class CapabilitySnapshot(ContractModel):
+    """Immutable discovery result for one composed, runtime-isolated endpoint."""
+    contract_version: str = GATEWAY_CONTRACT_VERSION
+    runtime_profile: RuntimeProfile
+    adapter_bindings: tuple[GatewayAdapterBinding, ...]
+    compatibility_profile: LMCacheCompatibilityProfile
+    endpoint_id: str = Field(pattern=ENDPOINT_ID_PATTERN)
+    endpoint_generation: int = Field(ge=0)
+    runtime_mode: str | None = None
+    observed_at: AwareDatetime = Field(default_factory=utc_now)
+    source: str = Field(min_length=1)
+    loaded_adapters: tuple[str, ...] = ()
+    l1_tiers: tuple[str, ...] = ()
+    l2_tiers: tuple[str, ...] = ()
+    token_lookup: SupportState = SupportState.UNKNOWN
+    range_coverage: SupportState = SupportState.UNKNOWN
+    object_listing: SupportState = SupportState.UNKNOWN
+    object_deletion: SupportState = SupportState.UNKNOWN
+    warm_prefetch: SupportState = SupportState.UNKNOWN
+    operation_status: SupportState = SupportState.UNKNOWN
+    pin_unpin: SupportState = SupportState.UNKNOWN
+    tier_capacity_usage: SupportState = SupportState.UNKNOWN
+    maintenance_eviction: SupportState = SupportState.UNKNOWN
+    metrics_events: SupportState = SupportState.UNKNOWN
+    batching: SupportState = SupportState.UNKNOWN
+    lock_lease: SupportState = SupportState.UNKNOWN
+    async_completion: SupportState = SupportState.UNKNOWN
+    cancellation: SupportState = SupportState.UNKNOWN
+    artifact_lookup: SupportState = SupportState.UNKNOWN
+    cache_observation: SupportState = SupportState.UNKNOWN
+
+    @field_validator("contract_version")
+    @classmethod
+    def version(cls, value):
+        if value != GATEWAY_CONTRACT_VERSION: raise ValueError("unsupported gateway contract version")
+        return value
+
+    @field_validator("runtime_profile", mode="before")
+    @classmethod
+    def resolved(cls, value):
+        value = RuntimeProfile.normalize(value)
+        if value is RuntimeProfile.AUTO: raise ValueError("runtime_profile 'auto' is startup-only")
+        return value
+
+    @field_validator("observed_at")
+    @classmethod
+    def utc_only(cls, value: datetime):
+        if value.utcoffset() != timedelta(0): raise ValueError("observed_at must use UTC")
+        return value
+
+    @field_validator("loaded_adapters", "l1_tiers", "l2_tiers")
+    @classmethod
+    def ordered_unique_names(cls, value):
+        if any(not item.strip() for item in value) or len(set(value)) != len(value):
+            raise ValueError("adapter and tier names must be non-empty and unique")
+        return value
+
+    @model_validator(mode="after")
+    def coherent_profile(self):
+        # Composition is ordered, but a transport cannot be bound twice.
+        kinds = tuple(binding.transport_kind for binding in self.adapter_bindings)
+        if not kinds or len(set(kinds)) != len(kinds):
+            raise ValueError("adapter_bindings must be non-empty and unique by transport kind")
+        mp = {GatewayTransportKind.MP_HTTP_API, GatewayTransportKind.MP_COORDINATOR,
+              GatewayTransportKind.MP_SDK, GatewayTransportKind.MP_METRICS_EVENTS,
+              GatewayTransportKind.MP_L2_PLUGIN, GatewayTransportKind.UNKNOWN_FUTURE}
+        allowed = (set(kinds) <= mp if self.runtime_profile is RuntimeProfile.V1 else
+                   set(kinds) == {GatewayTransportKind.LEGACY_REDIS} if self.runtime_profile is RuntimeProfile.LEGACY else
+                   set(kinds) == {GatewayTransportKind.MOCK})
+        if not allowed:
+            # Runtime isolation prevents Mock or Legacy behavior leaking into v1.
+            raise ValueError("adapter bindings are incompatible with runtime profile")
+        if self.runtime_profile is not RuntimeProfile.LEGACY and self.endpoint_generation == 0:
+            raise ValueError("endpoint_generation=0 is only valid for Legacy capabilities")
+        return self
+
+    def supports_adapter(self, kind: GatewayTransportKind | str) -> bool:
+        requested = GatewayTransportKind(kind)
+        return any(item.transport_kind is requested for item in self.adapter_bindings)
diff --git a/kdn_server/gateway/factory.py b/kdn_server/gateway/factory.py
new file mode 100644
index 0000000..8218845
--- /dev/null
+++ b/kdn_server/gateway/factory.py
@@ -0,0 +1,20 @@
+"""Gateway construction boundary; production transports are intentionally absent."""
+from .legacy import LegacyCacheAdapter
+from .mock import MockGateway
+from .profiles import GatewayTransportKind
+
+
+def create_gateway(transport_kind, capabilities, **fixtures):
+    """Construct available adapters only after validating binding membership.
+
+    Production transports remain representable in contracts but intentionally
+    have no construction or I/O implementation in this package.
+    """
+    kind = GatewayTransportKind(transport_kind)
+    if not capabilities.supports_adapter(kind):
+        raise ValueError("requested transport is not present in capability adapter bindings")
+    if kind is GatewayTransportKind.MOCK:
+        return MockGateway(capabilities, **fixtures)
+    if kind is GatewayTransportKind.LEGACY_REDIS:
+        return LegacyCacheAdapter(capabilities, **fixtures)
+    raise NotImplementedError(f"production gateway transport {kind.value!r} is not implemented")
diff --git a/kdn_server/gateway/legacy.py b/kdn_server/gateway/legacy.py
new file mode 100644
index 0000000..1bd6a00
--- /dev/null
+++ b/kdn_server/gateway/legacy.py
@@ -0,0 +1,69 @@
+"""Explicit read-only Legacy gateway boundary (no Redis or filesystem I/O)."""
+from kdn_server.contracts.cache_service import (
+    CancelOperationResponse, CreateClearIntentRequest, CreateClearIntentResponse,
+    CreatePinIntentRequest, CreatePinIntentResponse, CreatePrefetchIntentRequest,
+    CreatePrefetchIntentResponse, CreateRebuildIntentRequest, CreateRebuildIntentResponse,
+    CreateUnpinIntentRequest, CreateUnpinIntentResponse, GetCacheObservationResponse,
+    GetLMCacheEndpointsResponse, GetMaintenanceStatusResponse, GetOperationStatusResponse,
+    GetTierAndAdapterSummaryResponse, LookupArtifactResponse, LookupTokensResponse,
+)
+from kdn_server.contracts.common import utc_now
+from kdn_server.contracts.errors import OutcomeCode
+from .base import GatewayAdapterBase
+
+
+class LegacyCacheAdapter(GatewayAdapterBase):
+    """Read-only Legacy projection with no Redis or filesystem access.
+
+    Legacy generation zero means unknown; shared negotiation rejects v1 calls
+    before this adapter can project or return Legacy state.
+    """
+    read_only = True
+
+    def __init__(self, capabilities, *, artifacts=(), observations=()):
+        self.capabilities = capabilities
+        self.artifacts = {x.artifact_id: x for x in artifacts}
+        self.observations = {x.artifact_id: x for x in observations}
+
+    def lookup_artifact(self, request):
+        if (guard := self._gate(request, self.capabilities.artifact_lookup)): return LookupArtifactResponse.model_validate(guard.model_dump())
+        artifact = self.artifacts.get(request.artifact_id)
+        if artifact is None: return self._response(request, response_type=LookupArtifactResponse, outcome=OutcomeCode.STALE, message="artifact not observed")
+        if artifact.runtime_profile is not request.runtime_profile or artifact.compatibility_profile_id != request.compatibility_profile_id:
+            return self._response(request, response_type=LookupArtifactResponse, outcome=OutcomeCode.INCOMPATIBLE, message="artifact provenance mismatch")
+        return self._response(request, response_type=LookupArtifactResponse, artifact=artifact)
+
+    def get_cache_observation(self, request):
+        response_at = utc_now()
+        if (guard := self._gate(request, self.capabilities.cache_observation)): return GetCacheObservationResponse.model_validate(guard.model_dump())
+        observation = self.observations.get(request.artifact_id)
+        if observation is None: return self._response(request, response_type=GetCacheObservationResponse, outcome=OutcomeCode.STALE, message="Legacy observation is absent")
+        incompatible = any((observation.artifact_id != request.artifact_id,
+            observation.runtime_profile is not request.runtime_profile,
+            observation.compatibility_profile_id != request.compatibility_profile_id,
+            observation.endpoint_id != request.endpoint_id))
+        if incompatible: return self._response(request, response_type=GetCacheObservationResponse, outcome=OutcomeCode.INCOMPATIBLE, message="Legacy observation provenance mismatch")
+        if observation.endpoint_generation != request.endpoint_generation:
+            return self._response(request, response_type=GetCacheObservationResponse, outcome=OutcomeCode.STALE, message="Legacy observation generation mismatch")
+        if not observation.is_fresh(at=response_at): return self._response(request, response_type=GetCacheObservationResponse, timestamp=response_at, outcome=OutcomeCode.STALE, message="Legacy observation is stale", observation=observation)
+        return self._response(request, response_type=GetCacheObservationResponse, timestamp=response_at, observation=observation)
+
+    def _unsupported(self, request, response_type):
+        if (guard := self._negotiate(request)): return response_type.model_validate(guard.model_dump())
+        return self._response(request, response_type=response_type, outcome=OutcomeCode.UNSUPPORTED,
+                              message="Legacy adapter is read-only and does not provide this capability")
+
+    def lookup_tokens(self, request): return self._unsupported(request, LookupTokensResponse)
+    def submit_operation(self, request):
+        response_type = {CreatePrefetchIntentRequest: CreatePrefetchIntentResponse,
+            CreatePinIntentRequest: CreatePinIntentResponse, CreateUnpinIntentRequest: CreateUnpinIntentResponse,
+            CreateClearIntentRequest: CreateClearIntentResponse, CreateRebuildIntentRequest: CreateRebuildIntentResponse}[type(request)]
+        return self._unsupported(request, response_type)
+    def get_operation_status(self, request): return self._unsupported(request, GetOperationStatusResponse)
+    def cancel_operation(self, request): return self._unsupported(request, CancelOperationResponse)
+    def get_tier_adapter_summary(self, request): return self._unsupported(request, GetTierAndAdapterSummaryResponse)
+    def get_maintenance_status(self, request): return self._unsupported(request, GetMaintenanceStatusResponse)
+    def get_endpoints(self, request):
+        if (guard := self._discovery_negotiate(request, GetLMCacheEndpointsResponse)): return guard
+        return self._response(request, response_type=GetLMCacheEndpointsResponse, outcome=OutcomeCode.UNSUPPORTED,
+                              message="Legacy endpoint discovery is unavailable")
diff --git a/kdn_server/gateway/mock.py b/kdn_server/gateway/mock.py
new file mode 100644
index 0000000..c5461ac
--- /dev/null
+++ b/kdn_server/gateway/mock.py
@@ -0,0 +1,160 @@
+"""Deterministic CPU-only gateway for contract tests and policy development."""
+from __future__ import annotations
+
+from kdn_server.contracts.cache_service import *
+from kdn_server.contracts.common import utc_now
+from kdn_server.contracts.errors import OutcomeCode
+from kdn_server.domain import CacheOperationState, CacheOperationTask, CacheOperationType
+from .base import GatewayAdapterBase
+from .capabilities import SupportState
+
+
+_OPERATION_CAPABILITY = {
+    CacheOperationType.PREFETCH: "warm_prefetch", CacheOperationType.REBUILD: "warm_prefetch",
+    CacheOperationType.PIN: "pin_unpin", CacheOperationType.UNPIN: "pin_unpin",
+    CacheOperationType.CLEAR: "object_deletion",
+}
+
+
+class MockGateway(GatewayAdapterBase):
+    """Fixture-only CPU adapter; it performs no backend or device I/O."""
+    def __init__(self, capabilities, *, artifacts=(), observations=(), token_fixtures=None,
+                 endpoints=(), adapter_summary=None, tier_summary=None, maintenance_summary=None):
+        self.capabilities = capabilities
+        self.artifacts = {x.artifact_id: x for x in artifacts}
+        self.observations = {x.artifact_id: x for x in observations}
+        self.token_fixtures = dict(token_fixtures or {})
+        self.endpoints = tuple(endpoints)
+        self.adapter_summary, self.tier_summary = adapter_summary, tier_summary
+        self.maintenance_summary = maintenance_summary
+        self._tasks, self._logical = {}, {}
+
+    def lookup_artifact(self, request):
+        if (guard := self._gate(request, self.capabilities.artifact_lookup)): return LookupArtifactResponse.model_validate(guard.model_dump())
+        value = self.artifacts.get(request.artifact_id)
+        if value is None: return self._response(request, response_type=LookupArtifactResponse, outcome=OutcomeCode.STALE, message="artifact not observed")
+        if value.runtime_profile is not request.runtime_profile or value.compatibility_profile_id != request.compatibility_profile_id:
+            return self._response(request, response_type=LookupArtifactResponse, outcome=OutcomeCode.INCOMPATIBLE, message="artifact provenance mismatch")
+        return self._response(request, response_type=LookupArtifactResponse, artifact=value)
+
+    def get_cache_observation(self, request):
+        response_at = utc_now()
+        if (guard := self._gate(request, self.capabilities.cache_observation)): return GetCacheObservationResponse.model_validate(guard.model_dump())
+        value = self.observations.get(request.artifact_id)
+        if value is None:
+            return self._response(request, response_type=GetCacheObservationResponse, outcome=OutcomeCode.STALE, message="observation is absent or stale",
+                                  observation=None)
+        incompatible = any((value.artifact_id != request.artifact_id, value.runtime_profile is not request.runtime_profile,
+            value.compatibility_profile_id != request.compatibility_profile_id, value.endpoint_id != request.endpoint_id))
+        if incompatible: return self._response(request, response_type=GetCacheObservationResponse, outcome=OutcomeCode.INCOMPATIBLE, message="observation provenance mismatch")
+        if value.endpoint_generation != request.endpoint_generation: return self._response(request, response_type=GetCacheObservationResponse, outcome=OutcomeCode.STALE, message="observation generation mismatch")
+        if not value.is_fresh(at=response_at):
+            return self._response(request, response_type=GetCacheObservationResponse, timestamp=response_at, outcome=OutcomeCode.STALE, message="observation is stale", observation=value)
+        return self._response(request, response_type=GetCacheObservationResponse, timestamp=response_at, observation=value)
+
+    def lookup_tokens(self, request):
+        if (guard := self._gate(request, self.capabilities.token_lookup)): return LookupTokensResponse.model_validate(guard.model_dump())
+        key = request.tokens.token_ids or request.tokens.token_reference.reference_id
+        coverage = self.token_fixtures.get(key)
+        if coverage is None:
+            return self._response(request, response_type=LookupTokensResponse, outcome=OutcomeCode.TEXT_FALLBACK, message="tokens are not cached",
+                                  fallback_eligible=True)
+        # Whole-request lookup is useful even when exact range reporting is unknown.
+        if self.capabilities.range_coverage is not SupportState.SUPPORTED and coverage.covered_ranges:
+            coverage = coverage.model_copy(update={"covered_ranges": ()})
+        return self._response(request, response_type=LookupTokensResponse, token_coverage=coverage)
+
+    def submit_operation(self, request):
+        operation = INTENT_OPERATION_TYPES.get(type(request))
+        if operation is None: raise TypeError("unknown operation intent")
+        capability = getattr(self.capabilities, _OPERATION_CAPABILITY[operation])
+        response_type = {CreatePrefetchIntentRequest: CreatePrefetchIntentResponse, CreatePinIntentRequest: CreatePinIntentResponse,
+            CreateUnpinIntentRequest: CreateUnpinIntentResponse, CreateClearIntentRequest: CreateClearIntentResponse,
+            CreateRebuildIntentRequest: CreateRebuildIntentResponse}[type(request)]
+        if (guard := self._gate(request, capability)): return response_type.model_validate(guard.model_dump())
+        logical = (operation.value, request.artifact_id, request.endpoint_id, request.endpoint_generation,
+                   request.runtime_profile.value, request.compatibility_profile_id)
+        existing = self._tasks.get(request.idempotency_key)
+        if existing:
+            # Idempotency covers the complete logical target, not merely the key text.
+            if self._logical[request.idempotency_key] != logical:
+                return self._response(request, response_type=response_type, outcome=OutcomeCode.IDEMPOTENCY_CONFLICT,
+                    message="idempotency key was used for a different logical request")
+            return self._response(request, response_type=response_type, operation=existing)
+        task = CacheOperationTask(idempotency_key=request.idempotency_key, operation=operation,
+            artifact_id=request.artifact_id, runtime_profile=request.runtime_profile,
+            compatibility_profile_id=request.compatibility_profile_id, gateway_profile="mock",
+            endpoint_id=request.endpoint_id, endpoint_generation=request.endpoint_generation)
+        self._tasks[request.idempotency_key], self._logical[request.idempotency_key] = task, logical
+        return self._response(request, response_type=response_type, operation=task)
+
+    def complete(self, task_id, *, failed=False):
+        """Deterministically simulate asynchronous execution without worker I/O."""
+        key, task = self._find(task_id)
+        if task is None: raise KeyError(task_id)
+        if task.state is CacheOperationState.PENDING: task = task.transition("running")
+        task = task.transition("failed" if failed else "succeeded")
+        self._tasks[key] = task
+        return task
+
+    def start(self, task_id):
+        key, task = self._find(task_id)
+        if task is None: raise KeyError(task_id)
+        task = task.transition("running"); self._tasks[key] = task
+        return task
+
+    def _find(self, task_id):
+        return next(((key, task) for key, task in self._tasks.items() if task.task_id == task_id), (None, None))
+
+    def get_operation_status(self, request):
+        if (guard := self._gate(request, self.capabilities.operation_status)): return GetOperationStatusResponse.model_validate(guard.model_dump())
+        _, task = self._find(request.task_id)
+        if task is None: return self._response(request, response_type=GetOperationStatusResponse, outcome=OutcomeCode.STALE, message="operation was not found")
+        return self._response(request, response_type=GetOperationStatusResponse, operation=task)
+
+    def cancel_operation(self, request):
+        if (guard := self._gate(request, self.capabilities.cancellation)): return CancelOperationResponse.model_validate(guard.model_dump())
+        key, task = self._find(request.task_id)
+        if task is None: return self._response(request, response_type=CancelOperationResponse, outcome=OutcomeCode.STALE, message="operation was not found")
+        if task.terminal: return self._response(request, response_type=CancelOperationResponse, operation=task)
+        task = task.transition("cancelled"); self._tasks[key] = task
+        return self._response(request, response_type=CancelOperationResponse, outcome=OutcomeCode.CANCELLED, message="operation cancelled", operation=task)
+
+    def get_endpoints(self, request):
+        if (guard := self._discovery_negotiate(request, GetLMCacheEndpointsResponse)): return guard
+        for endpoint in self.endpoints:
+            if any((endpoint.runtime_profile is not self.capabilities.runtime_profile,
+                    endpoint.compatibility_profile_id != self.capabilities.compatibility_profile.compatibility_profile_id,
+                    endpoint.endpoint_id != self.capabilities.endpoint_id,
+                    endpoint.generation != self.capabilities.endpoint_generation)):
+                return self._response(request, response_type=GetLMCacheEndpointsResponse, outcome=OutcomeCode.INCOMPATIBLE, message="endpoint fixture provenance mismatch")
+        return self._response(request, response_type=GetLMCacheEndpointsResponse, endpoints=self.endpoints)
+
+    def get_tier_adapter_summary(self, request):
+        if (guard := self._gate(request, self.capabilities.tier_capacity_usage)): return GetTierAndAdapterSummaryResponse.model_validate(guard.model_dump())
+        if self.adapter_summary is None or self.tier_summary is None:
+            # Absence is a stale discovery fact, not a malformed success response.
+            return self._response(request, response_type=GetTierAndAdapterSummaryResponse,
+                outcome=OutcomeCode.STALE, message="tier or adapter summary fixture is absent")
+        if not self._summary_matches(request, self.adapter_summary) or not self._summary_matches(request, self.tier_summary):
+            # A valid fixture for another endpoint is still incompatible here.
+            return self._response(request, response_type=GetTierAndAdapterSummaryResponse,
+                outcome=OutcomeCode.INCOMPATIBLE, message="tier or adapter summary provenance mismatch")
+        return self._response(request, response_type=GetTierAndAdapterSummaryResponse, adapter_summary=self.adapter_summary, tier_summary=self.tier_summary)
+
+    def get_maintenance_status(self, request):
+        if (guard := self._gate(request, self.capabilities.maintenance_eviction)): return GetMaintenanceStatusResponse.model_validate(guard.model_dump())
+        if self.maintenance_summary is None:
+            return self._response(request, response_type=GetMaintenanceStatusResponse,
+                outcome=OutcomeCode.STALE, message="maintenance summary fixture is absent")
+        if not self._summary_matches(request, self.maintenance_summary):
+            return self._response(request, response_type=GetMaintenanceStatusResponse,
+                outcome=OutcomeCode.INCOMPATIBLE, message="maintenance summary provenance mismatch")
+        return self._response(request, response_type=GetMaintenanceStatusResponse, maintenance_summary=self.maintenance_summary)
+
+    @staticmethod
+    def _summary_matches(request, summary):
+        return all((summary.runtime_profile is request.runtime_profile,
+                    summary.compatibility_profile_id == request.compatibility_profile_id,
+                    summary.endpoint_id == request.endpoint_id,
+                    summary.endpoint_generation == request.endpoint_generation))
diff --git a/kdn_server/gateway/profiles.py b/kdn_server/gateway/profiles.py
new file mode 100644
index 0000000..671c6c8
--- /dev/null
+++ b/kdn_server/gateway/profiles.py
@@ -0,0 +1,34 @@
+"""Gateway adapter and LMCache compatibility profile vocabulary."""
+from enum import Enum
+
+from pydantic import Field
+
+from kdn_server.contracts.common import ContractModel
+
+
+class GatewayTransportKind(str, Enum):
+    MP_HTTP_API = "mp_http_api"
+    MP_COORDINATOR = "mp_coordinator"
+    MP_SDK = "mp_sdk"
+    MP_METRICS_EVENTS = "mp_metrics_events"
+    MP_L2_PLUGIN = "mp_l2_plugin"
+    LEGACY_REDIS = "legacy_redis"
+    MOCK = "mock"
+    UNKNOWN_FUTURE = "unknown_future"
+
+
+class GatewayAdapterBinding(ContractModel):
+    transport_kind: GatewayTransportKind
+    binding_id: str = Field(min_length=1)
+
+
+class LMCacheCompatibilityProfile(ContractModel):
+    compatibility_profile_id: str = Field(min_length=1)
+    lmcache_version: str | None = None
+    lmcache_build: str | None = None
+    config_profile: str | None = None
+    layout_profile: str | None = None
+    serde_profile: str | None = None
+    chunk_size: int | None = Field(default=None, gt=0)
+    connector_profile: str | None = None
+    key_hash_profile: str | None = None
diff --git a/kdn_server/gateway/protocol.py b/kdn_server/gateway/protocol.py
new file mode 100644
index 0000000..2dc7ac2
--- /dev/null
+++ b/kdn_server/gateway/protocol.py
@@ -0,0 +1,32 @@
+"""Dependency-light interface implemented only by gateway adapters."""
+from typing import Protocol, runtime_checkable
+
+from kdn_server.contracts.cache_service import (
+    ArtifactRequest, CancelOperationRequest, CancelOperationResponse,
+    CreateClearIntentResponse, CreatePinIntentResponse, CreatePrefetchIntentResponse,
+    CreateRebuildIntentResponse, CreateUnpinIntentResponse,
+    GetCacheObservationResponse, GetLMCacheEndpointsRequest, GetLMCacheEndpointsResponse,
+    GetMaintenanceStatusRequest, GetMaintenanceStatusResponse, GetOperationStatusRequest,
+    GetOperationStatusResponse, GetTierAndAdapterSummaryRequest,
+    GetTierAndAdapterSummaryResponse, LookupArtifactResponse, LookupTokensRequest,
+    LookupTokensResponse, OperationIntentRequest,
+)
+from .capabilities import CapabilitySnapshot
+
+
+@runtime_checkable
+class LMCacheGateway(Protocol):
+    """Stable orchestration boundary implemented by all future adapters."""
+    def discover_capabilities(self) -> CapabilitySnapshot: ...
+    def lookup_artifact(self, request: ArtifactRequest) -> LookupArtifactResponse: ...
+    def lookup_tokens(self, request: LookupTokensRequest) -> LookupTokensResponse: ...
+    def get_cache_observation(self, request: ArtifactRequest) -> GetCacheObservationResponse: ...
+    def submit_operation(self, request: OperationIntentRequest) -> (
+        CreatePrefetchIntentResponse | CreatePinIntentResponse | CreateUnpinIntentResponse |
+        CreateClearIntentResponse | CreateRebuildIntentResponse
+    ): ...
+    def get_operation_status(self, request: GetOperationStatusRequest) -> GetOperationStatusResponse: ...
+    def cancel_operation(self, request: CancelOperationRequest) -> CancelOperationResponse: ...
+    def get_endpoints(self, request: GetLMCacheEndpointsRequest) -> GetLMCacheEndpointsResponse: ...
+    def get_tier_adapter_summary(self, request: GetTierAndAdapterSummaryRequest) -> GetTierAndAdapterSummaryResponse: ...
+    def get_maintenance_status(self, request: GetMaintenanceStatusRequest) -> GetMaintenanceStatusResponse: ...
diff --git a/test/kdn/README.md b/test/kdn/README.md
new file mode 100644
index 0000000..71e95e6
--- /dev/null
+++ b/test/kdn/README.md
@@ -0,0 +1,98 @@
+# CPU-only KDN contract and Gateway workflows
+
+[Back to the KDN architecture](../../kdn_server/README.md) · [Contract reference](../../kdn_server/contracts/README.md) · [Gateway reference](../../kdn_server/gateway/README.md) · [Domain reference](../../kdn_server/domain/README.md)
+
+These tests validate control-plane contracts and adapter behavior without a network, Redis, filesystem cache, GPU, vLLM, or LMCache runtime.
+
+## Test groups
+
+- [`test_domain_models.py`](test_domain_models.py) checks canonical identities, observation freshness/provenance, Legacy projection, and task/queue transitions.
+- Runtime/Profile compatibility tests cover startup resolution and explicit v1, Legacy, and test/mock isolation.
+- [`test_cache_service_contracts.py`](test_cache_service_contracts.py) checks every Cache Service family, serialization, negotiation, capability gates, and dedicated response invariants.
+- The same contract suite executes deterministic Mock and read-only Legacy workflows, including idempotency, lifecycle, cancellation, stale and incompatible targets, and fallback.
+- Parameterized security tests prove contracts reject secrets and physical cache representations.
+
+## Executable workflow
+
+The authoritative executable workflow is [`test_cache_service_contracts.py`](test_cache_service_contracts.py). The following setup is a minimal interactive starting point using non-routable logical identities:
+
+```python
+from kdn_server.contracts import GetLMCacheEndpointsRequest
+from kdn_server.gateway import (
+    CapabilitySnapshot, GatewayAdapterBinding, LMCacheCompatibilityProfile,
+    MockGateway,
+)
+
+profile = LMCacheCompatibilityProfile(
+    compatibility_profile_id="example-compatible-v1",
+    key_hash_profile="sha256-v1",
+)
+snapshot = CapabilitySnapshot(
+    runtime_profile="test/mock",
+    adapter_bindings=(GatewayAdapterBinding(
+        transport_kind="mock", binding_id="local-fixture"),),
+    compatibility_profile=profile,
+    endpoint_id="endpoint_" + "1" * 32,
+    endpoint_generation=1,
+    source="cpu-test",
+)
+gateway = MockGateway(snapshot)
+response = gateway.get_endpoints(
+    GetLMCacheEndpointsRequest(runtime_profile="test/mock")
+)
+assert response.outcome.value == "success"
+```
+
+The focused tests extend this setup through the complete workflow:
+
+| Step | Action | Expected visible result | Invariant proved |
+|---:|---|---|---|
+| 1 | Create `LMCacheCompatibilityProfile`. | Frozen compatibility identity. | Transport and compatibility identity remain separate. |
+| 2 | Create `CapabilitySnapshot`. | Ordered bindings and tri-state capabilities validate. | Runtime/transport isolation and explicit uncertainty. |
+| 3 | Construct `MockGateway`. | In-memory adapter is Protocol-compatible. | No production dependency is required. |
+| 4 | Discover endpoints. | Dedicated success response with endpoint tuple. | Discovery preserves Runtime Profile and correlation. |
+| 5 | Look up a fixture `CacheArtifact`. | `success` plus the compatible artifact. | Logical artifact provenance is checked. |
+| 6 | Retrieve a fresh observation. | `success` only inside its TTL. | Freshness and endpoint generation are response-time facts. |
+| 7 | Look up tokens with range support on/off. | Whole hit/miss remains; ranges appear only when supported. | `token_lookup` and `range_coverage` are independent. |
+| 8 | Submit a prefetch intent. | A pending prefetch task. | Intent response type matches operation type. |
+| 9 | Repeat the same intent/key. | The same task ID. | Equivalent submission is idempotent. |
+| 10 | Reuse the key for different work. | `idempotency_conflict`. | A key cannot alias distinct logical work. |
+| 11 | Transition pending → running → succeeded. | Explicit state at each step. | CPU-only asynchronous lifecycle simulation. |
+| 12 | Cancel a pending task. | `cancelled` and cancelled task state. | Cancellation is an explicit terminal transition. |
+| 13 | Cancel terminal work. | `success`; original terminal state is unchanged. | Cancellation is an idempotent no-op after completion. |
+| 14 | Request an old generation. | `stale`. | Facts cannot cross endpoint generations. |
+| 15 | Request a different profile. | `incompatible`. | Target identity is negotiated before execution. |
+| 16 | Send a Legacy write. | `unsupported`. | Legacy is read-only. |
+| 17 | Send v1 work to Legacy. | `incompatible`. | Runtime profiles cannot be mixed implicitly. |
+| 18 | Add a secret/physical field. | Pydantic validation failure. | The wire boundary is storage-neutral and safe. |
+| 19 | Construct an MP production transport. | `NotImplementedError`. | Production Gateway I/O is intentionally absent. |
+
+Run the focused test with `-k` to inspect any invariant, for example:
+
+```bash
+python3 -m pytest -q test/kdn/test_cache_service_contracts.py -k idempotency
+```
+
+## Validation commands
+
+Install [`requirements.txt`](../../requirements.txt) for the complete application and test dependency surface, then run from the repository root:
+
+```bash
+python3 -m pytest -q test/kdn
+python3 -m pytest -q test/kdn/test_cache_service_contracts.py
+python3 -m pytest -q test/test_runtime_compat.py
+python3 -m pytest -q \
+  test/test_instance_capability.py \
+  test/test_instance_capability_registration.py
+
+python3 -m compileall -q \
+  cacheroute_compat \
+  kdn_server/contracts \
+  kdn_server/domain \
+  kdn_server/gateway \
+  test/kdn
+
+git diff --check
+```
+
+A successful contract run reports only passed tests. Validation errors, raw backend exceptions, unexpected range data, changed task identity, or a non-structured Gateway failure indicate a contract regression.
diff --git a/test/kdn/test_cache_service_contracts.py b/test/kdn/test_cache_service_contracts.py
new file mode 100644
index 0000000..f55154e
--- /dev/null
+++ b/test/kdn/test_cache_service_contracts.py
@@ -0,0 +1,395 @@
+from datetime import datetime, timedelta, timezone
+from typing import get_args
+
+import pytest
+from pydantic import ValidationError
+
+from kdn_server.contracts import *
+from kdn_server.contracts.cache_service import CacheServiceResponse, GatewayTargetedResponse
+from kdn_server.domain import CacheArtifact, CacheReplicaObservation, LMCacheEndpoint
+from kdn_server.gateway import *
+
+NOW = datetime.now(timezone.utc)
+ENDPOINT_ID = "endpoint_" + "1" * 32
+
+
+def binding(kind="mock"):
+    return GatewayAdapterBinding(transport_kind=kind, binding_id=f"{kind}-binding")
+
+
+def capabilities(**changes):
+    values = dict(runtime_profile="test/mock", adapter_bindings=(binding(),),
+        compatibility_profile=LMCacheCompatibilityProfile(compatibility_profile_id="cpu-test", key_hash_profile="sha256-v1"),
+        endpoint_id=ENDPOINT_ID, endpoint_generation=1, source="fixture", loaded_adapters=("a", "b"),
+        token_lookup="supported", warm_prefetch="supported", pin_unpin="supported",
+        object_deletion="supported", operation_status="supported", cancellation="supported",
+        tier_capacity_usage="supported", maintenance_eviction="supported",
+        artifact_lookup="supported", cache_observation="supported")
+    return CapabilitySnapshot(**(values | changes))
+
+
+def target(**changes):
+    return dict(runtime_profile="test/mock", compatibility_profile_id="cpu-test",
+                endpoint_id=ENDPOINT_ID, endpoint_generation=1) | changes
+
+
+def artifact(knowledge_id="kid", runtime_profile="test/mock"):
+    return CacheArtifact(knowledge_id=knowledge_id, model_profile="model", tokenizer_profile="tokenizer",
+        cache_data_profile="layout", compatibility_profile_id="cpu-test", runtime_profile=runtime_profile)
+
+
+def intent(cls=CreatePrefetchIntentRequest, **changes):
+    return cls(**(target() | dict(artifact_id=artifact().artifact_id, idempotency_key="same") | changes))
+
+
+def observation(*, fresh=True, runtime_profile="test/mock", legacy=False):
+    observed = datetime.now(timezone.utc) - timedelta(seconds=1) if fresh else NOW - timedelta(minutes=20)
+    return CacheReplicaObservation(artifact_id=artifact(runtime_profile=runtime_profile).artifact_id,
+        state="available", source="legacy_projection" if legacy else "mock", endpoint_id=ENDPOINT_ID,
+        endpoint_generation=0 if legacy else 1, runtime_profile=runtime_profile,
+        gateway_profile="legacy_gateway" if legacy else "mock", compatibility_profile_id="cpu-test",
+        source_observed_at=observed, projected_at=max(NOW, observed), expires_at=observed + timedelta(minutes=10),
+        legacy_projection=legacy, compatibility_uncertain=legacy)
+
+
+def error(code, fallback=False):
+    return ContractErrorDetail(code=code, message=code.value, fallback_eligible=fallback)
+
+
+def test_versions_profiles_freezing_and_round_trips():
+    requests = [
+        GetCacheObservationRequest(**target(), artifact_id=artifact().artifact_id),
+        LookupArtifactRequest(**target(), artifact_id=artifact().artifact_id),
+        LookupTokensRequest(**target(), tokens=TokenInput(token_ids=(1, 2))), intent(),
+        GetOperationStatusRequest(**target(), task_id="cacheop_" + "1" * 32),
+        CancelOperationRequest(**target(), task_id="cacheop_" + "1" * 32),
+        GetLMCacheEndpointsRequest(runtime_profile="test/mock"), GetTierAndAdapterSummaryRequest(**target()),
+        GetMaintenanceStatusRequest(**target()), ResolveKnowledgeRequest(runtime_profile="v1", knowledge_id="kid"),
+    ]
+    for request in requests:
+        assert type(request).model_validate_json(request.model_dump_json()) == request
+        with pytest.raises(ValidationError): request.runtime_profile = "legacy"
+    with pytest.raises(ValidationError): ResolveKnowledgeRequest(runtime_profile="auto", knowledge_id="kid")
+    with pytest.raises(ValidationError): ResolveKnowledgeRequest(runtime_profile="v1", knowledge_id="kid", contract_version="v2")
+
+
+@pytest.mark.parametrize("request_cls,values", [
+    (LookupArtifactRequest, {"artifact_id": artifact().artifact_id}),
+    (GetCacheObservationRequest, {"artifact_id": artifact().artifact_id}),
+    (LookupTokensRequest, {"tokens": TokenInput(token_ids=(1,))}),
+    (GetOperationStatusRequest, {"task_id": "cacheop_" + "1" * 32}),
+    (CancelOperationRequest, {"task_id": "cacheop_" + "1" * 32}),
+    (GetTierAndAdapterSummaryRequest, {}), (GetMaintenanceStatusRequest, {}),
+])
+def test_target_metadata_is_required_and_validated(request_cls, values):
+    for missing in ("compatibility_profile_id", "endpoint_id", "endpoint_generation"):
+        data = target() | values; data.pop(missing)
+        with pytest.raises(ValidationError): request_cls(**data)
+    with pytest.raises(ValidationError): request_cls(**(target(endpoint_id="bad") | values))
+    with pytest.raises(ValidationError): request_cls(**(target(endpoint_generation=0) | values))
+
+
+def test_token_coverage_boundaries():
+    assert TokenCoverage(whole_request_hit=True, total_tokens=5, covered_ranges=((0, 2), (2, 5)))
+    for ranges in (((-1, 1),), ((1, 1),), ((1, 6),), ((2, 4), (1, 2)), ((0, 3), (2, 4))):
+        with pytest.raises(ValidationError): TokenCoverage(whole_request_hit=False, total_tokens=5, covered_ranges=ranges)
+
+
+def test_transport_runtime_matrix_composition_and_factory_membership():
+    mp = ("mp_http_api", "mp_coordinator", "mp_sdk", "mp_metrics_events", "mp_l2_plugin", "unknown_future")
+    composed = capabilities(runtime_profile="v1", adapter_bindings=tuple(binding(x) for x in mp))
+    assert tuple(x.transport_kind.value for x in composed.adapter_bindings) == mp
+    for profile, good in (("v1", "mp_sdk"), ("legacy", "legacy_redis"), ("test/mock", "mock")):
+        assert capabilities(runtime_profile=profile, adapter_bindings=(binding(good),))
+        for bad in set(x.value for x in GatewayTransportKind) - ({*mp} if profile == "v1" else {good}):
+            with pytest.raises(ValidationError): capabilities(runtime_profile=profile, adapter_bindings=(binding(bad),))
+    with pytest.raises(ValueError): create_gateway("legacy_redis", capabilities())
+    with pytest.raises(ValidationError): capabilities(adapter_bindings=(binding(), binding()))
+
+
+def test_names_compatibility_profile_and_snapshot_immutability():
+    assert capabilities().compatibility_profile.key_hash_profile == "sha256-v1"
+    for changes in ({"loaded_adapters": ("a", "a")}, {"loaded_adapters": ("",)}, {"l1_tiers": ("x", "x")}):
+        with pytest.raises(ValidationError): capabilities(**changes)
+    snapshot = capabilities(token_lookup="unknown")
+    assert not snapshot.token_lookup
+    with pytest.raises(ValidationError): snapshot.loaded_adapters = ("new",)
+
+
+def test_negotiation_mismatch_and_generation_stale_preserve_correlation():
+    gateway = MockGateway(capabilities())
+    base = dict(tokens=TokenInput(token_ids=(1,)), request_id="request", correlation_id="correlation")
+    for changes, outcome in (({"runtime_profile": "legacy"}, OutcomeCode.INCOMPATIBLE),
+                             ({"compatibility_profile_id": "other"}, OutcomeCode.INCOMPATIBLE),
+                             ({"endpoint_id": "endpoint_" + "2" * 32}, OutcomeCode.INCOMPATIBLE),
+                             ({"endpoint_generation": 2}, OutcomeCode.STALE)):
+        response = gateway.lookup_tokens(LookupTokensRequest(**(target(**changes) | base)))
+        assert response.outcome is outcome and response.request_id == "request" and response.correlation_id == "correlation"
+        assert response.error.code is outcome
+
+
+@pytest.mark.parametrize("request_factory,capability", [
+    (lambda: LookupArtifactRequest(**target(), artifact_id=artifact().artifact_id), "artifact_lookup"),
+    (lambda: GetCacheObservationRequest(**target(), artifact_id=artifact().artifact_id), "cache_observation"),
+    (lambda: LookupTokensRequest(**target(), tokens=TokenInput(token_ids=(1,))), "token_lookup"),
+    (lambda: intent(), "warm_prefetch"), (lambda: intent(CreateRebuildIntentRequest), "warm_prefetch"),
+    (lambda: intent(CreatePinIntentRequest), "pin_unpin"), (lambda: intent(CreateUnpinIntentRequest), "pin_unpin"),
+    (lambda: intent(CreateClearIntentRequest), "object_deletion"),
+    (lambda: GetOperationStatusRequest(**target(), task_id="cacheop_" + "1" * 32), "operation_status"),
+    (lambda: CancelOperationRequest(**target(), task_id="cacheop_" + "1" * 32), "cancellation"),
+    (lambda: GetTierAndAdapterSummaryRequest(**target()), "tier_capacity_usage"),
+    (lambda: GetMaintenanceStatusRequest(**target()), "maintenance_eviction"),
+])
+@pytest.mark.parametrize("state", ["unsupported", "unknown"])
+def test_every_capability_gate_rejects_unsupported_and_unknown(request_factory, capability, state):
+    gateway = MockGateway(capabilities(**{capability: state}))
+    request = request_factory()
+    method = (gateway.lookup_tokens if isinstance(request, LookupTokensRequest) else
+              gateway.lookup_artifact if isinstance(request, LookupArtifactRequest) else
+              gateway.get_cache_observation if isinstance(request, GetCacheObservationRequest) else
+              gateway.submit_operation if isinstance(request, OperationIntentRequest) else
+              gateway.get_operation_status if isinstance(request, GetOperationStatusRequest) and not isinstance(request, CancelOperationRequest) else
+              gateway.cancel_operation if isinstance(request, CancelOperationRequest) else
+              gateway.get_tier_adapter_summary if isinstance(request, GetTierAndAdapterSummaryRequest) else gateway.get_maintenance_status)
+    assert method(request).outcome is OutcomeCode.UNSUPPORTED
+
+
+def test_lookup_fallback_idempotency_completion_and_terminal_cancel():
+    cache = artifact(); gateway = MockGateway(capabilities(), artifacts=(cache,))
+    assert gateway.lookup_artifact(LookupArtifactRequest(**target(), artifact_id=cache.artifact_id)).artifact == cache
+    fallback = gateway.lookup_tokens(LookupTokensRequest(**target(), tokens=TokenInput(token_ids=(1, 2))))
+    assert fallback.outcome is OutcomeCode.TEXT_FALLBACK and fallback.error.fallback_eligible
+    first = gateway.submit_operation(intent()).operation
+    assert gateway.submit_operation(intent()).operation.task_id == first.task_id
+    conflict = gateway.submit_operation(intent(artifact_id=artifact("different").artifact_id))
+    assert conflict.outcome is OutcomeCode.IDEMPOTENCY_CONFLICT
+    assert conflict.runtime_profile.value == "test/mock" and conflict.endpoint_id == ENDPOINT_ID
+    running = gateway.start(first.task_id)
+    cancelled = gateway.cancel_operation(CancelOperationRequest(**target(), task_id=running.task_id))
+    assert cancelled.outcome is OutcomeCode.CANCELLED and cancelled.operation.state.value == "cancelled"
+    done = gateway.submit_operation(intent(idempotency_key="done")).operation
+    gateway.complete(done.task_id)
+    terminal = gateway.cancel_operation(CancelOperationRequest(**target(), task_id=done.task_id))
+    assert terminal.outcome is OutcomeCode.SUCCESS and terminal.operation.state.value == "succeeded"
+
+
+def test_summary_models_and_response_consistency():
+    provenance = dict(source="mock", runtime_profile="test/mock", compatibility_profile_id="cpu-test",
+        endpoint_id=ENDPOINT_ID, endpoint_generation=1)
+    capacity = CapacityUsageObservation(**provenance, support="supported", tier_name="cpu", tier_level="l1", capacity_bytes=10, used_bytes=4)
+    adapter = AdapterSummary(**provenance, support="supported", loaded_adapters=("a",))
+    tier = TierSummary(**provenance, support="supported", l1_tiers=("cpu",), capacity=(capacity,))
+    maintenance = MaintenanceSummary(**provenance, support="unknown", partial=True)
+    for model in (capacity, adapter, tier, maintenance): assert type(model).model_validate_json(model.model_dump_json()) == model
+    gateway = MockGateway(capabilities(), adapter_summary=adapter, tier_summary=tier, maintenance_summary=maintenance)
+    assert gateway.get_tier_adapter_summary(GetTierAndAdapterSummaryRequest(**target())).tier_summary == tier
+    assert gateway.get_maintenance_status(GetMaintenanceStatusRequest(**target())).maintenance_summary == maintenance
+    with pytest.raises(ValidationError): CacheServiceResponse(runtime_profile="test/mock", outcome="success", error=error(OutcomeCode.FAILED))
+    with pytest.raises(ValidationError): CacheServiceResponse(runtime_profile="test/mock", outcome="failed")
+    with pytest.raises(ValidationError): CacheServiceResponse(runtime_profile="test/mock", outcome="text_fallback", error=error(OutcomeCode.TEXT_FALLBACK))
+
+
+def test_legacy_fresh_stale_read_only_and_v1_incompatible():
+    caps = capabilities(runtime_profile="legacy", adapter_bindings=(binding("legacy_redis"),),
+        endpoint_generation=0, token_lookup="unsupported", warm_prefetch="unsupported", cancellation="unsupported")
+    fresh, stale = observation(fresh=True, runtime_profile="legacy", legacy=True), observation(fresh=False, runtime_profile="legacy", legacy=True)
+    request = GetCacheObservationRequest(**target(runtime_profile="legacy", endpoint_generation=0), artifact_id=fresh.artifact_id)
+    assert LegacyCacheAdapter(caps, observations=(fresh,)).get_cache_observation(request).outcome is OutcomeCode.SUCCESS
+    assert LegacyCacheAdapter(caps, observations=(stale,)).get_cache_observation(request).outcome is OutcomeCode.STALE
+    legacy = create_gateway("legacy_redis", caps)
+    assert legacy.submit_operation(intent(runtime_profile="legacy", endpoint_generation=0)).outcome is OutcomeCode.UNSUPPORTED
+    assert legacy.submit_operation(intent(runtime_profile="v1")).outcome is OutcomeCode.INCOMPATIBLE
+    endpoints = legacy.get_endpoints(GetLMCacheEndpointsRequest(runtime_profile="legacy", request_id="r", correlation_id="c"))
+    assert endpoints.outcome is OutcomeCode.UNSUPPORTED and (endpoints.request_id, endpoints.correlation_id) == ("r", "c")
+
+
+@pytest.mark.parametrize("field", ["password", "api_key", "credentials", "authorization", "redis_key",
+    "kv_bytes", "tensor", "device_pointer", "chunk_index", "block_index", "lmcache_object"])
+@pytest.mark.parametrize("factory", [
+    lambda **x: ResolveKnowledgeRequest(runtime_profile="v1", knowledge_id="kid", **x),
+    lambda **x: LookupArtifactRequest(**target(), artifact_id=artifact().artifact_id, **x),
+    lambda **x: LookupTokensRequest(**target(), tokens=TokenInput(token_ids=(1,)), **x),
+    lambda **x: intent(**x), lambda **x: GetOperationStatusRequest(**target(), task_id="cacheop_" + "1" * 32, **x),
+    lambda **x: GetLMCacheEndpointsRequest(runtime_profile="test/mock", **x),
+])
+def test_contract_families_reject_secrets_and_physical_payloads(factory, field):
+    with pytest.raises(ValidationError): factory(**{field: "forbidden"})
+
+
+def test_protocol_conformance_and_dependency_isolation(monkeypatch):
+    gateway = MockGateway(capabilities())
+    assert isinstance(gateway, LMCacheGateway)
+    legacy_caps = capabilities(runtime_profile="legacy", endpoint_generation=0,
+        adapter_bindings=(binding("legacy_redis"),))
+    assert isinstance(LegacyCacheAdapter(legacy_caps), LMCacheGateway)
+    forbidden = {"socket", "redis", "torch", "vllm", "lmcache"}
+    assert not forbidden.intersection(type(gateway).__module__.split("."))
+
+
+def test_legacy_unknown_generation_and_mixed_generation_rejection():
+    assert GetCacheObservationRequest(**target(runtime_profile="legacy", endpoint_generation=0), artifact_id=artifact("x", "legacy").artifact_id)
+    with pytest.raises(ValidationError): GetCacheObservationRequest(**target(endpoint_generation=0), artifact_id=artifact().artifact_id)
+    with pytest.raises(ValidationError): CapabilitySnapshot(**(capabilities().model_dump() | {"endpoint_generation": 0}))
+    legacy_observation = observation(fresh=True, runtime_profile="legacy", legacy=True)
+    with pytest.raises(ValidationError):
+        GetCacheObservationResponse(**target(runtime_profile="legacy", endpoint_generation=1),
+            outcome="success", observation=legacy_observation)
+
+
+def test_fixture_provenance_and_dedicated_response_invariants():
+    wrong_artifact = artifact(runtime_profile="legacy")
+    gateway = MockGateway(capabilities(), artifacts=(wrong_artifact,))
+    request = LookupArtifactRequest(**target(), artifact_id=wrong_artifact.artifact_id)
+    assert gateway.lookup_artifact(request).outcome is OutcomeCode.INCOMPATIBLE
+    endpoint = LMCacheEndpoint(name="different", runtime_profile="test/mock", gateway_profile="mock",
+        compatibility_profile_id="cpu-test", generation=1)
+    endpoint_result = MockGateway(capabilities(), endpoints=(endpoint,)).get_endpoints(
+        GetLMCacheEndpointsRequest(runtime_profile="test/mock"))
+    assert endpoint_result.outcome is OutcomeCode.INCOMPATIBLE
+    required = ((GetCacheObservationResponse, "observation"), (LookupArtifactResponse, "artifact"),
+        (LookupTokensResponse, "token_coverage"), (GetOperationStatusResponse, "operation"),
+        (GetLMCacheEndpointsResponse, "endpoints"),
+        (GetTierAndAdapterSummaryResponse, "adapter_summary"),
+        (GetMaintenanceStatusResponse, "maintenance_summary"))
+    for response_type, _ in required:
+        with pytest.raises(ValidationError): response_type(runtime_profile="test/mock")
+
+
+def test_deterministic_stale_round_trip_and_error_version():
+    stale = observation(fresh=False)
+    response = GetCacheObservationResponse(**target(), timestamp=NOW, outcome="stale",
+        error=error(OutcomeCode.STALE), observation=stale)
+    assert GetCacheObservationResponse.model_validate_json(response.model_dump_json()) == response
+    with pytest.raises(ValidationError): ContractErrorDetail(code="failed", message="bad", contract_version="v2")
+
+
+def test_public_exports_are_explicit():
+    import kdn_server.contracts as contracts
+    import kdn_server.gateway as gateway
+    assert "BaseModel" not in contracts.__all__ and "datetime" not in contracts.__all__
+    assert "CapabilitySnapshot" in gateway.__all__ and "MockGateway" in gateway.__all__
+
+
+def test_targeted_error_responses_require_target_metadata():
+    for response_type in (GetCacheObservationResponse, LookupArtifactResponse, LookupTokensResponse,
+                          GetOperationStatusResponse, GetTierAndAdapterSummaryResponse,
+                          GetMaintenanceStatusResponse):
+        with pytest.raises(ValidationError):
+            response_type(runtime_profile="test/mock", outcome="unsupported",
+                          error=error(OutcomeCode.UNSUPPORTED))
+
+
+def test_response_level_object_provenance_and_freshness():
+    stale = observation(fresh=False)
+    with pytest.raises(ValidationError):
+        GetCacheObservationResponse(**target(), timestamp=NOW, observation=stale)
+    wrong_artifact = artifact(runtime_profile="legacy")
+    with pytest.raises(ValidationError):
+        LookupArtifactResponse(**target(), artifact=wrong_artifact)
+    fresh = observation(fresh=True)
+    for changes in ({"runtime_profile": "legacy"}, {"compatibility_profile_id": "other"},
+                    {"endpoint_id": "endpoint_" + "2" * 32}):
+        with pytest.raises(ValidationError):
+            GetCacheObservationResponse(**target(**changes), observation=fresh)
+    gateway = MockGateway(capabilities())
+    operation = gateway.submit_operation(intent()).operation
+    with pytest.raises(ValidationError):
+        GetOperationStatusResponse(**target(compatibility_profile_id="other"), operation=operation)
+
+
+def test_nested_capacity_provenance_and_discovery_runtime_negotiation():
+    provenance = dict(source="mock", runtime_profile="test/mock", compatibility_profile_id="cpu-test",
+        endpoint_id=ENDPOINT_ID, endpoint_generation=1, support="supported")
+    capacity = CapacityUsageObservation(**provenance, tier_name="cpu", tier_level="l1")
+    with pytest.raises(ValidationError):
+        TierSummary(**(provenance | {"compatibility_profile_id": "other"}),
+                    l1_tiers=("cpu",), capacity=(capacity,))
+    request = GetLMCacheEndpointsRequest(runtime_profile="v1")
+    assert MockGateway(capabilities()).get_endpoints(request).outcome is OutcomeCode.INCOMPATIBLE
+    legacy_caps = capabilities(runtime_profile="legacy", endpoint_generation=0,
+        adapter_bindings=(binding("legacy_redis"),))
+    assert LegacyCacheAdapter(legacy_caps).get_endpoints(request).outcome is OutcomeCode.INCOMPATIBLE
+
+
+def test_protocol_dedicated_return_annotations_and_repeat_round_trip():
+    import inspect
+    expected = {"lookup_artifact": "LookupArtifactResponse", "lookup_tokens": "LookupTokensResponse",
+        "get_cache_observation": "GetCacheObservationResponse",
+        "get_operation_status": "GetOperationStatusResponse", "cancel_operation": "CancelOperationResponse",
+        "get_endpoints": "GetLMCacheEndpointsResponse",
+        "get_tier_adapter_summary": "GetTierAndAdapterSummaryResponse",
+        "get_maintenance_status": "GetMaintenanceStatusResponse"}
+    for method, return_name in expected.items():
+        assert inspect.signature(getattr(LMCacheGateway, method)).return_annotation.__name__ == return_name
+    submit_annotation = inspect.signature(LMCacheGateway.submit_operation).return_annotation
+    assert len(submit_annotation.__args__) == 5
+    value = LookupTokensResponse(**target(), token_coverage=TokenCoverage(whole_request_hit=True, total_tokens=2, covered_ranges=((0, 2),)))
+    encoded = value.model_dump_json()
+    for _ in range(3):
+        value = LookupTokensResponse.model_validate_json(encoded)
+        assert value.model_dump_json() == encoded
+
+
+@pytest.mark.parametrize("lookup", ["supported", "unsupported", "unknown"])
+@pytest.mark.parametrize("ranges", ["supported", "unsupported", "unknown"])
+def test_token_and_range_capabilities_are_independent(lookup, ranges):
+    coverage = TokenCoverage(whole_request_hit=False, total_tokens=4, covered_ranges=((0, 2),))
+    gateway = MockGateway(capabilities(token_lookup=lookup, range_coverage=ranges),
+                          token_fixtures={(1, 2, 3, 4): coverage})
+    response = gateway.lookup_tokens(LookupTokensRequest(**target(), tokens=TokenInput(token_ids=(1, 2, 3, 4))))
+    if lookup != "supported":
+        assert response.outcome is OutcomeCode.UNSUPPORTED and response.token_coverage is None
+    else:
+        assert response.outcome is OutcomeCode.SUCCESS
+        assert response.token_coverage.whole_request_hit is False
+        assert bool(response.token_coverage.covered_ranges) is (ranges == "supported")
+
+
+def test_missing_and_mismatched_summary_fixtures_are_structured():
+    gateway = MockGateway(capabilities())
+    tier_request, maintenance_request = GetTierAndAdapterSummaryRequest(**target()), GetMaintenanceStatusRequest(**target())
+    assert gateway.get_tier_adapter_summary(tier_request).outcome is OutcomeCode.STALE
+    assert gateway.get_maintenance_status(maintenance_request).outcome is OutcomeCode.STALE
+    wrong = dict(source="mock", runtime_profile="test/mock", compatibility_profile_id="wrong",
+        endpoint_id=ENDPOINT_ID, endpoint_generation=1, support="supported")
+    adapter = AdapterSummary(**wrong, loaded_adapters=("a",))
+    tier = TierSummary(**wrong, l1_tiers=("cpu",))
+    maintenance = MaintenanceSummary(**wrong, active=True)
+    gateway = MockGateway(capabilities(), adapter_summary=adapter, tier_summary=tier, maintenance_summary=maintenance)
+    assert gateway.get_tier_adapter_summary(tier_request).outcome is OutcomeCode.INCOMPATIBLE
+    assert gateway.get_maintenance_status(maintenance_request).outcome is OutcomeCode.INCOMPATIBLE
+
+
+def test_operation_intent_response_types_reject_every_wrong_operation():
+    gateway = MockGateway(capabilities())
+    pairs = ((CreatePrefetchIntentRequest, CreatePrefetchIntentResponse),
+             (CreatePinIntentRequest, CreatePinIntentResponse),
+             (CreateUnpinIntentRequest, CreateUnpinIntentResponse),
+             (CreateClearIntentRequest, CreateClearIntentResponse),
+             (CreateRebuildIntentRequest, CreateRebuildIntentResponse))
+    tasks = [gateway.submit_operation(intent(request_type, idempotency_key=request_type.__name__)).operation
+             for request_type, _ in pairs]
+    for index, (_, response_type) in enumerate(pairs):
+        for wrong_index, task in enumerate(tasks):
+            if index != wrong_index:
+                with pytest.raises(ValidationError): response_type(**target(), operation=task)
+
+
+def test_cancel_response_success_requires_terminal_task():
+    gateway = MockGateway(capabilities())
+    pending = gateway.submit_operation(intent(idempotency_key="pending")).operation
+    running = gateway.start(gateway.submit_operation(intent(idempotency_key="running")).operation.task_id)
+    succeeded = gateway.complete(gateway.submit_operation(intent(idempotency_key="succeeded")).operation.task_id)
+    failed = gateway.complete(gateway.submit_operation(intent(idempotency_key="failed")).operation.task_id, failed=True)
+    cancelled = pending.transition("cancelled")
+    for task in (succeeded, failed, cancelled): assert CancelOperationResponse(**target(), operation=task)
+    for task in (pending, running):
+        with pytest.raises(ValidationError): CancelOperationResponse(**target(), operation=task)
+    assert CancelOperationResponse(**target(), outcome="cancelled", error=error(OutcomeCode.CANCELLED), operation=cancelled)
+
+
+def test_generic_response_cannot_bypass_target_metadata():
+    coverage = TokenCoverage(whole_request_hit=True, total_tokens=1)
+    for base in (CacheServiceResponse, GatewayTargetedResponse):
+        with pytest.raises(ValidationError): base(runtime_profile="test/mock", token_coverage=coverage)
+    assert "CacheServiceResponse" not in __import__("kdn_server.contracts", fromlist=["__all__"]).__all__