feat: define shard lifecycle status contract

.scratch/distributed-gguf-runtime/evidence/DGR-022/README.md

@@ -0,0 +1,46 @@
+# DGR-022 evidence — Shard lifecycle and structured status RPC contract
+
+**Completed:** 2026-07-17
+**Branch:** `ralph/distributed-gguf-runtime`
+**Authority:** `.scratch/distributed-gguf-runtime/prd.json`
+
+## Outcome
+
+Implemented the versioned, backend-neutral lifecycle/status contract consumed by a future generated gRPC binding. The contract keeps Meshnet routing, identity, authentication policy, billing, and llama.cpp ownership outside the worker contract.
+
+## Implemented
+
+- `packages/node/meshnet_node/shard_lifecycle.py`
+  - capability, health, session, cancellation, release, and metrics RPC names
+  - schema version negotiation and fail-closed unsupported-version handling
+  - structured status/error taxonomy with retryability and details
+  - lifecycle state machine for prefill/decode/cancel/release transitions
+  - monotonic idempotency-step enforcement and duplicate rejection
+  - bounded frame/byte flow control with cancellation-aware waits
+  - explicit cache expectation/result types
+  - deadline policy and TLS/auth transport hooks
+  - deterministic contract serialization round-trip
+- `tests/test_shard_lifecycle.py`
+  - contract round-trip and RPC coverage
+  - unsupported-version rejection
+  - malformed transition and idempotency rejection
+  - cancellation/release behavior
+  - bounded flow-control behavior
+  - TLS hook and incomplete-contract fail-closed behavior
+
+## Verification
+
+```text
+$ PYTHONPATH=packages/node pytest -q tests/test_shard_lifecycle.py tests/test_activation_envelope.py
+17 passed in 0.10s
+```
+
+The existing DGR-021 activation-envelope tests remain green alongside DGR-022.
+
+## Scope limitation
+
+This story defines the lifecycle/status contract only. Generated Python/C++ protobuf bindings and the concrete `shard_runtime.proto` generation pipeline are DGR-023 and remain separate.
+
+## Dependency handoff
+
+DGR-023 may consume the RPC names, status taxonomy, version identity, deadlines, flow-control limits, and TLS/auth hooks when the canonical `.proto` schema and toolchain are provisioned.

.scratch/distributed-gguf-runtime/prd.json

@@ -483,8 +483,8 @@
         "Add compatibility tests for supported versions and fail-closed tests for unsupported versions and malformed lifecycle transitions.",
         "Applicable shared quality gates in `prd.json` pass, and the evidence handoff records exact commands/results, changed files, limitations, and dependency handoff."
       ],
-      "passes": false,
-      "notes": "Generated source issue: .scratch/distributed-gguf-runtime/issues/022-define-shard-lifecycle-and-structured-status-rpcs.md; prd.json is authoritative.",
+      "passes": true,
+      "notes": "Completed from isolated DGR-022 Ralph worktree; verified with 17 focused pytest cases and retained DGR-021 envelope compatibility.",
       "blocks": [
         "DGR-024",
         "DGR-033",

packages/node/meshnet_node/shard_lifecycle.py

@@ -0,0 +1,472 @@
+"""Versioned Shard lifecycle and structured status contract.
+
+This module defines the semantic contract consumed by a future generated gRPC
+binding. It deliberately contains no Meshnet routing, authentication policy,
+billing, or llama.cpp types.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from enum import Enum
+import json
+import threading
+from typing import Any, Callable, ClassVar
+
+
+SCHEMA_NAME = "meshnet.shard-runtime"
+SCHEMA_VERSION = 1
+SUPPORTED_VERSIONS = frozenset({SCHEMA_VERSION})
+DEFAULT_MAX_INFLIGHT_FRAMES = 32
+DEFAULT_MAX_INFLIGHT_BYTES = 8 * 1024 * 1024
+
+
+class RpcName(str, Enum):
+    CAPABILITY = "GetCapability"
+    HEALTH = "CheckHealth"
+    SESSION = "OpenSession"
+    CANCEL = "CancelSession"
+    RELEASE = "ReleaseSession"
+    METRICS = "GetMetrics"
+
+
+class StatusCode(str, Enum):
+    OK = "OK"
+    INVALID_ARGUMENT = "INVALID_ARGUMENT"
+    UNSUPPORTED_VERSION = "UNSUPPORTED_VERSION"
+    FAILED_PRECONDITION = "FAILED_PRECONDITION"
+    MALFORMED_LIFECYCLE = "MALFORMED_LIFECYCLE"
+    NOT_FOUND = "NOT_FOUND"
+    ALREADY_EXISTS = "ALREADY_EXISTS"
+    CANCELLED = "CANCELLED"
+    DEADLINE_EXCEEDED = "DEADLINE_EXCEEDED"
+    RESOURCE_EXHAUSTED = "RESOURCE_EXHAUSTED"
+    UNAUTHENTICATED = "UNAUTHENTICATED"
+    PERMISSION_DENIED = "PERMISSION_DENIED"
+    DATA_LOSS = "DATA_LOSS"
+    UNAVAILABLE = "UNAVAILABLE"
+    INTERNAL = "INTERNAL"
+
+
+class LifecycleState(str, Enum):
+    OPEN = "OPEN"
+    PREFILLING = "PREFILLING"
+    DECODING = "DECODING"
+    CANCELLING = "CANCELLING"
+    CANCELLED = "CANCELLED"
+    RELEASING = "RELEASING"
+    RELEASED = "RELEASED"
+    FAILED = "FAILED"
+
+
+class CacheExpectation(str, Enum):
+    NONE = "NONE"
+    OPTIONAL = "OPTIONAL"
+    REQUIRED = "REQUIRED"
+
+
+class CacheResult(str, Enum):
+    NOT_REQUESTED = "NOT_REQUESTED"
+    HIT = "HIT"
+    MISS = "MISS"
+    INVALIDATED = "INVALIDATED"
+    STORED = "STORED"
+
+
+class SessionPhase(str, Enum):
+    PREFILL = "PREFILL"
+    DECODE = "DECODE"
+
+
+class LifecycleContractError(ValueError):
+    """A protocol violation represented by a structured status."""
+
+    def __init__(self, status: "StructuredStatus") -> None:
+        self.status = status
+        super().__init__(status.message)
+
+
+@dataclass(frozen=True)
+class StructuredStatus:
+    code: StatusCode
+    message: str
+    retryable: bool = False
+    details: dict[str, str] = field(default_factory=dict)
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "code": self.code.value,
+            "message": self.message,
+            "retryable": self.retryable,
+            "details": dict(sorted(self.details.items())),
+        }
+
+    @classmethod
+    def from_dict(cls, value: dict[str, Any]) -> "StructuredStatus":
+        return cls(
+            code=StatusCode(str(value["code"])),
+            message=str(value["message"]),
+            retryable=bool(value.get("retryable", False)),
+            details={str(k): str(v) for k, v in value.get("details", {}).items()},
+        )
+
+
+@dataclass(frozen=True)
+class DeadlinePolicy:
+    capability_seconds: float = 5.0
+    health_seconds: float = 2.0
+    session_open_seconds: float = 10.0
+    session_idle_seconds: float = 30.0
+    cancel_seconds: float = 2.0
+    release_seconds: float = 5.0
+    metrics_seconds: float = 5.0
+
+    def __post_init__(self) -> None:
+        if any(value <= 0 for value in self.__dict__.values()):
+            raise ValueError("all RPC deadlines must be positive")
+
+
+@dataclass(frozen=True)
+class TlsAuthHooks:
+    """Transport hooks only; Meshnet remains the identity and billing authority."""
+
+    tls_required: bool = True
+    server_name: str = "shard.meshnet"
+    client_certificate_hook: str | None = None
+    peer_identity_hook: str | None = None
+    auth_metadata_hook: str | None = None
+
+    def validate(self) -> None:
+        if self.tls_required and not self.server_name:
+            raise ValueError("TLS server_name is required when TLS is enabled")
+
+
+@dataclass(frozen=True)
+class FlowControlLimits:
+    max_inflight_frames: int = DEFAULT_MAX_INFLIGHT_FRAMES
+    max_inflight_bytes: int = DEFAULT_MAX_INFLIGHT_BYTES
+    max_frame_bytes: int = 1024 * 1024
+
+    def __post_init__(self) -> None:
+        if min(self.max_inflight_frames, self.max_inflight_bytes, self.max_frame_bytes) <= 0:
+            raise ValueError("flow-control limits must be positive")
+
+
+class FlowControl:
+    """Bounded sender window; callers block or fail instead of growing unbounded."""
+
+    def __init__(self, limits: FlowControlLimits = FlowControlLimits()) -> None:
+        self.limits = limits
+        self._condition = threading.Condition()
+        self._frames = 0
+        self._bytes = 0
+
+    @property
+    def outstanding(self) -> tuple[int, int]:
+        with self._condition:
+            return self._frames, self._bytes
+
+    def acquire(self, size: int, *, wait: bool = False, cancelled: Callable[[], bool] | None = None) -> None:
+        if size < 0 or size > self.limits.max_frame_bytes:
+            raise LifecycleContractError(StructuredStatus(
+                StatusCode.RESOURCE_EXHAUSTED, "frame exceeds bounded flow-control window",
+                details={"max_frame_bytes": str(self.limits.max_frame_bytes)},
+            ))
+        with self._condition:
+            while (
+                self._frames >= self.limits.max_inflight_frames
+                or self._bytes + size > self.limits.max_inflight_bytes
+            ):
+                if cancelled and cancelled():
+                    raise LifecycleContractError(StructuredStatus(StatusCode.CANCELLED, "flow-control wait cancelled"))
+                if not wait:
+                    raise LifecycleContractError(StructuredStatus(
+                        StatusCode.RESOURCE_EXHAUSTED, "flow-control window is full", retryable=True,
+                    ))
+                self._condition.wait(timeout=0.05)
+            self._frames += 1
+            self._bytes += size
+
+    def release(self, size: int) -> None:
+        with self._condition:
+            self._frames = max(0, self._frames - 1)
+            self._bytes = max(0, self._bytes - max(0, size))
+            self._condition.notify_all()
+
+
+class CancellationToken:
+    def __init__(self) -> None:
+        self._event = threading.Event()
+
+    @property
+    def cancelled(self) -> bool:
+        return self._event.is_set()
+
+    def cancel(self) -> None:
+        self._event.set()
+
+
+@dataclass(frozen=True)
+class CapabilityRequest:
+    schema_version: int = SCHEMA_VERSION
+
+
+@dataclass(frozen=True)
+class CapabilityResponse:
+    status: StructuredStatus
+    schema_version: int
+    supported_versions: tuple[int, ...]
+    artifact_fingerprint: str = ""
+    runtime_fingerprint: str = ""
+    shard_start: int = 0
+    shard_end: int = 0
+    effective_start: int = 0
+    max_sessions: int = 0
+    max_frame_bytes: int = DEFAULT_MAX_INFLIGHT_BYTES
+
+    def validate(self) -> None:
+        if self.status.code is StatusCode.OK and self.schema_version not in SUPPORTED_VERSIONS:
+            raise LifecycleContractError(StructuredStatus(
+                StatusCode.UNSUPPORTED_VERSION, "worker selected an unsupported schema version",
+            ))
+        if self.shard_end < self.shard_start or self.effective_start < self.shard_start:
+            raise LifecycleContractError(StructuredStatus(
+                StatusCode.INVALID_ARGUMENT, "invalid authoritative shard range",
+            ))
+
+
+@dataclass(frozen=True)
+class HealthRequest:
+    schema_version: int = SCHEMA_VERSION
+    include_metrics: bool = False
+
+
+@dataclass(frozen=True)
+class HealthResponse:
+    status: StructuredStatus
+    serving: bool
+    state: str
+    active_sessions: int = 0
+
+
+@dataclass(frozen=True)
+class SessionRequest:
+    schema_version: int
+    request_id: str
+    work_id: str
+    route_session: str
+    route_epoch: int
+    artifact_fingerprint: str
+    runtime_fingerprint: str
+    shard_start: int
+    shard_end: int
+    effective_start: int
+    cache_expectation: CacheExpectation = CacheExpectation.NONE
+    deadline_seconds: float = 30.0
+
+    def validate(self) -> None:
+        if self.schema_version not in SUPPORTED_VERSIONS:
+            raise LifecycleContractError(StructuredStatus(
+                StatusCode.UNSUPPORTED_VERSION, "unsupported session schema version",
+                details={"requested": str(self.schema_version), "supported": "1"},
+            ))
+        if not self.request_id or not self.work_id or not self.route_session:
+            raise LifecycleContractError(StructuredStatus(
+                StatusCode.INVALID_ARGUMENT, "request, work, and route-session IDs are required",
+            ))
+        if self.route_epoch < 0 or self.shard_start < 0 or self.shard_end <= self.shard_start:
+            raise LifecycleContractError(StructuredStatus(
+                StatusCode.INVALID_ARGUMENT, "invalid route epoch or shard range",
+            ))
+        if not self.shard_start <= self.effective_start <= self.shard_end:
+            raise LifecycleContractError(StructuredStatus(
+                StatusCode.INVALID_ARGUMENT, "effective start must be inside the shard range",
+            ))
+        if self.deadline_seconds <= 0:
+            raise LifecycleContractError(StructuredStatus(
+                StatusCode.INVALID_ARGUMENT, "session deadline must be positive",
+            ))
+
+
+@dataclass(frozen=True)
+class SessionFrame:
+    phase: SessionPhase
+    position: int
+    idempotency_step: int
+    payload: Any
+    cache_expectation: CacheExpectation = CacheExpectation.NONE
+    size_bytes: int = 0
+
+    def validate(self, limits: FlowControlLimits) -> None:
+        if self.position < 0 or self.idempotency_step < 0:
+            raise LifecycleContractError(StructuredStatus(
+                StatusCode.INVALID_ARGUMENT, "position and idempotency step must be non-negative",
+            ))
+        if self.size_bytes < 0 or self.size_bytes > limits.max_frame_bytes:
+            raise LifecycleContractError(StructuredStatus(
+                StatusCode.RESOURCE_EXHAUSTED, "session frame exceeds max_frame_bytes",
+            ))
+
+
+@dataclass(frozen=True)
+class SessionResult:
+    status: StructuredStatus
+    cache_result: CacheResult = CacheResult.NOT_REQUESTED
+    position: int = 0
+    idempotency_step: int = 0
+    payload: Any = None
+
+
+@dataclass(frozen=True)
+class CancelRequest:
+    schema_version: int
+    request_id: str
+    work_id: str
+    route_session: str
+    route_epoch: int
+    reason: str = ""
+
+
+@dataclass(frozen=True)
+class ReleaseRequest:
+    schema_version: int
+    request_id: str
+    work_id: str
+    route_session: str
+    route_epoch: int
+
+
+@dataclass(frozen=True)
+class MetricsRequest:
+    schema_version: int = SCHEMA_VERSION
+
+
+@dataclass(frozen=True)
+class MetricsResponse:
+    status: StructuredStatus
+    active_sessions: int
+    queued_frames: int
+    inflight_bytes: int
+    kv_entries: int
+    generated_tokens: int
+    cancelled_sessions: int
+
+
+@dataclass
+class SessionLifecycle:
+    """Fail-closed state machine for one Route Session Activation Seam."""
+
+    request: SessionRequest
+    state: LifecycleState = LifecycleState.OPEN
+    cancellation: CancellationToken = field(default_factory=CancellationToken)
+    last_idempotency_step: int = -1
+    _seen_steps: set[int] = field(default_factory=set, init=False, repr=False)
+
+    _ALLOWED: ClassVar[dict[LifecycleState, frozenset[LifecycleState]]] = {
+        LifecycleState.OPEN: frozenset({LifecycleState.PREFILLING, LifecycleState.CANCELLING, LifecycleState.RELEASING, LifecycleState.FAILED}),
+        LifecycleState.PREFILLING: frozenset({LifecycleState.PREFILLING, LifecycleState.DECODING, LifecycleState.CANCELLING, LifecycleState.RELEASING, LifecycleState.FAILED}),
+        LifecycleState.DECODING: frozenset({LifecycleState.DECODING, LifecycleState.CANCELLING, LifecycleState.RELEASING, LifecycleState.FAILED}),
+        LifecycleState.CANCELLING: frozenset({LifecycleState.CANCELLED, LifecycleState.RELEASING, LifecycleState.FAILED}),
+        LifecycleState.CANCELLED: frozenset({LifecycleState.RELEASING, LifecycleState.RELEASED}),
+        LifecycleState.RELEASING: frozenset({LifecycleState.RELEASED, LifecycleState.FAILED}),
+        LifecycleState.RELEASED: frozenset(),
+        LifecycleState.FAILED: frozenset({LifecycleState.RELEASING, LifecycleState.RELEASED}),
+    }
+
+    def _transition(self, target: LifecycleState) -> None:
+        if target not in self._ALLOWED[self.state]:
+            raise LifecycleContractError(StructuredStatus(
+                StatusCode.MALFORMED_LIFECYCLE,
+                f"cannot transition from {self.state.value} to {target.value}",
+                details={"state": self.state.value, "target": target.value},
+            ))
+        self.state = target
+
+    def apply(self, frame: SessionFrame) -> None:
+        frame.validate(FlowControlLimits())
+        if self.cancellation.cancelled and frame.phase is not SessionPhase.PREFILL:
+            raise LifecycleContractError(StructuredStatus(StatusCode.CANCELLED, "session cancellation propagated"))
+        if frame.idempotency_step in self._seen_steps:
+            raise LifecycleContractError(StructuredStatus(
+                StatusCode.ALREADY_EXISTS, "duplicate idempotency step", details={"step": str(frame.idempotency_step)},
+            ))
+        if frame.idempotency_step <= self.last_idempotency_step:
+            raise LifecycleContractError(StructuredStatus(
+                StatusCode.MALFORMED_LIFECYCLE, "idempotency steps must increase monotonically",
+            ))
+        target = LifecycleState.PREFILLING if frame.phase is SessionPhase.PREFILL else LifecycleState.DECODING
+        self._transition(target)
+        self._seen_steps.add(frame.idempotency_step)
+        self.last_idempotency_step = frame.idempotency_step
+
+    def cancel(self, reason: str = "") -> StructuredStatus:
+        if self.state in {LifecycleState.RELEASED, LifecycleState.RELEASING}:
+            return StructuredStatus(StatusCode.FAILED_PRECONDITION, "session is already releasing or released")
+        if self.state is LifecycleState.CANCELLED:
+            return StructuredStatus(StatusCode.OK, "session already cancelled")
+        self._transition(LifecycleState.CANCELLING)
+        self.cancellation.cancel()
+        self._transition(LifecycleState.CANCELLED)
+        return StructuredStatus(StatusCode.CANCELLED, reason or "session cancelled")
+
+    def release(self) -> StructuredStatus:
+        if self.state is LifecycleState.RELEASED:
+            return StructuredStatus(StatusCode.OK, "session already released")
+        self._transition(LifecycleState.RELEASING)
+        self.cancellation.cancel()
+        self._transition(LifecycleState.RELEASED)
+        return StructuredStatus(StatusCode.OK, "session released")
+
+
+@dataclass(frozen=True)
+class ShardRpcContract:
+    """Service/method and operational rules for generated gRPC bindings."""
+
+    schema: str = SCHEMA_NAME
+    version: int = SCHEMA_VERSION
+    methods: tuple[RpcName, ...] = tuple(RpcName)
+    deadlines: DeadlinePolicy = DeadlinePolicy()
+    flow_control: FlowControlLimits = FlowControlLimits()
+    tls_auth: TlsAuthHooks = TlsAuthHooks()
+
+    def validate(self) -> None:
+        if self.schema != SCHEMA_NAME or self.version not in SUPPORTED_VERSIONS:
+            raise LifecycleContractError(StructuredStatus(
+                StatusCode.UNSUPPORTED_VERSION, "unsupported Shard RPC contract version",
+            ))
+        if self.methods != tuple(RpcName):
+            raise LifecycleContractError(StructuredStatus(
+                StatusCode.FAILED_PRECONDITION, "contract must expose the complete lifecycle RPC set",
+            ))
+        self.tls_auth.validate()
+
+    def to_bytes(self) -> bytes:
+        self.validate()
+        data = {
+            "schema": self.schema,
+            "version": self.version,
+            "methods": [method.value for method in self.methods],
+            "deadlines": self.deadlines.__dict__,
+            "flow_control": self.flow_control.__dict__,
+            "tls_auth": self.tls_auth.__dict__,
+        }
+        return json.dumps(data, sort_keys=True, separators=(",", ":")).encode("utf-8")
+
+    @classmethod
+    def from_bytes(cls, raw: bytes) -> "ShardRpcContract":
+        data = json.loads(raw)
+        if int(data.get("version", -1)) not in SUPPORTED_VERSIONS:
+            raise LifecycleContractError(StructuredStatus(
+                StatusCode.UNSUPPORTED_VERSION, "unsupported Shard RPC contract version",
+            ))
+        methods = tuple(RpcName(item) for item in data.get("methods", ()))
+        contract = cls(
+            schema=str(data.get("schema", "")),
+            version=int(data["version"]),
+            methods=methods,
+            deadlines=DeadlinePolicy(**data.get("deadlines", {})),
+            flow_control=FlowControlLimits(**data.get("flow_control", {})),
+            tls_auth=TlsAuthHooks(**data.get("tls_auth", {})),
+        )
+        contract.validate()
+        return contract

tests/test_shard_lifecycle.py

@@ -0,0 +1,182 @@
+"""DGR-022 lifecycle and structured-status contract tests."""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from meshnet_node.shard_lifecycle import (
+    CacheExpectation,
+    CacheResult,
+    CapabilityResponse,
+    CancellationToken,
+    FlowControl,
+    FlowControlLimits,
+    LifecycleContractError,
+    LifecycleState,
+    RpcName,
+    SessionFrame,
+    SessionLifecycle,
+    SessionPhase,
+    SessionRequest,
+    ShardRpcContract,
+    StatusCode,
+    StructuredStatus,
+)
+
+
+def _request(**overrides) -> SessionRequest:
+    values = dict(
+        schema_version=1,
+        request_id="request-1",
+        work_id="work-1",
+        route_session="route-1",
+        route_epoch=3,
+        artifact_fingerprint="artifact-sha",
+        runtime_fingerprint="runtime-sha",
+        shard_start=0,
+        shard_end=8,
+        effective_start=0,
+        cache_expectation=CacheExpectation.OPTIONAL,
+    )
+    values.update(overrides)
+    return SessionRequest(**values)
+
+
+def _frame(phase: SessionPhase, step: int, position: int = 0) -> SessionFrame:
+    return SessionFrame(
+        phase=phase,
+        position=position,
+        idempotency_step=step,
+        payload=b"frame",
+        size_bytes=5,
+    )
+
+
+def test_contract_roundtrip_exposes_all_lifecycle_rpcs_and_operational_hooks():
+    contract = ShardRpcContract()
+    restored = ShardRpcContract.from_bytes(contract.to_bytes())
+
+    assert restored == contract
+    assert restored.methods == tuple(RpcName)
+    assert restored.deadlines.session_idle_seconds > restored.deadlines.health_seconds
+    assert restored.flow_control.max_inflight_frames == 32
+    assert restored.tls_auth.tls_required is True
+
+
+def test_supported_version_roundtrip_and_status_are_stable():
+    status = StructuredStatus(StatusCode.RESOURCE_EXHAUSTED, "window full", retryable=True, details={"limit": "32"})
+    restored = StructuredStatus.from_dict(status.to_dict())
+
+    assert restored == status
+    assert CapabilityResponse(
+        status=StructuredStatus(StatusCode.OK, "ready"),
+        schema_version=1,
+        supported_versions=(1,),
+    ).validate() is None
+
+
+@pytest.mark.parametrize("version", [0, 2, 99])
+def test_unsupported_versions_fail_closed(version):
+    with pytest.raises(LifecycleContractError) as exc:
+        ShardRpcContract.from_bytes(json.dumps({
+            "schema": "meshnet.shard-runtime",
+            "version": version,
+            "methods": [method.value for method in RpcName],
+        }).encode())
+    assert exc.value.status.code is StatusCode.UNSUPPORTED_VERSION
+
+    with pytest.raises(LifecycleContractError) as exc:
+        _request(schema_version=version).validate()
+    assert exc.value.status.code is StatusCode.UNSUPPORTED_VERSION
+
+
+def test_malformed_lifecycle_transitions_fail_closed():
+    session = SessionLifecycle(_request())
+
+    with pytest.raises(LifecycleContractError) as exc:
+        session.apply(_frame(SessionPhase.DECODE, 0))
+    assert exc.value.status.code is StatusCode.MALFORMED_LIFECYCLE
+    assert session.state is LifecycleState.OPEN
+
+    session.apply(_frame(SessionPhase.PREFILL, 0))
+    session.apply(_frame(SessionPhase.DECODE, 1, position=1))
+    assert session.state is LifecycleState.DECODING
+
+    with pytest.raises(LifecycleContractError) as exc:
+        session.apply(_frame(SessionPhase.PREFILL, 2))
+    assert exc.value.status.code is StatusCode.MALFORMED_LIFECYCLE
+
+
+def test_duplicate_and_non_monotonic_idempotency_steps_are_rejected():
+    session = SessionLifecycle(_request())
+    session.apply(_frame(SessionPhase.PREFILL, 4))
+
+    with pytest.raises(LifecycleContractError) as exc:
+        session.apply(_frame(SessionPhase.PREFILL, 4))
+    assert exc.value.status.code is StatusCode.ALREADY_EXISTS
+
+    with pytest.raises(LifecycleContractError) as exc:
+        session.apply(_frame(SessionPhase.PREFILL, 3))
+    assert exc.value.status.code is StatusCode.MALFORMED_LIFECYCLE
+
+
+def test_cancel_propagates_to_waiters_and_release_is_idempotent():
+    session = SessionLifecycle(_request())
+    assert session.cancel("client disconnected").code is StatusCode.CANCELLED
+    assert session.cancellation.cancelled is True
+    assert session.state is LifecycleState.CANCELLED
+
+    with pytest.raises(LifecycleContractError) as exc:
+        session.apply(_frame(SessionPhase.DECODE, 0))
+    assert exc.value.status.code is StatusCode.CANCELLED
+
+    assert session.release().code is StatusCode.OK
+    assert session.state is LifecycleState.RELEASED
+    assert session.release().code is StatusCode.OK
+
+
+def test_bounded_flow_control_rejects_oversize_and_full_windows():
+    flow = FlowControl(FlowControlLimits(max_inflight_frames=1, max_inflight_bytes=10, max_frame_bytes=8))
+    flow.acquire(8)
+
+    with pytest.raises(LifecycleContractError) as exc:
+        flow.acquire(1)
+    assert exc.value.status.code is StatusCode.RESOURCE_EXHAUSTED
+    assert exc.value.status.retryable is True
+
+    with pytest.raises(LifecycleContractError) as exc:
+        flow.acquire(9)
+    assert exc.value.status.code is StatusCode.RESOURCE_EXHAUSTED
+
+    flow.release(8)
+    assert flow.outstanding == (0, 0)
+
+
+def test_flow_control_wait_honours_cancellation():
+    token = CancellationToken()
+    flow = FlowControl(FlowControlLimits(max_inflight_frames=1, max_inflight_bytes=10, max_frame_bytes=8))
+    flow.acquire(4)
+    token.cancel()
+
+    with pytest.raises(LifecycleContractError) as exc:
+        flow.acquire(1, wait=True, cancelled=lambda: token.cancelled)
+    assert exc.value.status.code is StatusCode.CANCELLED
+
+
+def test_cache_expectation_and_result_are_explicit():
+    request = _request(cache_expectation=CacheExpectation.REQUIRED)
+    assert request.cache_expectation is CacheExpectation.REQUIRED
+    assert CacheResult.MISS.value == "MISS"
+
+
+def test_invalid_contract_methods_and_tls_hooks_fail_closed():
+    with pytest.raises(LifecycleContractError) as exc:
+        ShardRpcContract(methods=(RpcName.HEALTH,)).validate()
+    assert exc.value.status.code is StatusCode.FAILED_PRECONDITION
+
+    with pytest.raises(ValueError, match="server_name"):
+        ShardRpcContract(
+            tls_auth=type(ShardRpcContract().tls_auth)(tls_required=True, server_name="")
+        ).validate()