diff --git a/rfcs/008-environment-auto-validation.md b/rfcs/008-environment-auto-validation.md
index ea0213c8bf..e8ddff0a7a 100644
--- a/rfcs/008-environment-auto-validation.md
+++ b/rfcs/008-environment-auto-validation.md
@@ -197,9 +197,10 @@ configuration (model, version, params) in the manifest; the oracle check becomes
 bit-exact. RFC 004 rubrics are **leveraged, not required**: the contract stays spec-neutral
 (graders read the manifest), but for the served OpenEnv format the rubric tree is the native
 satisfaction path — `LLMJudge` is the in-repo `llm_judged` implementation, and the
-introspectability and reward-attribution graders read `named_rubrics()` / `state_dict()` /
-per-child scores. Judge pinning stays a *manifest* declaration because the rubric object does not
-serialize model/version/params today.
+introspectability and reward-attribution graders read `named_rubrics()`, explicit
+`validation_config()` and fresh per-child scores. `state_dict()` is never serialized
+as validation configuration. Judge pinning stays a *manifest* declaration because
+the rubric object does not serialize model/version/params today.
 
 Tolerances, margins, and variance bounds are author-declared in the manifest, **bounded by the
 versioned severity policy**, and carried verbatim in reports so hubs can apply stricter ceilings.
@@ -624,7 +625,35 @@ attached to the **same** replay connection, reject unauthorized/cross-session
 reads and never expose telemetry as agent MCP tools. A second WebSocket creates
 another environment and cannot supply evidence for the measured instance.
 A validator transcript alone cannot pass subject-emitted trajectory recording.
-These authorization requirements do not add new public wire messages in this slice.
+The initial contracts slice added no public wire messages. PR4 implements the
+following opt-in protocol.
+
+#### Session telemetry protocol (PR4)
+
+The server opts in only when `OPENENV_VALIDATION_TOKEN` is provisioned explicitly
+by the validation supervisor. On its existing simulation `/ws` connection, the
+collector sends `validation_open` with `data: {schema_version: 1, token: ...}`.
+The reply returns a random capability bound to that connection. Subsequent
+`validation_read` messages carry that capability and return a `validation`
+snapshot. Disabled, unauthorized and cross-connection requests fail without
+echoing credentials. Closing the connection destroys the capability. MCP and
+production endpoints never expose these operations. Authentication exchanges
+are excluded from persisted evidence.
+
+Snapshots identify requested and actually forwarded seed arguments; successful
+reset alone is not acceptance. They include a named rubric tree rooted at `root`,
+explicit safe configuration, and per-step attribution with operation-local
+evaluation flags. Unevaluated gated children never reuse an earlier score.
+Stock container aggregation is named explicitly; custom rubrics may supply
+`validation_config()` to expose public JSON configuration. Arbitrary attributes
+and `state_dict()` are never serialized as configuration.
+
+The subject server also emits a bounded record of the reset/step/state request
+and response envelopes it executed. The validator captures the wire separately
+and later compares the two. The record is bound to the authenticated session,
+limited to 100 steps/202 operations and 8 MiB, and marks truncation explicitly.
+Missing configuration, attribution or records cannot be inferred from other
+successful operations. This transport introduces no new passing grader by itself.
 
 Applicability predicates must distinguish empty declared sets from absent
 capabilities. Missing subject features, missing provider support and checks whose
diff --git a/src/openenv/core/env_server/http_server.py b/src/openenv/core/env_server/http_server.py
index 5fa105763a..ca8ab3e999 100644
--- a/src/openenv/core/env_server/http_server.py
+++ b/src/openenv/core/env_server/http_server.py
@@ -14,6 +14,7 @@
 import json
 import logging
 import os
+import secrets
 import time
 import uuid
 from concurrent.futures import ThreadPoolExecutor
@@ -62,6 +63,17 @@
 )
 from .route_config import GetEndpointConfig, register_get_endpoints
 from .serialization import deserialize_action, serialize_observation
+from .session_telemetry import (
+    rubric_counts,
+    rubric_snapshot,
+    SeedAcceptance,
+    SessionTelemetry,
+    ValidationOpenedData,
+    ValidationOpenedResponse,
+    ValidationOpenMessage,
+    ValidationReadMessage,
+    ValidationResponse,
+)
 from .types import (
     Action,
     ConcurrencyConfig,
@@ -910,6 +922,12 @@ def register_routes(
                     f"Invalid mode: '{mode}'. Must be one of: {valid_modes}"
                 )
 
+        # Only explicitly provisioned simulation servers accept validation controls.
+        validation_token = os.environ.get("OPENENV_VALIDATION_TOKEN", "")
+        validation_enabled = (
+            mode == ServerMode.SIMULATION and 32 <= len(validation_token) <= 256
+        )
+
         # Wire up session lifecycle via app events
         server_ref = self
 
@@ -1867,6 +1885,8 @@ async def websocket_endpoint(websocket: WebSocket):
             session_env = None
             owns_session = False
             attached_session = False
+            telemetry = None
+            operations_started = False
 
             try:
                 requested_session_id = websocket.query_params.get("session_id")
@@ -1936,6 +1956,64 @@ async def websocket_endpoint(websocket: WebSocket):
 
                         msg_type = message_dict.get("type", "")
 
+                        if msg_type in {"validation_open", "validation_read"}:
+                            # Do not return Pydantic input/error details: these messages
+                            # contain credentials and must never echo or log them.
+                            try:
+                                if not validation_enabled or not owns_session:
+                                    raise ValueError("Validation unavailable")
+                                if msg_type == "validation_open":
+                                    auth = ValidationOpenMessage.model_validate(
+                                        message_dict
+                                    )
+                                    if telemetry is not None or operations_started:
+                                        raise ValueError("Validation already started")
+                                    if not secrets.compare_digest(
+                                        validation_token.encode(),
+                                        auth.data.token.get_secret_value().encode(),
+                                    ):
+                                        raise ValueError("Unauthorized")
+                                    telemetry = SessionTelemetry()
+                                    response = ValidationOpenedResponse(
+                                        data=ValidationOpenedData(
+                                            capability=telemetry.capability
+                                        )
+                                    )
+                                else:
+                                    auth = ValidationReadMessage.model_validate(
+                                        message_dict
+                                    )
+                                    if telemetry is None or not telemetry.authorized(
+                                        auth.data.capability.get_secret_value()
+                                    ):
+                                        raise ValueError("Unauthorized")
+                                    response = ValidationResponse(
+                                        data=telemetry.snapshot
+                                    )
+                            except Exception:
+                                response = WSErrorResponse(
+                                    data={
+                                        "message": "Validation unavailable or unauthorized",
+                                        "code": WSErrorCode.VALIDATION_ERROR,
+                                    }
+                                )
+                            await websocket.send_text(response.model_dump_json())
+                            continue
+
+                        seed_acceptance = None
+                        before_scores = None
+                        if msg_type in {"reset", "step", "state"}:
+                            operations_started = True
+                        if telemetry is not None and msg_type == "step":
+                            try:
+                                before_scores = rubric_counts(
+                                    getattr(session_env, "rubric", None)
+                                )
+                            except Exception:
+                                telemetry.snapshot.rubric_error = (
+                                    "Rubric introspection unavailable"
+                                )
+
                         try:
                             match msg_type:
                                 case "reset":
@@ -1967,6 +2045,12 @@ async def websocket_endpoint(websocket: WebSocket):
                                             )
                                         )
 
+                                    if telemetry is not None:
+                                        seed_acceptance = SeedAcceptance(
+                                            requested="seed" in msg.data,
+                                            value=msg.data.get("seed"),
+                                            accepted="seed" in valid_kwargs,
+                                        )
                                     self._update_session_activity(session_id)
 
                                     response = WSObservationResponse(
@@ -2044,9 +2128,39 @@ async def websocket_endpoint(websocket: WebSocket):
                                         }
                                     )
 
+                            if telemetry is not None and msg_type in {
+                                "reset",
+                                "step",
+                                "state",
+                            }:
+                                nodes = None
+                                try:
+                                    if msg_type in {"reset", "step"}:
+                                        nodes = rubric_snapshot(
+                                            getattr(session_env, "rubric", None),
+                                            before_scores,
+                                        )
+                                except Exception:
+                                    nodes = []
+                                    telemetry.snapshot.rubric_error = (
+                                        "Rubric introspection unavailable"
+                                    )
+                                telemetry.append(
+                                    msg_type,
+                                    message_dict,
+                                    response.model_dump(mode="json"),
+                                    seed=seed_acceptance,
+                                    rubric=nodes,
+                                )
                             await websocket.send_text(response.model_dump_json())
 
                         except ValidationError as e:
+                            if telemetry is not None and msg_type in {
+                                "reset",
+                                "step",
+                                "state",
+                            }:
+                                telemetry.incomplete("Session operation failed")
                             error_resp = WSErrorResponse(
                                 data={
                                     "message": "Invalid message",
@@ -2056,6 +2170,12 @@ async def websocket_endpoint(websocket: WebSocket):
                             )
                             await websocket.send_text(error_resp.model_dump_json())
                         except Exception as e:
+                            if telemetry is not None and msg_type in {
+                                "reset",
+                                "step",
+                                "state",
+                            }:
+                                telemetry.incomplete("Session operation failed")
                             error_resp = WSErrorResponse(
                                 data={
                                     "message": str(e),
diff --git a/src/openenv/core/env_server/session_telemetry.py b/src/openenv/core/env_server/session_telemetry.py
new file mode 100644
index 0000000000..fe9c3eb0f4
--- /dev/null
+++ b/src/openenv/core/env_server/session_telemetry.py
@@ -0,0 +1,241 @@
+# SPDX-License-Identifier: BSD-3-Clause
+
+"""Bounded, opt-in evidence emitted by the subject's replay session."""
+
+import json
+import secrets
+from typing import Any, Literal
+
+from openenv.core.rubrics.base import Rubric
+from openenv.core.rubrics.containers import Gate, Sequential, WeightedSum
+from pydantic import BaseModel, ConfigDict, Field, SecretStr
+
+MAX_TELEMETRY_BYTES = 8 * 1024 * 1024
+MAX_TRAJECTORY_ACTIONS = 100
+MAX_TRAJECTORY_RECORDS = 202
+MAX_RUBRIC_NODES = 128
+
+
+class _WireModel(BaseModel):
+    model_config = ConfigDict(extra="forbid", strict=True)
+
+
+class ValidationOpenData(_WireModel):
+    schema_version: Literal[1]
+    token: SecretStr = Field(min_length=32, max_length=256)
+
+
+class ValidationOpenMessage(_WireModel):
+    type: Literal["validation_open"]
+    data: ValidationOpenData
+
+
+class ValidationReadData(_WireModel):
+    schema_version: Literal[1]
+    capability: SecretStr = Field(min_length=32, max_length=256)
+
+
+class ValidationReadMessage(_WireModel):
+    type: Literal["validation_read"]
+    data: ValidationReadData
+
+
+class ValidationOpenedData(_WireModel):
+    schema_version: Literal[1] = 1
+    capability: str
+
+
+class ValidationOpenedResponse(_WireModel):
+    type: Literal["validation_open"] = "validation_open"
+    data: ValidationOpenedData
+
+
+class SeedAcceptance(_WireModel):
+    requested: bool
+    value: Any = None
+    accepted: bool
+
+
+class RubricNode(_WireModel):
+    name: str
+    class_name: str
+    children: list[str]
+    aggregation: Literal["weighted_sum", "sequential", "gate", "leaf", "unknown"]
+    config: dict[str, Any]
+    config_available: bool
+    score: Any = None
+    evaluated: bool = False
+
+
+class StepAttribution(_WireModel):
+    step_index: int
+    rubric: list[RubricNode]
+
+
+class TrajectoryRecord(_WireModel):
+    operation: Literal["reset", "step", "state"]
+    request: dict[str, Any]
+    response: dict[str, Any]
+
+
+class SubjectTrajectory(_WireModel):
+    schema_version: Literal[1] = 1
+    source: Literal["openenv-server"] = "openenv-server"
+    records: list[TrajectoryRecord] = Field(default_factory=list)
+    complete: bool = True
+    reason: str | None = None
+
+
+class ValidationSnapshot(_WireModel):
+    schema_version: Literal[1] = 1
+    seed: SeedAcceptance | None = None
+    rubric: list[RubricNode] = Field(default_factory=list)
+    rubric_error: str | None = None
+    attribution: list[StepAttribution] = Field(default_factory=list)
+    trajectory: SubjectTrajectory = Field(default_factory=SubjectTrajectory)
+
+
+class ValidationResponse(_WireModel):
+    type: Literal["validation"] = "validation"
+    data: ValidationSnapshot
+
+
+def _rubrics(root: Rubric | None) -> list[tuple[str, Rubric]]:
+    """Visit the named tree with finite size, rejecting cycles and aliases."""
+    if root is None:
+        return []
+    result, pending, seen = [], [("root", root)], set()
+    while pending:
+        name, rubric = pending.pop()
+        if id(rubric) in seen or len(result) >= MAX_RUBRIC_NODES:
+            raise ValueError("Rubric tree is cyclic, shared, or exceeds 128 nodes")
+        seen.add(id(rubric))
+        result.append((name, rubric))
+        children = list(rubric.named_children())
+        if any(not key or "." in key for key, _ in children):
+            raise ValueError("Rubric child names must be nonempty path segments")
+        pending.extend((f"{name}.{key}", child) for key, child in reversed(children))
+    return result
+
+
+def rubric_counts(root: Rubric | None) -> dict[int, int]:
+    return {id(r): r._evaluation_count for _, r in _rubrics(root)}
+
+
+def rubric_snapshot(
+    root: Rubric | None, before: dict[int, int] | None = None
+) -> list[RubricNode]:
+    nodes = []
+    for name, rubric in _rubrics(root):
+        children = [f"{name}.{key}" for key, _ in rubric.named_children()]
+        aggregation = "unknown" if children else "leaf"
+        config = None
+        # Exact types: a subclass can override scoring, so its semantics are unknown.
+        if type(rubric) is WeightedSum:
+            aggregation, config = "weighted_sum", {"weights": list(rubric._weights)}
+        elif type(rubric) is Sequential:
+            aggregation, config = "sequential", {}
+        elif type(rubric) is Gate:
+            aggregation, config = "gate", {"threshold": rubric.threshold}
+        else:
+            config = rubric.validation_config()
+        evaluated = before is not None and rubric._evaluation_count > before.get(
+            id(rubric), 0
+        )
+        nodes.append(
+            RubricNode(
+                name=name,
+                class_name=f"{type(rubric).__module__}.{type(rubric).__qualname__}",
+                children=children,
+                aggregation=aggregation,
+                config={} if config is None else config,
+                config_available=config is not None,
+                score=rubric.last_score if evaluated else None,
+                evaluated=evaluated,
+            )
+        )
+    json.dumps([node.model_dump() for node in nodes], allow_nan=False)
+    return nodes
+
+
+class SessionTelemetry:
+    """Own one socket's capability and a detached record of completed operations."""
+
+    def __init__(self):
+        self.capability = secrets.token_urlsafe(32)
+        self.snapshot = ValidationSnapshot()
+        self._bytes = 0
+        self._steps = 0
+
+    def authorized(self, capability: str) -> bool:
+        return secrets.compare_digest(self.capability.encode(), capability.encode())
+
+    def append(
+        self,
+        operation: Literal["reset", "step", "state"],
+        request: dict[str, Any],
+        response: dict[str, Any],
+        *,
+        seed: SeedAcceptance | None = None,
+        rubric: list[RubricNode] | None = None,
+    ) -> None:
+        if not self.snapshot.trajectory.complete:
+            return
+        try:
+            # JSON round-trip detaches mutable observations/actions from the record.
+            record = TrajectoryRecord.model_validate(
+                json.loads(
+                    json.dumps(
+                        {
+                            "operation": operation,
+                            "request": request,
+                            "response": response,
+                        },
+                        allow_nan=False,
+                    )
+                )
+            )
+            attribution = (
+                StepAttribution(step_index=self._steps, rubric=rubric or [])
+                if operation == "step"
+                else None
+            )
+            byte_count = len(record.model_dump_json().encode())
+            if attribution is not None:
+                byte_count += len(attribution.model_dump_json().encode())
+            byte_count += (
+                len(seed.model_dump_json().encode()) if seed is not None else 0
+            )
+            # Reserve room for the current rubric, envelope, and seed metadata.
+            current_bytes = len(
+                json.dumps(
+                    [
+                        node.model_dump()
+                        for node in (self.snapshot.rubric if rubric is None else rubric)
+                    ],
+                    allow_nan=False,
+                ).encode()
+            )
+            if (
+                len(self.snapshot.trajectory.records) >= MAX_TRAJECTORY_RECORDS
+                or operation == "step"
+                and self._steps >= MAX_TRAJECTORY_ACTIONS
+                or self._bytes + byte_count + current_bytes + 4096 > MAX_TELEMETRY_BYTES
+            ):
+                self.incomplete("Session telemetry budget exceeded")
+                return
+            self._bytes += byte_count
+            self.snapshot.trajectory.records.append(record)
+            if seed is not None:
+                self.snapshot.seed = seed.model_copy(deep=True)
+            if rubric is not None:
+                self.snapshot.rubric = [node.model_copy(deep=True) for node in rubric]
+            if attribution is not None:
+                self.snapshot.attribution.append(attribution.model_copy(deep=True))
+                self._steps += 1
+        except (TypeError, ValueError):
+            self.incomplete("Session telemetry contains non-JSON data")
+
+    def incomplete(self, reason: str) -> None:
+        self.snapshot.trajectory.complete = False
+        self.snapshot.trajectory.reason = reason
diff --git a/src/openenv/core/rubrics/base.py b/src/openenv/core/rubrics/base.py
index 96bf880195..276de7b3c4 100644
--- a/src/openenv/core/rubrics/base.py
+++ b/src/openenv/core/rubrics/base.py
@@ -46,6 +46,7 @@ def __init__(self):
         object.__setattr__(self, "_forward_hooks", [])
         object.__setattr__(self, "_forward_pre_hooks", [])
         object.__setattr__(self, "last_score", None)
+        object.__setattr__(self, "_evaluation_count", 0)
 
     def __setattr__(self, name: str, value: Any) -> None:
         # Auto-register child rubrics when assigned as attributes
@@ -118,6 +119,7 @@ def _run_leading_sync_pre_hooks(self, action: Any, observation: Any) -> int:
     def _finish_forward(self, action: Any, observation: Any, result: float) -> float:
         """Store the result and run post-forward hooks synchronously."""
         self.last_score = result
+        self._evaluation_count += 1
 
         # Post-forward hooks
         for hook in self._forward_hooks:
@@ -130,6 +132,7 @@ async def _finish_forward_async(
     ) -> float:
         """Store the result and run post-forward hooks from an async call path."""
         self.last_score = result
+        self._evaluation_count += 1
 
         # Post-forward hooks
         for hook in self._forward_hooks:
@@ -242,6 +245,15 @@ def reset(self) -> None:
         for child in self.children():
             child.reset()
 
+    def validation_config(self) -> Optional[Dict[str, Any]]:
+        """Return explicitly public configuration for authorized validation.
+
+        Override to expose JSON configuration without credentials or runtime state.
+        ``None`` means configuration introspection is unavailable. Stock aggregation
+        containers expose their weights/threshold directly through the server.
+        """
+        return None
+
     def state_dict(self) -> Dict[str, Any]:
         """Serialize rubric configuration for checkpointing."""
         return {}
diff --git a/src/openenv/validation/runner.py b/src/openenv/validation/runner.py
index f3ade58d47..0645a07320 100644
--- a/src/openenv/validation/runner.py
+++ b/src/openenv/validation/runner.py
@@ -3,6 +3,7 @@
 import hashlib
 import json
 import os
+import secrets
 import stat
 import time
 import uuid
@@ -300,6 +301,7 @@ def _runtime(subject, graders, *, skip_build, provider):
             resources=manifest.resources,
             network=manifest.network,
             run_id="validation-" + uuid.uuid4().hex,
+            env_vars={"OPENENV_VALIDATION_TOKEN": secrets.token_urlsafe(32)},
         )
         attempted = True
         image_ref = provider.build(subject.root, manifest.execution)
@@ -319,6 +321,7 @@ def _runtime(subject, graders, *, skip_build, provider):
             running.base_url,
             plan,
             episode_timeout_s=manifest.resources.episode_timeout_s,
+            validation_token=spec.env_vars["OPENENV_VALIDATION_TOKEN"],
         )
         # Protocol collection must finish before its dependent contract checks run.
         if evidence.failure_reason:
@@ -341,6 +344,8 @@ def _runtime(subject, graders, *, skip_build, provider):
                         plan,
                         episode_timeout_s=manifest.resources.episode_timeout_s,
                         episodes=2,
+                        validation_token=spec.env_vars["OPENENV_VALIDATION_TOKEN"],
+                        request_telemetry=False,
                     )
                 except RuntimeCollectionUnsupported as exc:
                     sequential = exc.evidence
@@ -350,6 +355,8 @@ def _runtime(subject, graders, *, skip_build, provider):
                         running.base_url,
                         plan,
                         episode_timeout_s=manifest.resources.episode_timeout_s,
+                        validation_token=spec.env_vars["OPENENV_VALIDATION_TOKEN"],
+                        request_telemetry=False,
                     )
                 except RuntimeCollectionUnsupported as exc:
                     fresh = exc.evidence
diff --git a/src/openenv/validation/runtime/artifacts.py b/src/openenv/validation/runtime/artifacts.py
index ead3fa87fa..9c4cef0468 100644
--- a/src/openenv/validation/runtime/artifacts.py
+++ b/src/openenv/validation/runtime/artifacts.py
@@ -106,6 +106,7 @@ def _recording(evidence, trace_file):
             "omitted_trace_fields": omitted_fields,
             "failure_phase": evidence.failure_phase,
             "failure_reason": evidence.failure_reason,
+            "telemetry_error": evidence.telemetry_error,
             "complete": evidence.failure_phase is None
             and evidence.failure_reason is None,
         }
@@ -207,6 +208,21 @@ def write_runtime_bundle(
         trace, metadata = _recording(recording, trace_file)
         if metadata is not None:
             metadata["session"] = session
+            if recording.telemetry_json is not None:
+                try:
+                    telemetry = loads(
+                        recording.telemetry_json, object_pairs_hook=_unique_object
+                    )
+                    telemetry_file = (
+                        "session-telemetry.json"
+                        if session == "collector"
+                        else f"{session}-telemetry.json"
+                    )
+                    files[telemetry_file] = telemetry
+                    metadata["redacted"] |= _redact(telemetry) != telemetry
+                except (ValueError, ArithmeticError, RecursionError):
+                    metadata["telemetry_error"] = "malformed telemetry omitted"
+                    metadata["redacted"] = True
         files[trace_file] = trace
         files[f"{session}-evidence.json"] = metadata
     probes = []
diff --git a/src/openenv/validation/runtime/collector.py b/src/openenv/validation/runtime/collector.py
index a3fe86e8af..34007649aa 100644
--- a/src/openenv/validation/runtime/collector.py
+++ b/src/openenv/validation/runtime/collector.py
@@ -6,6 +6,7 @@
 from urllib.parse import urlsplit, urlunsplit
 
 import httpx
+from websockets.exceptions import ConnectionClosed
 from websockets.sync.client import connect
 
 from ...core.env_server.types import WSErrorCode
@@ -17,6 +18,16 @@
 MAX_TRACE_BYTES = 8 * 1024 * 1024
 
 
+def _server_error_code(response):
+    if isinstance(response, dict) and response.get("type") == "error":
+        data = response.get("data")
+        code = data.get("code") if isinstance(data, dict) else None
+        # Only protocol constants are safe diagnostics, never subject error text.
+        if isinstance(code, str) and code in {item.value for item in WSErrorCode}:
+            return code
+    return None
+
+
 def _bounded_call(connection, operation, timeout_s):
     if timeout_s <= 0:
         raise TimeoutError("episode deadline exceeded")
@@ -70,6 +81,8 @@ def collect_runtime_evidence(
     episode_timeout_s: float,
     request_timeout_s: float | None = None,
     episodes: int = 1,
+    validation_token: str | None = None,
+    request_telemetry: bool = True,
 ) -> RuntimeEvidence:
     """
     Preserve schema and reset/step/state responses without model coercion.
@@ -94,6 +107,12 @@ def collect_runtime_evidence(
             episodes. Each later episode gets a fresh deadline and trace budget
             at its reset, so one episode never spends what another saved and the
             collection is bounded by `episodes` single-episode budgets.
+        validation_token (`str`, *optional*):
+            Run-scoped credential to omit from evidence; also authorizes optional
+            telemetry when `request_telemetry` is enabled.
+        request_telemetry (`bool`, *optional*, defaults to `True`):
+            Request session telemetry with the provided authorization. Disable
+            for scrub-only collection with the ordinary protocol byte limits.
 
     Returns:
         [`~openenv.validation.runtime.contracts.RuntimeEvidence`]: raw evidence.
@@ -103,6 +122,8 @@ def collect_runtime_evidence(
     schema_json = reset_schema_json = schema_response_json = None
     phase = "schema"
     trace_bytes = 0
+    telemetry_json = None
+    telemetry_error = None
     server_code = None
 
     def remaining() -> float:
@@ -140,14 +161,25 @@ def remaining() -> float:
                     if len(payload) + len(chunk) > MAX_MESSAGE_BYTES:
                         raise ValueError("schema exceeds size bound")
                     payload.extend(chunk)
-            schema_response_json = payload.decode("utf-8")
+            raw_schema = payload.decode("utf-8")
+            if not validation_token:
+                schema_response_json = raw_schema
+            # Only retain credential-aware schemas after both scans and parsing
+            # succeed. The raw body can include values lost to duplicate keys.
+            if validation_token and validation_token in raw_schema:
+                raise ValueError("schema contains validation credentials")
             schema = loads(payload)
+            if validation_token and validation_token in dumps(schema):
+                raise ValueError("schema contains validation credentials")
+            schema_response_json = raw_schema
             if not isinstance(schema, dict) or "observation" not in schema:
                 raise ValueError("missing observation schema")
-            schema_json = dumps(schema["observation"])
+            schema_payload = dumps(schema["observation"])
+            schema_json = schema_payload
 
             if "reset_observation" in schema:
-                reset_schema_json = dumps(schema["reset_observation"])
+                reset_schema_payload = dumps(schema["reset_observation"])
+                reset_schema_json = reset_schema_payload
 
         endpoint = urlsplit(base_url)
         ws_url = urlunsplit(
@@ -165,13 +197,100 @@ def remaining() -> float:
             proxy=None,
             open_timeout=remaining(),
             close_timeout=1,
-            max_size=MAX_MESSAGE_BYTES,
+            max_size=(
+                MAX_TRACE_BYTES
+                if validation_token and request_telemetry
+                else MAX_MESSAGE_BYTES
+            ),
             max_queue=1,
             compression=None,
         )
         complete = False
         try:
 
+            def receive_response(request):
+                nonlocal server_code
+                try:
+                    _bounded_call(
+                        connection, lambda: connection.send(request), remaining()
+                    )
+                except ConnectionClosed:
+                    # Session creation can fail before the first send. recv() still
+                    # delivers a queued error, but the failed send cannot succeed.
+                    try:
+                        raw = connection.recv(timeout=remaining())
+                        if (
+                            isinstance(raw, str)
+                            and len(raw.encode()) <= MAX_TRACE_BYTES
+                        ):
+                            server_code = _server_error_code(json.loads(raw))
+                    except Exception:
+                        # Preserve the original ConnectionClosed if diagnostics fail.
+                        pass
+                    raise
+                return connection.recv(timeout=remaining())
+
+            def telemetry_request(operation, data):
+                request = json.dumps({"type": operation, "data": data})
+                raw = receive_response(request)
+                if not isinstance(raw, str) or len(raw.encode()) > MAX_TRACE_BYTES:
+                    raise ValueError("invalid telemetry response")
+                response = loads(raw)
+                if not isinstance(response, dict):
+                    raise ValueError("invalid telemetry envelope")
+                return response
+
+            capability = None
+            if validation_token and request_telemetry:
+                phase = "validation_open"
+                try:
+                    response = telemetry_request(
+                        phase, {"schema_version": 1, "token": validation_token}
+                    )
+                except (ValueError, RecursionError) as exc:
+                    # A consumed malformed reply only invalidates optional telemetry.
+                    # Transport failure still aborts this same-session collection.
+                    telemetry_error = f"session telemetry failed ({type(exc).__name__})"
+                else:
+                    code = _server_error_code(response)
+                    if code in {
+                        WSErrorCode.FACTORY_ERROR,
+                        WSErrorCode.CAPACITY_REACHED,
+                        WSErrorCode.SESSION_ERROR,
+                    }:
+                        server_code = code
+                        raise ValueError("server returned a terminal error")
+                    data = response.get("data")
+                    if (
+                        response.get("type") == "validation_open"
+                        and isinstance(data, dict)
+                        and data.get("schema_version") == 1
+                        and isinstance(data.get("capability"), str)
+                        and 16 <= len(data["capability"]) <= 256
+                    ):
+                        capability = data["capability"]
+                    else:
+                        telemetry_error = (
+                            "session telemetry unavailable or authorization refused"
+                        )
+
+            def contains_credential(value):
+                pending = [value]
+                while pending:
+                    child = pending.pop()
+                    if isinstance(child, str):
+                        if any(
+                            secret and secret in child
+                            for secret in (validation_token, capability)
+                        ):
+                            return True
+                    elif isinstance(child, dict):
+                        pending.extend(child.keys())
+                        pending.extend(child.values())
+                    elif isinstance(child, list):
+                        pending.extend(child)
+                return False
+
             def exchange(operation: str, data: dict | None = None) -> dict:
                 nonlocal phase, trace_bytes, server_code
                 phase = operation
@@ -179,17 +298,29 @@ def exchange(operation: str, data: dict | None = None) -> dict:
                 if data is not None:
                     request["data"] = data
                 request_json = json.dumps(request, allow_nan=False)
-                _bounded_call(
-                    connection, lambda: connection.send(request_json), remaining()
-                )
-                raw = connection.recv(timeout=remaining())
+                raw = receive_response(request_json)
                 if not isinstance(raw, str):
                     raise ValueError("binary response is not the JSON protocol")
+                if len(raw.encode("utf-8")) > MAX_MESSAGE_BYTES:
+                    raise ValueError("response exceeds size bound")
                 trace_bytes += len(raw.encode("utf-8")) + len(
                     request_json.encode("utf-8")
                 )
                 if trace_bytes > MAX_TRACE_BYTES:
                     raise ValueError("episode trace exceeds size bound")
+                # Subjects can echo the authorization into arbitrary JSON fields.
+                # Check decoded values as well as raw text (which may be malformed).
+                if contains_credential(raw):
+                    raise ValueError("response contains validation credentials")
+                try:
+                    parsed = json.loads(raw)
+                except (ValueError, RecursionError):
+                    # Malformed JSON cannot be checked for escaped credentials.
+                    if validation_token:
+                        raise
+                    parsed = None
+                if contains_credential(parsed):
+                    raise ValueError("response contains validation credentials")
                 exchanges.append(
                     WireExchange(
                         operation=operation,
@@ -199,14 +330,7 @@ def exchange(operation: str, data: dict | None = None) -> dict:
                 )
                 response = json.loads(raw)
                 if isinstance(response, dict) and response.get("type") == "error":
-                    data = response.get("data")
-                    code = data.get("code") if isinstance(data, dict) else None
-                    # Only protocol constants are safe diagnostics; never echo an
-                    # arbitrary server message or a subject-defined error code.
-                    if isinstance(code, str) and code in {
-                        item.value for item in WSErrorCode
-                    }:
-                        server_code = code
+                    server_code = _server_error_code(response)
                     raise ValueError("server returned an error")
                 expected = "state" if operation == "state" else "observation"
                 if (
@@ -230,6 +354,24 @@ def exchange(operation: str, data: dict | None = None) -> dict:
                         break
                     observation = exchange("step", action)
                     exchange("state")
+            if capability:
+                phase = "validation_read"
+                try:
+                    response = telemetry_request(
+                        phase, {"schema_version": 1, "capability": capability}
+                    )
+                    if response.get("type") != "validation" or not isinstance(
+                        response.get("data"), dict
+                    ):
+                        raise ValueError("invalid telemetry envelope")
+                    if contains_credential(response["data"]):
+                        raise ValueError("telemetry contains validation credentials")
+                    snapshot = dumps(response["data"], ensure_ascii=False)
+                    if len(snapshot.encode("utf-8")) > MAX_TRACE_BYTES:
+                        raise ValueError("telemetry exceeds size bound")
+                    telemetry_json = snapshot
+                except Exception as exc:
+                    telemetry_error = f"session telemetry failed ({type(exc).__name__})"
             complete = True
         finally:
             # Teardown is best effort and cannot replace an in-episode failure
@@ -251,6 +393,8 @@ def exchange(operation: str, data: dict | None = None) -> dict:
             exchanges=tuple(exchanges),
             observation_schema_json=schema_json,
             schema_response_json=schema_response_json,
+            telemetry_json=telemetry_json,
+            telemetry_error=telemetry_error,
             reset_observation_schema_json=reset_schema_json,
         )
     except KeyboardInterrupt:
@@ -262,6 +406,8 @@ def exchange(operation: str, data: dict | None = None) -> dict:
                 reset_observation_schema_json=reset_schema_json,
                 failure_phase=phase,
                 failure_reason=f"{phase} failed (KeyboardInterrupt)",
+                telemetry_json=telemetry_json,
+                telemetry_error=telemetry_error,
             )
         ) from None
     except (ArithmeticError, RecursionError) as exc:
@@ -276,6 +422,8 @@ def exchange(operation: str, data: dict | None = None) -> dict:
                 schema_response_json=schema_response_json,
                 failure_phase=phase,
                 failure_reason=f"{phase} JSON exceeds validator support ({capability})",
+                telemetry_json=telemetry_json,
+                telemetry_error=telemetry_error,
             ),
             capability=capability,
         ) from None
@@ -288,4 +436,6 @@ def exchange(operation: str, data: dict | None = None) -> dict:
             reset_observation_schema_json=reset_schema_json,
             failure_phase=phase,
             failure_reason=f"{phase} failed ({server_code or type(exc).__name__})",
+            telemetry_json=telemetry_json,
+            telemetry_error=telemetry_error,
         )
diff --git a/src/openenv/validation/runtime/contracts.py b/src/openenv/validation/runtime/contracts.py
index d61d8d3879..3c11ee5db0 100644
--- a/src/openenv/validation/runtime/contracts.py
+++ b/src/openenv/validation/runtime/contracts.py
@@ -265,6 +265,10 @@ class RuntimeEvidence:
             Collector phase that failed; a truncated transcript cannot pass silently.
         failure_reason (`str`, *optional*):
             Credential-safe explanation of the collection failure.
+        telemetry_json (`str`, *optional*):
+            Subject-emitted session snapshot, independent of the wire transcript.
+        telemetry_error (`str`, *optional*):
+            Bounded telemetry failure without invalidating completed wire evidence.
         reset_observation_schema_json (`str`, *optional*):
             Explicit `/schema` reset_observation value; absent means use the step schema.
         schema_response_json (`str`, *optional*):
@@ -275,6 +279,8 @@ class RuntimeEvidence:
     observation_schema_json: str | None = None
     failure_phase: str | None = None
     failure_reason: str | None = None
+    telemetry_json: str | None = None
+    telemetry_error: str | None = None
     reset_observation_schema_json: str | None = None
     schema_response_json: str | None = None
 
diff --git a/src/openenv/validation/runtime/json_values.py b/src/openenv/validation/runtime/json_values.py
index 078b08acc3..8881ff19d8 100644
--- a/src/openenv/validation/runtime/json_values.py
+++ b/src/openenv/validation/runtime/json_values.py
@@ -35,22 +35,41 @@ def typed_json(value):
     return tuple(result)
 
 
-def dumps(value):
-    """Serialize depth-bounded, redacted values without rounding Decimal leaves."""
-    if isinstance(value, dict):
-        return (
-            "{"
-            + ",".join(
-                json.dumps(key) + ":" + dumps(value[key]) for key in sorted(value)
+def dumps(value, *, ensure_ascii=True):
+    """Serialize exact JSON numbers without recursing through containers."""
+    chunks, pending = [], [(False, value)]
+    while pending:
+        literal, child = pending.pop()
+        if literal:
+            chunks.append(child)
+        elif isinstance(child, dict):
+            chunks.append("{")
+            pending.append((True, "}"))
+            for index, key in reversed(list(enumerate(sorted(child)))):
+                pending.extend(
+                    [
+                        (False, child[key]),
+                        (True, ":"),
+                        (True, json.dumps(key, ensure_ascii=ensure_ascii)),
+                    ]
+                )
+                if index:
+                    pending.append((True, ","))
+        elif isinstance(child, list):
+            chunks.append("[")
+            pending.append((True, "]"))
+            for index, item in reversed(list(enumerate(child))):
+                pending.append((False, item))
+                if index:
+                    pending.append((True, ","))
+        elif isinstance(child, Decimal):
+            if not child.is_finite():
+                raise ValueError("non-finite JSON constant")
+            number = str(child)
+            # Decimal('1') came from a decimal/exponent token, never an integer.
+            chunks.append(
+                number if "." in number or "e" in number.lower() else number + ".0"
             )
-            + "}"
-        )
-    if isinstance(value, list):
-        return "[" + ",".join(dumps(child) for child in value) + "]"
-    if isinstance(value, Decimal):
-        if not value.is_finite():
-            raise ValueError("non-finite JSON constant")
-        number = str(value)
-        # Decimal('1') came from a decimal/exponent token, never an integer.
-        return number if "." in number or "e" in number.lower() else number + ".0"
-    return json.dumps(value, allow_nan=False)
+        else:
+            chunks.append(json.dumps(child, allow_nan=False, ensure_ascii=ensure_ascii))
+    return "".join(chunks)
diff --git a/tests/fixtures/validation/runtime/served_probe/app.py b/tests/fixtures/validation/runtime/served_probe/app.py
index 8227d17284..55ac0c261b 100644
--- a/tests/fixtures/validation/runtime/served_probe/app.py
+++ b/tests/fixtures/validation/runtime/served_probe/app.py
@@ -14,6 +14,7 @@
 from openenv.core.env_server.http_server import create_app
 from openenv.core.env_server.interfaces import Environment
 from openenv.core.env_server.types import Action, Observation, State
+from openenv.core.rubrics import Rubric, WeightedSum
 from pydantic import Field
 
 
@@ -25,6 +26,14 @@ class ProbeObservation(Observation):
     counter: int = Field(strict=True)
 
 
+class CounterRubric(Rubric):
+    def forward(self, action, observation):
+        return float(observation.counter >= 2)
+
+    def validation_config(self):
+        return {"threshold": 2}
+
+
 class ProbeEnvironment(Environment):
     SUPPORTS_CONCURRENT_SESSIONS = True
 
@@ -49,6 +58,8 @@ def worker():
             self.worker = Thread(target=worker, daemon=True)
             self.worker.start()
 
+        self.rubric = WeightedSum([CounterRubric(), CounterRubric()], [0.5, 0.5])
+
     def reset(self, seed=None, episode_id=None, **kwargs):
         if os.environ.get("VALIDATION_FAULT") != "sticky_episode":
             self.counter = 0
@@ -66,11 +77,13 @@ def step(self, action, timeout_s=None, **kwargs):
         else:
             self.counter += action.increment
         self._state.step_count += 1
-        return ProbeObservation(
+        observation = ProbeObservation(
             counter=self.counter,
             reward=float(self.counter >= 2),
             done=self._state.step_count >= 2,
         )
+        observation.reward = self.rubric(action, observation)
+        return observation
 
     @property
     def state(self):
diff --git a/tests/test_validation/integration/test_control_token_privacy.py b/tests/test_validation/integration/test_control_token_privacy.py
new file mode 100644
index 0000000000..230797e26e
--- /dev/null
+++ b/tests/test_validation/integration/test_control_token_privacy.py
@@ -0,0 +1,256 @@
+"""Run-scoped credentials must not survive any of the three real collections."""
+
+import hashlib
+import json
+import shutil
+from decimal import Decimal
+
+import pytest
+import yaml
+from fastapi import FastAPI, Response, WebSocket, WebSocketDisconnect
+from openenv.core.env_server.types import WSErrorCode
+from openenv.validation import runner
+from openenv.validation.runtime import collector
+from openenv.validation.runtime.json_values import loads
+from openenv.validation.types import CheckStatus, Level
+from support.runtime import FakeRuntimeProvider
+from test_late_collection_protocol import served
+from test_runtime_execution import FIXTURE
+
+
+PUBLIC_SCHEMA = '{ "observation": {"type":"object","minimum":1.0000000000000001}, "state": {"description":"public α"} }'
+
+
+def protocol_app(provider, attack_call, surface):
+    app = FastAPI()
+    schemas, requests = [], []
+
+    def control_token():
+        # Read the actual runner-minted launch credential, never a separate oracle.
+        return provider.launches[0].env_vars["OPENENV_VALIDATION_TOKEN"]
+
+    @app.get("/schema")
+    def schema():
+        call = len(schemas) + 1
+        raw = PUBLIC_SCHEMA
+        if call == attack_call and surface in {"schema", "raw_schema"}:
+            suffix = (
+                ',"note":'
+                + json.dumps(control_token())
+                + (',"note":"public"' if surface == "raw_schema" else "")
+            )
+            raw = raw[:-1] + suffix + "}"
+        schemas.append(raw)
+        return Response(raw, media_type="application/json")
+
+    @app.get("/oracle/solve.sh")
+    def oracle():
+        return Response("denied", status_code=403)
+
+    @app.websocket("/ws")
+    async def ws(connection: WebSocket):
+        call = len(schemas)
+        steps = 0
+        await connection.accept()
+        try:
+            while (request := await connection.receive_json())["type"] != "close":
+                operation = request["type"]
+                requests.append((call, operation))
+                if operation not in {"reset", "step", "state"}:
+                    if operation == "validation_open":
+                        assert request["data"]["token"] == control_token()
+                    await connection.send_json(
+                        {
+                            "type": "error",
+                            "data": {
+                                "code": WSErrorCode.UNKNOWN_TYPE,
+                                "message": f"Unknown message type: {operation}",
+                            },
+                        }
+                    )
+                    continue
+                if operation == "reset":
+                    steps = 0
+                elif operation == "step":
+                    steps += 1
+                if operation == "state":
+                    response = {
+                        "type": "state",
+                        "data": {"episode_id": "validation-probe", "step_count": steps},
+                    }
+                else:
+                    observation = {"counter": steps}
+                    if call == attack_call and surface == "wire":
+                        observation["message"] = control_token()
+                    response = {
+                        "type": "observation",
+                        "data": {
+                            "observation": observation,
+                            "reward": 0,
+                            "done": steps == 2,
+                        },
+                    }
+                await connection.send_json(response)
+        except WebSocketDisconnect:
+            # An intentionally rejected credential response closes this session.
+            pass
+
+    return app, schemas, requests
+
+
+@pytest.mark.parametrize(
+    "attack_call,surface",
+    [(None, None)]
+    + [
+        (call, surface)
+        for call in (1, 2, 3)
+        for surface in ("schema", "raw_schema", "wire")
+    ],
+    ids=["public-control"]
+    + [
+        f"{phase}-{surface}"
+        for phase in ("primary", "sequential", "fresh")
+        for surface in ("schema", "raw_schema", "wire")
+    ],
+)
+def test_control_token_omission_across_real_collections(
+    tmp_path, monkeypatch, attack_call, surface
+):
+    root = tmp_path / "subject"
+    shutil.copytree(FIXTURE, root, ignore=shutil.ignore_patterns("__pycache__"))
+    path = root / "openenv.yaml"
+    document = yaml.safe_load(path.read_text())
+    # This server declares only the episode protocol and oracle HTTP probe it serves.
+    for field in ("declared_tools", "rubric_tree", "task_api", "declared_task_count"):
+        document["validation"]["capabilities"].pop(field, None)
+    path.write_text(yaml.safe_dump(document))
+    provider = FakeRuntimeProvider()
+    app, schemas, requests = protocol_app(provider, attack_call, surface)
+    collected, connections = [], []
+    real_collect, real_connect = collector.collect_runtime_evidence, collector.connect
+
+    def collect(*args, **kwargs):
+        evidence = real_collect(*args, **kwargs)
+        collected.append((kwargs, evidence))
+        return evidence
+
+    def connect(*args, **kwargs):
+        connections.append(kwargs)
+        return real_connect(*args, **kwargs)
+
+    monkeypatch.setattr(runner, "collect_runtime_evidence", collect)
+    monkeypatch.setattr(collector, "connect", connect)
+    bundle = tmp_path / "bundle"
+    with served(app) as base_url:
+        provider.subject.base_url = base_url
+        report = runner.run_validation(
+            root, max_level=Level.RUNTIME, provider=provider, artifacts_dir=bundle
+        )
+
+    marker = provider.launches[0].env_vars["OPENENV_VALIDATION_TOKEN"]
+    api_contains_marker = any(marker in repr(evidence) for _, evidence in collected)
+    bundle_marker_files = [
+        p.name for p in bundle.iterdir() if marker.encode() in p.read_bytes()
+    ]
+    # Keep reproduction diagnostics separate from the credential-filtered bundle.
+    (tmp_path / "privacy-observation.json").write_text(
+        json.dumps(
+            {
+                "api_contains_marker": api_contains_marker,
+                "bundle_marker_files": bundle_marker_files,
+                "schema_raw_contains_marker": [marker in raw for raw in schemas],
+                "schema_decoded_contains_marker": [
+                    marker in json.dumps(json.loads(raw)) for raw in schemas
+                ],
+                "verdict": report.verdict.value,
+                "launch_token_sha256": hashlib.sha256(marker.encode()).hexdigest(),
+                "collector_token_sha256": [
+                    hashlib.sha256(kwargs["validation_token"].encode()).hexdigest()
+                    if kwargs.get("validation_token")
+                    else None
+                    for kwargs, _ in collected
+                ],
+                "websocket_max_sizes": [c["max_size"] for c in connections],
+                "operations": requests,
+            }
+        )
+    )
+    assert not api_contains_marker
+    assert not bundle_marker_files
+    assert provider.subject.stopped
+    assert len(provider.launches) == len(provider.builds) == 1
+    assert json.loads((bundle / "cleanup.json").read_text())["completed"] is True
+
+    expected_calls = 1 if attack_call == 1 else 3
+    assert len(schemas) == len(collected) == expected_calls
+    assert [kwargs.get("episodes", 1) for kwargs, _ in collected] == [1, 2, 1][
+        :expected_calls
+    ]
+    for index, (kwargs, _) in enumerate(collected):
+        assert kwargs["episode_timeout_s"] == 30.0
+        assert kwargs.get("request_timeout_s") is None
+        if index == 0:
+            assert kwargs["validation_token"] == marker
+        elif attack_call is not None:
+            assert kwargs["validation_token"] == marker
+            assert kwargs["request_telemetry"] is False
+    assert [call for call, operation in requests if operation == "validation_open"] == (
+        [] if attack_call == 1 and surface != "wire" else [1]
+    )
+    assert not any(operation == "validation_read" for _, operation in requests)
+    for connection in connections:
+        assert 0 < connection["open_timeout"] <= 30.0
+        assert connection["max_queue"] == 1
+        assert connection["compression"] is None
+    expected_sizes = [
+        8 * 1024 * 1024 if i == 0 else 1024 * 1024
+        for i in range(expected_calls)
+        if not (attack_call == i + 1 and surface != "wire")
+    ]
+    assert [c["max_size"] for c in connections] == expected_sizes
+
+    results = {r.check_id: r for r in report.results}
+    assert results["runtime.startup"].status is (
+        CheckStatus.FAIL if attack_call == 1 else CheckStatus.PASS
+    )
+    assert report.verdict.value == ("fail" if attack_call == 1 else "warn")
+    names = ["collector", "isolation-sequential", "isolation-fresh"]
+    for call, (_, evidence) in enumerate(collected, 1):
+        metadata = loads((bundle / f"{names[call - 1]}-evidence.json").read_text())
+        if call > 1:
+            assert evidence.telemetry_json is None
+            assert evidence.telemetry_error is None
+            assert not (bundle / f"{names[call - 1]}-telemetry.json").exists()
+        if call == attack_call:
+            phase = "reset" if surface == "wire" else "schema"
+            assert evidence.failure_phase == metadata["failure_phase"] == phase
+            assert (
+                evidence.failure_reason
+                == metadata["failure_reason"]
+                == f"{phase} failed (ValueError)"
+            )
+            assert metadata["complete"] is False
+            assert evidence.exchanges == ()
+            if surface != "wire":
+                assert evidence.schema_response_json is None
+                assert metadata["schema_response_available"] is False
+        else:
+            assert evidence.failure_reason is None
+            assert metadata["complete"] is True
+            assert len(evidence.exchanges) == (12 if call == 2 else 6)
+        if not (call == attack_call and surface != "wire"):
+            assert evidence.schema_response_json == PUBLIC_SCHEMA
+            assert metadata["schema_response_available"] is True
+            assert metadata["schema_response_parse_failed"] is False
+            assert metadata["schema_response"]["state"] == {"description": "public α"}
+            assert metadata["schema_response"]["observation"]["minimum"] == Decimal(
+                "1.0000000000000001"
+            )
+            assert metadata["observation_schema"]["minimum"] == Decimal(
+                "1.0000000000000001"
+            )
+        assert metadata["omitted_trace_fields"] == []
+        assert metadata["redacted"] is False
+    for line in (bundle / "SHA256SUMS").read_text().splitlines():
+        digest, name = line.split("  ")
+        assert hashlib.sha256((bundle / name).read_bytes()).hexdigest() == digest
diff --git a/tests/test_validation/integration/test_late_collection_protocol.py b/tests/test_validation/integration/test_late_collection_protocol.py
index f5fc63cc93..3e79dab6c6 100644
--- a/tests/test_validation/integration/test_late_collection_protocol.py
+++ b/tests/test_validation/integration/test_late_collection_protocol.py
@@ -191,8 +191,14 @@ def test_late_collection_limits_grade_completed_evidence(
     )
     if limit == "ws_depth":
         rows = json.loads((bundle / f"isolation-{phase}-trace.json").read_text())
-        assert len(rows) == (9 if unsupported_call == 2 else 3)
-        assert metadata["omitted_trace_fields"] == [
-            {"exchange_index": len(rows) - 1, "field": "response_json"}
-        ]
+        # Credential-aware collection cannot retain an unscannable response.
+        # The complete reset/state prefix still reaches grading and the bundle.
+        assert len(rows) == (8 if unsupported_call == 2 else 2)
+        assert rows[-1]["operation"] == "state"
+        assert metadata["omitted_trace_fields"] == []
+    else:
+        assert metadata["schema_response_available"] is False
+        assert metadata["schema_response_parse_failed"] is False
+        assert metadata["schema_available"] is False
+        assert metadata["reset_schema_available"] is False
     assert json.loads((bundle / "cleanup.json").read_text())["completed"] is True
diff --git a/tests/test_validation/integration/test_runtime_process.py b/tests/test_validation/integration/test_runtime_process.py
new file mode 100644
index 0000000000..f7e985e50b
--- /dev/null
+++ b/tests/test_validation/integration/test_runtime_process.py
@@ -0,0 +1,220 @@
+"""Real HTTP/WebSocket evidence with installed OpenEnv and a test-only process.
+
+This covers protocol and grading on CPU Jobs. It does not establish Docker,
+resource-limit, network-isolation or fresh-container acceptance.
+"""
+
+import hashlib
+import importlib.metadata
+import json
+import os
+import shutil
+import signal
+import socket
+import subprocess
+import sys
+import sysconfig
+import time
+from pathlib import Path
+
+import httpx
+import pytest
+from openenv.validation.providers import StartupError, UnsupportedCapability
+from openenv.validation.runner import run_validation
+from openenv.validation.types import CheckStatus, Level, ProviderCapability
+
+FIXTURE = Path(__file__).parents[2] / "fixtures/validation/runtime/served_probe"
+SERVER = """
+import importlib.util, os, sys, uvicorn
+spec = importlib.util.spec_from_file_location('process_probe', sys.argv[1])
+module = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(module)
+uvicorn.run(module.make_app(os.environ['VALIDATION_FAULT']),
+            fd=int(sys.argv[2]), log_level='warning')
+"""
+
+
+class ProcessSubject:
+    def __init__(self, process, port, log_path, record_sha256):
+        self.process = process
+        self.port = port
+        self.log_path = log_path
+        self.record_sha256 = record_sha256
+        self.base_url = f"http://127.0.0.1:{port}"
+
+    def inspect(self):
+        return {
+            "test_only": True,
+            "isolation": "process",
+            "pid": self.process.pid,
+            "installed_record_sha256": self.record_sha256,
+            "container_build_exercised": False,
+            "resource_limits_enforced": False,
+            "network_policy_enforced": False,
+        }
+
+    def logs(self, max_bytes=65536):
+        with self.log_path.open("rb") as stream:
+            stream.seek(max(0, self.log_path.stat().st_size - max_bytes))
+            return stream.read(max_bytes).decode("utf-8", "replace")
+
+    def exec(self, argv, timeout_s):
+        raise UnsupportedCapability("test process provider has no sandbox exec")
+
+    def stop(self):
+        if self.process.poll() is None:
+            os.killpg(self.process.pid, signal.SIGTERM)
+            try:
+                self.process.wait(timeout=3)
+            except subprocess.TimeoutExpired:
+                os.killpg(self.process.pid, signal.SIGKILL)
+                self.process.wait(timeout=3)
+        assert self.process.poll() is not None
+        with socket.socket() as probe:
+            probe.settimeout(1)
+            assert probe.connect_ex(("127.0.0.1", self.port)) != 0
+
+
+class ProcessProvider:
+    """Substitute only process launch; all wire collection and grading are real.
+
+    IMAGE_BUILD satisfies the runner's lifecycle seam in this test only. The
+    reference identifies installed wheel metadata, never a purported Docker image.
+    """
+
+    name = "test-process"
+    capabilities = frozenset({ProviderCapability.IMAGE_BUILD})
+    supported_network_modes = frozenset({"public"})
+
+    def __init__(self, work, mode="good"):
+        self.work = work
+        self.work.mkdir(parents=True)
+        self.mode = mode
+        self.subjects = []
+        installed = next(
+            item
+            for item in importlib.metadata.distributions(
+                path=[sysconfig.get_path("purelib")]
+            )
+            if item.metadata["Name"] == "openenv"
+        )
+        self.record_sha256 = hashlib.sha256(
+            installed.read_text("RECORD").encode()
+        ).hexdigest()
+
+    def build(self, root, execution):
+        self.package = self.work / "fixture"
+        shutil.copytree(
+            root, self.package, ignore=shutil.ignore_patterns("__pycache__")
+        )
+        return "sha256:" + self.record_sha256
+
+    def start(self, spec):
+        assert spec.network.mode == "public" and spec.resources.gpus == 0
+        log_path = self.work / f"server-{len(self.subjects)}.log"
+        with socket.socket() as listener, log_path.open("wb") as log:
+            listener.bind(("127.0.0.1", 0))
+            port = listener.getsockname()[1]
+            process = subprocess.Popen(
+                [
+                    sys.executable,
+                    "-I",
+                    "-B",
+                    "-c",
+                    SERVER,
+                    str(self.package / "app.py"),
+                    str(listener.fileno()),
+                ],
+                stdin=subprocess.DEVNULL,
+                stdout=log,
+                stderr=subprocess.STDOUT,
+                env={
+                    "PATH": os.defpath,
+                    "GRADIO_ANALYTICS_ENABLED": "False",
+                    "VALIDATION_FAULT": self.mode,
+                    **spec.env_vars,
+                },
+                pass_fds=(listener.fileno(),),
+                start_new_session=True,
+            )
+        subject = ProcessSubject(process, port, log_path, self.record_sha256)
+        self.subjects.append(subject)
+        deadline = time.monotonic() + min(spec.startup_timeout_s, 10)
+        try:
+            with httpx.Client(trust_env=False, timeout=0.2) as client:
+                while process.poll() is None and time.monotonic() < deadline:
+                    try:
+                        if client.get(subject.base_url + "/health").status_code == 200:
+                            return subject
+                    except httpx.HTTPError:
+                        pass  # The process may still be starting its HTTP listener.
+                    time.sleep(0.02)
+            raise StartupError("test process failed readiness")
+        except BaseException:
+            subject.stop()
+            raise
+
+
+@pytest.mark.parametrize(
+    "mode,failed_check",
+    [
+        ("good", None),
+        ("bad_reward", "runtime.reward_well_formed"),
+        ("bad_observation", "runtime.observation_schema"),
+        ("missing_done", "runtime.observation_schema"),
+        ("bad_state", "runtime.state_contract"),
+        ("startup_failure", "runtime.startup"),
+    ],
+)
+def test_installed_server_collector_and_graders_over_loopback(
+    tmp_path, mode, failed_check
+):
+    artifacts = (
+        Path(os.environ.get("OPENENV_VALIDATION_ARTIFACTS", tmp_path))
+        / "process"
+        / mode
+    )
+    provider = ProcessProvider(artifacts / "subject", mode)
+    try:
+        report = run_validation(
+            FIXTURE,
+            max_level=Level.RUNTIME,
+            provider=provider,
+            artifacts_dir=artifacts / "report",
+        )
+    finally:
+        for subject in provider.subjects:
+            subject.stop()
+    checks = {result.check_id: result for result in report.results}
+    assert checks["static.manifest"].status is CheckStatus.PASS
+    if failed_check:
+        assert checks[failed_check].status is CheckStatus.FAIL
+    else:
+        for check in (
+            "startup",
+            "reward_well_formed",
+            "observation_schema",
+            "state_contract",
+        ):
+            assert checks[f"runtime.{check}"].status is CheckStatus.PASS
+    if mode != "startup_failure":
+        trace = json.loads((artifacts / "report/collector-trace.json").read_text())
+        assert sum(row["operation"] == "step" for row in trace) == 2
+        manifest = json.loads((artifacts / "report/run-manifest.json").read_text())
+        assert manifest["provider"]["isolation"] == "process"
+        assert manifest["provider"]["container_build_exercised"] is False
+        telemetry_path = artifacts / "report/session-telemetry.json"
+        telemetry = json.loads(telemetry_path.read_text())
+        assert telemetry["seed"]["accepted"] is True
+        assert len(telemetry["trajectory"]["records"]) == len(trace)
+        assert len(telemetry["attribution"]) == 2
+        assert telemetry["trajectory"]["complete"] is True
+    for line in (artifacts / "report/SHA256SUMS").read_text().splitlines():
+        checksum, name = line.split("  ", 1)
+        assert (
+            hashlib.sha256((artifacts / "report" / name).read_bytes()).hexdigest()
+            == checksum
+        )
+    assert provider.subjects and all(
+        subject.process.poll() is not None for subject in provider.subjects
+    )
diff --git a/tests/test_validation/integration/test_session_telemetry_protocol.py b/tests/test_validation/integration/test_session_telemetry_protocol.py
new file mode 100644
index 0000000000..8062e73e40
--- /dev/null
+++ b/tests/test_validation/integration/test_session_telemetry_protocol.py
@@ -0,0 +1,362 @@
+"""Exercise the real replay WebSocket and its separate production MCP boundary."""
+
+import json
+import socket
+import threading
+import time
+
+import pytest
+import uvicorn
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+from fastmcp import FastMCP
+from openenv.core.env_server.http_server import create_app, HTTPEnvServer
+from openenv.core.env_server.interfaces import Environment
+from openenv.core.env_server.types import Action, Observation, State
+from openenv.core.rubrics import Rubric, WeightedSum
+from openenv.validation.runtime import collector
+from openenv.validation.runtime.contracts import RuntimePlan
+
+TOKEN = "per-run-validation-test-token-000000000000"
+
+
+class ValueAction(Action):
+    value: float
+
+
+class ValueScore(Rubric):
+    def forward(self, action, observation):
+        return action.value
+
+    def validation_config(self):
+        return {}
+
+
+class SessionEnv(Environment):
+    SUPPORTS_CONCURRENT_SESSIONS = True
+
+    def __init__(self):
+        super().__init__(rubric=WeightedSum([ValueScore(), ValueScore()], [0.25, 0.75]))
+        self.count = 0
+        self.seed = None
+        self.mcp_server = FastMCP("telemetry-test")
+
+    def reset(self, seed=None):
+        self.seed, self.count = seed, 0
+        return Observation(reward=0.0, metadata={"seed": seed})
+
+    def step(self, action):
+        self.count += 1
+        observation = Observation(metadata={"count": self.count})
+        observation.reward = self.rubric(action, observation)
+        return observation
+
+    @property
+    def state(self):
+        return State(episode_id="same-session", step_count=self.count)
+
+
+class DropsSeed(SessionEnv):
+    def reset(self):
+        return super().reset()
+
+
+class KwargSeed(SessionEnv):
+    def reset(self, **kwargs):
+        return super().reset(seed=kwargs.get("seed"))
+
+
+class AsyncSeed(SessionEnv):
+    async def reset_async(self, seed=None):
+        return super().reset(seed=seed)
+
+
+class AsyncDropsSeed(SessionEnv):
+    async def reset_async(self):
+        return super().reset()
+
+
+@pytest.mark.parametrize("closed_before_send", [False, True])
+def test_factory_error_survives_real_telemetry_handshake(
+    monkeypatch, closed_before_send
+):
+    class BrokenFactory(SessionEnv):
+        def __init__(self):
+            # Fail before base initialization to exercise factory errors.
+            raise RuntimeError(TOKEN)
+
+    monkeypatch.setenv("OPENENV_VALIDATION_TOKEN", TOKEN)
+    monkeypatch.setenv("ENABLE_WEB_INTERFACE", "false")
+    app = create_app(BrokenFactory, ValueAction, Observation)
+    server = uvicorn.Server(uvicorn.Config(app, log_level="critical"))
+    with socket.socket() as listener:
+        listener.bind(("127.0.0.1", 0))
+        port = listener.getsockname()[1]
+        thread = threading.Thread(target=server.run, kwargs={"sockets": [listener]})
+        thread.start()
+        try:
+            deadline = time.monotonic() + 5
+            while not server.started:
+                assert thread.is_alive() and time.monotonic() < deadline
+                time.sleep(0.01)
+            if closed_before_send:
+                real_connect = collector.connect
+
+                def connect_after_close(*args, **kwargs):
+                    connection = real_connect(*args, **kwargs)
+                    deadline = time.monotonic() + 2
+                    while connection.close_code is None:
+                        assert time.monotonic() < deadline
+                        time.sleep(0.01)
+                    return connection
+
+                monkeypatch.setattr(collector, "connect", connect_after_close)
+            plan = RuntimePlan.model_validate(
+                {
+                    "plan_schema_version": "1",
+                    "reset": {"episode_id": "factory-failure", "seed": 1},
+                    "actions": [{"value": 0.5}],
+                }
+            )
+            evidence = collector.collect_runtime_evidence(
+                f"http://127.0.0.1:{port}",
+                plan,
+                episode_timeout_s=3,
+                validation_token=TOKEN,
+            )
+            assert evidence.failure_reason == "validation_open failed (FACTORY_ERROR)"
+            assert evidence.exchanges == ()
+            assert evidence.telemetry_json is None
+            assert TOKEN not in repr(evidence)
+        finally:
+            server.should_exit = True
+            thread.join(timeout=5)
+            assert not thread.is_alive()
+
+
+def app_for(monkeypatch, env=SessionEnv, *, enabled=True, mode="simulation"):
+    if enabled:
+        monkeypatch.setenv("OPENENV_VALIDATION_TOKEN", TOKEN)
+    else:
+        monkeypatch.delenv("OPENENV_VALIDATION_TOKEN", raising=False)
+    app = FastAPI()
+    HTTPEnvServer(env, ValueAction, Observation, max_concurrent_envs=4).register_routes(
+        app, mode=mode
+    )
+    return app
+
+
+def exchange(ws, request):
+    ws.send_json(request)
+    return ws.receive_json()
+
+
+def authorize(ws):
+    response = exchange(
+        ws, {"type": "validation_open", "data": {"schema_version": 1, "token": TOKEN}}
+    )
+    assert response["type"] == "validation_open", response
+    return response["data"]["capability"]
+
+
+def read(ws, capability):
+    return exchange(
+        ws,
+        {
+            "type": "validation_read",
+            "data": {"schema_version": 1, "capability": capability},
+        },
+    )
+
+
+@pytest.mark.parametrize("invalid_action", [{}, {"value": -1}])
+def test_failed_operation_marks_subject_record_incomplete(monkeypatch, invalid_action):
+    class FailingStep(SessionEnv):
+        def step(self, action):
+            if action.value < 0:
+                raise RuntimeError("step failed")
+            return super().step(action)
+
+    with (
+        TestClient(app_for(monkeypatch, FailingStep)) as client,
+        client.websocket_connect("/ws") as ws,
+    ):
+        capability = authorize(ws)
+        exchange(ws, {"type": "reset", "data": {"seed": 1}})
+        exchange(ws, {"type": "step", "data": {"value": 0.5}})
+        assert exchange(ws, {"type": "step", "data": invalid_action})["type"] == "error"
+        assert exchange(ws, {"type": "state"})["type"] == "state"
+        trajectory = read(ws, capability)["data"]["trajectory"]
+        assert trajectory["complete"] is False
+        assert trajectory["reason"] == "Session operation failed"
+        assert len(trajectory["records"]) == 2
+
+
+@pytest.mark.parametrize(
+    "env,accepted",
+    [
+        (SessionEnv, True),
+        (DropsSeed, False),
+        (AsyncSeed, True),
+        (AsyncDropsSeed, False),
+        (KwargSeed, True),
+    ],
+)
+def test_same_session_seed_scores_and_subject_record(monkeypatch, env, accepted):
+    with TestClient(app_for(monkeypatch, env)) as client:
+        with client.websocket_connect("/ws") as ws:
+            capability = authorize(ws)
+            requests = [
+                {"type": "reset", "data": {"seed": 42}},
+                {"type": "state"},
+                {"type": "step", "data": {"value": 0.6}},
+                {"type": "state"},
+            ]
+            responses = [exchange(ws, request) for request in requests]
+            snapshot = read(ws, capability)["data"]
+            assert snapshot["seed"] == {
+                "requested": True,
+                "value": 42,
+                "accepted": accepted,
+            }
+            assert responses[-1]["data"]["step_count"] == 1
+            assert responses[0]["data"]["observation"]["metadata"]["seed"] == (
+                42 if accepted else None
+            )
+            assert snapshot["trajectory"]["complete"] is True
+            assert snapshot["trajectory"]["source"] == "openenv-server"
+            assert snapshot["trajectory"]["records"] == [
+                {"operation": req["type"], "request": req, "response": resp}
+                for req, resp in zip(requests, responses)
+            ]
+            assert len(snapshot["attribution"]) == 1
+            assert snapshot["attribution"][0]["step_index"] == 0
+            assert all(node["evaluated"] for node in snapshot["rubric"])
+            assert snapshot["rubric"][0]["score"] == pytest.approx(0.6)
+            # Editing the collector's copy cannot modify the subject's retained record.
+            responses[-1]["data"]["step_count"] = 999
+            assert (
+                read(ws, capability)["data"]["trajectory"]["records"][-1]["response"][
+                    "data"
+                ]["step_count"]
+                == 1
+            )
+            assert TOKEN not in json.dumps(snapshot)
+            assert capability not in json.dumps(snapshot)
+
+
+@pytest.mark.parametrize("enabled,mode", [(False, "simulation"), (True, "production")])
+def test_validation_is_opt_in_and_simulation_only(monkeypatch, enabled, mode):
+    with TestClient(app_for(monkeypatch, enabled=enabled, mode=mode)) as client:
+        with client.websocket_connect("/ws") as ws:
+            denied = exchange(
+                ws,
+                {
+                    "type": "validation_open",
+                    "data": {"schema_version": 1, "token": TOKEN},
+                },
+            )
+            assert denied["type"] == "error"
+            assert TOKEN not in json.dumps(denied)
+            # Ordinary replay clients do not need to know about telemetry.
+            assert (
+                exchange(ws, {"type": "reset", "data": {"seed": 4}})["type"]
+                == "observation"
+            )
+
+
+def test_missing_wrong_malformed_cross_socket_and_expired_capabilities(monkeypatch):
+    with TestClient(app_for(monkeypatch)) as client:
+        with (
+            client.websocket_connect("/ws") as first,
+            client.websocket_connect("/ws") as second,
+        ):
+            assert read(first, "x" * 32)["type"] == "error"
+            for data in (
+                {},
+                {"schema_version": 1, "token": "wrong-" * 8},
+                {"schema_version": 999, "token": TOKEN},
+            ):
+                denied = exchange(first, {"type": "validation_open", "data": data})
+                assert denied["type"] == "error"
+                assert TOKEN not in json.dumps(denied)
+                assert "wrong-" not in json.dumps(denied)
+            first_cap, second_cap = authorize(first), authorize(second)
+            assert first_cap != second_cap
+            assert read(second, first_cap)["type"] == "error"
+            assert read(first, second_cap)["type"] == "error"
+            assert read(first, first_cap)["type"] == "validation"
+        with client.websocket_connect("/ws") as fresh:
+            authorize(fresh)
+            assert read(fresh, first_cap)["type"] == "error"
+
+
+def test_late_open_cannot_discard_prior_replay_operations(monkeypatch):
+    with (
+        TestClient(app_for(monkeypatch)) as client,
+        client.websocket_connect("/ws") as ws,
+    ):
+        exchange(ws, {"type": "reset", "data": {"seed": 0}})
+        assert (
+            exchange(
+                ws,
+                {
+                    "type": "validation_open",
+                    "data": {"schema_version": 1, "token": TOKEN},
+                },
+            )["type"]
+            == "error"
+        )
+
+
+def test_production_mcp_has_no_telemetry_or_reset_tools(monkeypatch):
+    with TestClient(app_for(monkeypatch, mode="production")) as client:
+        assert client.post("/reset", json={}).status_code == 404
+        with client.websocket_connect("/mcp") as ws:
+            listed = exchange(
+                ws, {"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}}
+            )
+            assert listed["result"]["tools"] == []
+            for name in ("reset", "validation_open", "validation_read"):
+                result = exchange(
+                    ws,
+                    {
+                        "jsonrpc": "2.0",
+                        "id": 2,
+                        "method": "tools/call",
+                        "params": {"name": name, "arguments": {}},
+                    },
+                )
+                assert "error" in result
+
+
+@pytest.mark.parametrize("broken", [False, True])
+def test_absent_or_uninspectable_rubric_does_not_hide_subject_record(
+    monkeypatch, broken
+):
+    class NoRubricEnv(SessionEnv):
+        def __init__(self):
+            super().__init__()
+            if broken:
+                # Exact stock containers have known config; use a custom leaf.
+                self.rubric = ValueScore()
+                self.rubric.validation_config = lambda: {"private": object()}
+            else:
+                self.rubric = None
+
+        def step(self, action):
+            self.count += 1
+            return Observation(reward=0.5)
+
+    with (
+        TestClient(app_for(monkeypatch, NoRubricEnv)) as client,
+        client.websocket_connect("/ws") as ws,
+    ):
+        capability = authorize(ws)
+        exchange(ws, {"type": "reset", "data": {"seed": 1}})
+        exchange(ws, {"type": "step", "data": {"value": 0.5}})
+        snapshot = read(ws, capability)["data"]
+        assert snapshot["trajectory"]["complete"] is True
+        assert len(snapshot["trajectory"]["records"]) == 2
+        assert snapshot["rubric"] == []
+        assert bool(snapshot["rubric_error"]) is broken
diff --git a/tests/test_validation/test_runtime_artifacts.py b/tests/test_validation/test_runtime_artifacts.py
index 7a798a451b..f169c880d1 100644
--- a/tests/test_validation/test_runtime_artifacts.py
+++ b/tests/test_validation/test_runtime_artifacts.py
@@ -551,6 +551,24 @@ def test_malformed_wire_omission_is_visible_in_metadata(tmp_path):
     ]
 
 
+@pytest.mark.parametrize(
+    "telemetry",
+    [
+        '{"rubric":[{"config":{"api_key":"private-value"}}]}',
+        "{broken",
+        '{"rubric":[],"rubric":[]}',
+    ],
+)
+def test_telemetry_redaction_or_omission_marks_bundle_modified(tmp_path, telemetry):
+    original = replace(measured(), telemetry_json=telemetry)
+    write_runtime_bundle(tmp_path, report(), evidence=original)
+    metadata = json.loads((tmp_path / "collector-evidence.json").read_text())
+    assert metadata["redacted"] is True
+    assert "private-value" not in "".join(
+        path.read_text() for path in tmp_path.iterdir()
+    )
+
+
 def test_redaction_preserves_token_metadata_and_filters_secret_keys():
     from openenv.validation.runtime.artifacts import _redact
 
diff --git a/tests/test_validation/test_runtime_collector.py b/tests/test_validation/test_runtime_collector.py
index c8e3376902..d2b8a64013 100644
--- a/tests/test_validation/test_runtime_collector.py
+++ b/tests/test_validation/test_runtime_collector.py
@@ -168,6 +168,48 @@ def test_uncompressed_schema_still_obeys_total_byte_budget(monkeypatch, plan):
     assert evidence.observation_schema_json is None
 
 
+@pytest.mark.parametrize("schema_key", ["observation", "reset_observation"])
+@pytest.mark.parametrize("escaped", [False, True])
+@pytest.mark.parametrize("request_telemetry", [False, True])
+def test_schema_cannot_persist_validation_credential(
+    monkeypatch, plan, schema_key, escaped, request_telemetry
+):
+    token = "validation-secret-value-" * 2
+    schemas = {"observation": {"type": "object"}}
+    schemas[schema_key] = {"properties": {token: {"description": token}}}
+    payload = json.dumps(schemas)
+    if escaped:
+        payload = payload.replace(token, "".join(f"\\u{ord(c):04x}" for c in token))
+    stream = TrackedStream(payload.encode())
+    schema_transport(monkeypatch, stream)
+    evidence = collector.collect_runtime_evidence(
+        "http://127.0.0.1:8000",
+        plan,
+        episode_timeout_s=2,
+        validation_token=token,
+        request_telemetry=request_telemetry,
+    )
+    assert getattr(evidence, f"{schema_key}_schema_json") is None
+    assert evidence.failure_phase == "schema"
+    assert token not in str(evidence)
+
+
+def test_raw_schema_cannot_hide_validation_credential_in_duplicate_member(
+    monkeypatch, plan
+):
+    token = "validation-secret-value-" * 2
+    payload = json.dumps(
+        {"description": token, "observation": {"type": "object"}}
+    ).replace(f'"{token}"', f'"{token}","description":"safe"')
+    schema_transport(monkeypatch, TrackedStream(payload.encode()))
+    evidence = collector.collect_runtime_evidence(
+        "http://127.0.0.1:8000", plan, episode_timeout_s=2, validation_token=token
+    )
+    assert evidence.failure_phase == "schema"
+    assert evidence.schema_response_json is None
+    assert token not in repr(evidence)
+
+
 @pytest.mark.parametrize("slow_phase", ["headers", "body"])
 def test_schema_deadline_aborts_trickling_http_server(plan, slow_phase):
     body = json.dumps({"observation": {"type": "object"}}).encode()
@@ -271,3 +313,66 @@ def test_reset_schema_shares_entire_response_byte_budget(monkeypatch, plan):
     )
     assert evidence.failure_phase == "schema"
     assert evidence.reset_observation_schema_json is None
+
+
+@pytest.mark.parametrize(
+    "schema_key", ["observation", "reset_observation", "action", "state"]
+)
+@pytest.mark.parametrize("escaped", [False, True])
+@pytest.mark.parametrize("request_telemetry", [False, True])
+def test_full_schema_cannot_persist_validation_credential(
+    monkeypatch, plan, schema_key, escaped, request_telemetry
+):
+    token = "validation-private-value-" * 2
+    schemas = {"observation": {"type": "object"}, schema_key: {"description": token}}
+    payload = json.dumps(schemas)
+    if escaped:
+        payload = payload.replace(token, "".join(f"\\u{ord(c):04x}" for c in token))
+    schema_transport(monkeypatch, TrackedStream(payload.encode()))
+    evidence = collector.collect_runtime_evidence(
+        "http://127.0.0.1:8000",
+        plan,
+        episode_timeout_s=2,
+        validation_token=token,
+        request_telemetry=request_telemetry,
+    )
+    assert evidence.failure_phase == "schema"
+    assert evidence.schema_response_json is None
+    assert token not in repr(evidence)
+
+
+@pytest.mark.parametrize("request_telemetry", [False, True])
+@pytest.mark.parametrize("stage", ["loads", "dumps"])
+@pytest.mark.parametrize(
+    "failure",
+    [ValueError, ArithmeticError, RecursionError, RuntimeError, KeyboardInterrupt],
+)
+def test_schema_parse_or_scan_failure_omits_unchecked_raw_body(
+    monkeypatch, plan, request_telemetry, stage, failure
+):
+    payload = b'{"observation":{"type":"object"}}'
+    schema_transport(monkeypatch, TrackedStream(payload))
+
+    def fail(value):
+        raise failure("untrusted diagnostic")
+
+    monkeypatch.setattr(collector, stage, fail)
+    try:
+        evidence = collector.collect_runtime_evidence(
+            "http://127.0.0.1:8000",
+            plan,
+            episode_timeout_s=2,
+            validation_token="synthetic-control-credential",
+            request_telemetry=request_telemetry,
+        )
+    except (
+        collector.RuntimeCollectionUnsupported,
+        collector.RuntimeCollectionInterrupted,
+    ) as exc:
+        evidence = exc.evidence
+    assert evidence.failure_phase == "schema"
+    assert evidence.schema_response_json is None
+    assert evidence.observation_schema_json is None
+    assert evidence.reset_observation_schema_json is None
+    assert evidence.exchanges == ()
+    assert "untrusted diagnostic" not in repr(evidence)
diff --git a/tests/test_validation/test_runtime_execution.py b/tests/test_validation/test_runtime_execution.py
index 07cf6026a2..42afaf2282 100644
--- a/tests/test_validation/test_runtime_execution.py
+++ b/tests/test_validation/test_runtime_execution.py
@@ -95,6 +95,12 @@ def collect(*args, **kwargs):
     results = {r.check_id: r.status for r in report.results}
     # The measured episode, then a same-session pair and a fresh episode.
     assert [kwargs.get("episodes", 1) for _, kwargs in calls] == [1, 2, 1]
+    assert len({kwargs["validation_token"] for _, kwargs in calls}) == 1
+    assert [kwargs.get("request_telemetry", True) for _, kwargs in calls] == [
+        True,
+        False,
+        False,
+    ]
     assert len(provider.builds) == len(provider.launches) == 1
     assert provider.subject.stopped
     assert report.report_schema_version == "2"
diff --git a/tests/test_validation/test_runtime_telemetry_collector.py b/tests/test_validation/test_runtime_telemetry_collector.py
new file mode 100644
index 0000000000..95458bd969
--- /dev/null
+++ b/tests/test_validation/test_runtime_telemetry_collector.py
@@ -0,0 +1,343 @@
+"""Telemetry cannot expand its byte budget or retain replay credentials."""
+
+import json
+
+import httpx
+import pytest
+from openenv.validation.runtime import collector
+from openenv.validation.runtime.contracts import RuntimePlan
+from websockets.exceptions import ConnectionClosedError
+
+TOKEN = "run-authorization-" + "x" * 32
+CAPABILITY = "socket-capability-" + "y" * 32
+
+
+def collect(
+    monkeypatch,
+    *,
+    snapshot=None,
+    snapshot_raw=None,
+    leaked_response=None,
+    leak_operation="reset",
+    escaped=False,
+    malformed=False,
+    opening_reply=None,
+    opening_error=None,
+    send_error=None,
+    request_telemetry=True,
+):
+    original = httpx.Client
+    monkeypatch.setattr(
+        collector.httpx,
+        "Client",
+        lambda **kwargs: original(
+            transport=httpx.MockTransport(
+                lambda request: httpx.Response(200, json={"observation": {}})
+            ),
+            **kwargs,
+        ),
+    )
+
+    class Connection:
+        def send(self, raw):
+            self.request = json.loads(raw)
+            if self.request["type"] == "validation_open" and send_error is not None:
+                raise send_error
+
+        def recv(self, timeout):
+            operation = self.request["type"]
+            if operation == "validation_open":
+                if opening_error is not None:
+                    raise opening_error
+                if opening_reply is not None:
+                    return opening_reply
+                response = {
+                    "type": operation,
+                    "data": {"schema_version": 1, "capability": CAPABILITY},
+                }
+            elif operation == "validation_read":
+                if snapshot_raw is not None:
+                    return '{"type":"validation","data":' + snapshot_raw + "}"
+                response = {
+                    "type": "validation",
+                    "data": snapshot or {"schema_version": 1},
+                }
+            elif operation == "state":
+                response = {
+                    "type": "state",
+                    "data": {"episode_id": "test", "step_count": 0},
+                }
+                if operation == leak_operation and leaked_response is not None:
+                    response["data"]["message"] = leaked_response
+            else:
+                response = {
+                    "type": "observation",
+                    "data": {
+                        "observation": {
+                            "message": leaked_response
+                            if operation == leak_operation
+                            else None
+                        },
+                        "done": False,
+                        "reward": 0.0,
+                    },
+                }
+            raw = json.dumps(response, ensure_ascii=False, separators=(",", ":"))
+            if escaped and leaked_response:
+                raw = raw.replace(
+                    leaked_response,
+                    "".join(f"\\u{ord(c):04x}" for c in leaked_response),
+                )
+            return raw[:-1] if malformed and operation == leak_operation else raw
+
+        def close(self):
+            pass
+
+    monkeypatch.setattr(collector, "connect", lambda *args, **kwargs: Connection())
+    plan = RuntimePlan.model_validate(
+        {
+            "plan_schema_version": "1",
+            "reset": {"episode_id": "test", "seed": 1},
+            "actions": [{"increment": 1}],
+        }
+    )
+    return collector.collect_runtime_evidence(
+        "http://127.0.0.1:8000",
+        plan,
+        episode_timeout_s=2,
+        validation_token=TOKEN,
+        request_telemetry=request_telemetry,
+    )
+
+
+def test_utf8_telemetry_stays_inside_received_budget(monkeypatch):
+    monkeypatch.setattr(collector, "MAX_TRACE_BYTES", 2048)
+    result = collect(monkeypatch, snapshot={"schema_version": 1, "text": "😀" * 400})
+    assert result.telemetry_error is None
+    assert len(result.telemetry_json.encode()) <= 2048
+    assert "😀" in result.telemetry_json
+
+
+@pytest.mark.parametrize("credential", [TOKEN, CAPABILITY])
+def test_telemetry_cannot_persist_credentials_under_arbitrary_keys(
+    monkeypatch, credential
+):
+    result = collect(monkeypatch, snapshot={"schema_version": 1, "debug": [credential]})
+    assert result.telemetry_json is None
+    assert result.telemetry_error == "session telemetry failed (ValueError)"
+    assert result.failure_reason is None
+    assert len(result.exchanges) == 4
+    assert credential not in repr(result)
+
+
+@pytest.mark.parametrize(
+    "credential,escaped",
+    [(TOKEN, False), (CAPABILITY, False), (TOKEN, True), (CAPABILITY, True)],
+)
+def test_normal_wire_cannot_persist_plain_or_escaped_credentials(
+    monkeypatch, credential, escaped
+):
+    result = collect(monkeypatch, leaked_response=credential, escaped=escaped)
+    assert result.exchanges == ()
+    assert result.failure_reason == "reset failed (ValueError)"
+    assert credential not in repr(result)
+
+
+@pytest.mark.parametrize("operation", ["reset", "step", "state"])
+@pytest.mark.parametrize("escaped", [False, True])
+@pytest.mark.parametrize("malformed", [False, True])
+def test_scrub_only_wire_omits_credentials_without_collecting_telemetry(
+    monkeypatch, operation, escaped, malformed
+):
+    result = collect(
+        monkeypatch,
+        leaked_response=TOKEN,
+        leak_operation=operation,
+        escaped=escaped,
+        malformed=malformed,
+        request_telemetry=False,
+    )
+    assert TOKEN not in repr(result)
+    assert result.failure_phase == operation
+    assert (
+        result.failure_reason
+        == f"{operation} failed ({'JSONDecodeError' if escaped and malformed else 'ValueError'})"
+    )
+    assert len(result.exchanges) == {"reset": 0, "state": 1, "step": 2}[operation]
+    assert result.telemetry_json is None
+    assert result.telemetry_error is None
+
+
+@pytest.mark.parametrize(
+    "reply,error_name",
+    [
+        (TOKEN.encode(), "ValueError"),
+        ("not-json-" + TOKEN, "JSONDecodeError"),
+        (json.dumps([TOKEN]), "ValueError"),
+    ],
+    ids=["binary", "non-json", "list"],
+)
+def test_malformed_opening_reply_keeps_the_ordinary_session(
+    monkeypatch, reply, error_name
+):
+    result = collect(monkeypatch, opening_reply=reply)
+    assert result.telemetry_error == f"session telemetry failed ({error_name})"
+    assert result.telemetry_json is None
+    assert result.failure_reason is None
+    assert [row.operation for row in result.exchanges] == [
+        "reset",
+        "state",
+        "step",
+        "state",
+    ]
+    assert TOKEN not in repr(result)
+
+
+@pytest.mark.parametrize(
+    "error", [ConnectionClosedError(None, None), TimeoutError(TOKEN)]
+)
+def test_opening_transport_failure_still_aborts_collection(monkeypatch, error):
+    result = collect(monkeypatch, opening_error=error)
+    assert result.failure_phase == "validation_open"
+    assert result.failure_reason == f"validation_open failed ({type(error).__name__})"
+    assert result.telemetry_json is None
+    assert not result.exchanges
+    assert TOKEN not in repr(result)
+
+
+def test_opening_cancellation_is_not_downgraded_to_optional_telemetry(monkeypatch):
+    with pytest.raises(collector.RuntimeCollectionInterrupted) as error:
+        collect(monkeypatch, opening_error=KeyboardInterrupt(TOKEN))
+    assert error.value.evidence.failure_phase == "validation_open"
+    assert not error.value.evidence.exchanges
+    assert TOKEN not in repr(error.value.evidence)
+
+
+@pytest.mark.parametrize("code", ["FACTORY_ERROR", "CAPACITY_REACHED", "SESSION_ERROR"])
+@pytest.mark.parametrize("closed_before_send", [False, True])
+def test_terminal_opening_error_preserves_only_its_safe_code(
+    monkeypatch, code, closed_before_send
+):
+    result = collect(
+        monkeypatch,
+        opening_reply=json.dumps(
+            {"type": "error", "data": {"code": code, "message": TOKEN}}
+        ),
+        send_error=ConnectionClosedError(None, None) if closed_before_send else None,
+    )
+    assert result.failure_reason == f"validation_open failed ({code})"
+    assert result.exchanges == ()
+    assert result.telemetry_json is None
+    assert TOKEN not in repr(result)
+
+
+@pytest.mark.parametrize("code", ["UNKNOWN_TYPE", "VALIDATION_ERROR", TOKEN])
+def test_optional_opening_refusal_does_not_fail_the_episode(monkeypatch, code):
+    result = collect(
+        monkeypatch,
+        opening_reply=json.dumps(
+            {"type": "error", "data": {"code": code, "message": TOKEN}}
+        ),
+    )
+    assert result.failure_reason is None
+    assert result.telemetry_json is None
+    assert (
+        result.telemetry_error
+        == "session telemetry unavailable or authorization refused"
+    )
+    assert [row.operation for row in result.exchanges] == [
+        "reset",
+        "state",
+        "step",
+        "state",
+    ]
+    assert TOKEN not in repr(result)
+
+
+@pytest.mark.parametrize(
+    "reply",
+    [
+        "not-json-" + TOKEN,
+        json.dumps(
+            {
+                "type": "validation_open",
+                "data": {"schema_version": 1, "capability": CAPABILITY},
+            }
+        ),
+    ],
+)
+def test_closed_send_cannot_become_an_optional_or_successful_handshake(
+    monkeypatch, reply
+):
+    result = collect(
+        monkeypatch,
+        opening_reply=reply,
+        send_error=ConnectionClosedError(None, None),
+    )
+    assert result.failure_reason == "validation_open failed (ConnectionClosedError)"
+    assert result.exchanges == ()
+    assert result.telemetry_json is None
+    assert TOKEN not in repr(result)
+
+
+@pytest.mark.parametrize("credential", [TOKEN, CAPABILITY])
+@pytest.mark.parametrize("operation", ["reset", "step"])
+def test_malformed_wire_cannot_retain_escaped_credentials(
+    monkeypatch, credential, operation
+):
+    result = collect(
+        monkeypatch,
+        leaked_response=credential,
+        leak_operation=operation,
+        escaped=True,
+        malformed=True,
+    )
+    assert [row.operation for row in result.exchanges] == (
+        [] if operation == "reset" else ["reset", "state"]
+    )
+    assert result.failure_reason == f"{operation} failed (JSONDecodeError)"
+
+
+@pytest.mark.parametrize("surface", ["wire", "telemetry"])
+@pytest.mark.parametrize("credential", [None, TOKEN, CAPABILITY])
+def test_deep_json_retains_credential_checks_without_recursion(
+    monkeypatch, surface, credential
+):
+    value = credential or "public"
+    for _ in range(500):
+        value = [value]
+    result = collect(
+        monkeypatch,
+        **(
+            {"snapshot": {"deep": value}}
+            if surface == "telemetry"
+            else {"leaked_response": value}
+        ),
+    )
+    if credential:
+        assert credential not in repr(result)
+        if surface == "telemetry":
+            assert result.telemetry_json is None
+            assert result.telemetry_error == "session telemetry failed (ValueError)"
+            assert result.failure_reason is None
+        else:
+            assert result.exchanges == ()
+            assert result.failure_reason == "reset failed (ValueError)"
+    else:
+        assert result.failure_reason is None
+        assert result.telemetry_error is None
+        assert result.telemetry_json is not None
+        assert len(result.exchanges) == 4
+
+
+@pytest.mark.parametrize("number", ["1.00000000000000001", "1e-999"])
+def test_telemetry_collection_preserves_exact_json_numbers(monkeypatch, number):
+    from decimal import Decimal
+
+    from openenv.validation.runtime.json_values import loads
+
+    result = collect(
+        monkeypatch, snapshot_raw='{"schema_version":1,"value":' + number + "}"
+    )
+    assert result.telemetry_error is None
+    assert loads(result.telemetry_json)["value"] == Decimal(number)
diff --git a/tests/test_validation/test_session_telemetry.py b/tests/test_validation/test_session_telemetry.py
new file mode 100644
index 0000000000..c3a313e8f5
--- /dev/null
+++ b/tests/test_validation/test_session_telemetry.py
@@ -0,0 +1,102 @@
+"""Evidence remains detached, bounded, and fresh across rubric evaluation paths."""
+
+import asyncio
+
+import pytest
+from openenv.core.env_server import session_telemetry
+from openenv.core.env_server.session_telemetry import (
+    rubric_counts,
+    rubric_snapshot,
+    SessionTelemetry,
+)
+from openenv.core.rubrics import Gate, Rubric, Sequential, WeightedSum
+
+
+class PublicScore(Rubric):
+    def forward(self, action, observation):
+        return action
+
+    def validation_config(self):
+        return {}
+
+
+class AsyncScore(PublicScore):
+    async def forward(self, action, observation):
+        return action
+
+
+@pytest.mark.parametrize("score_cls", [PublicScore, AsyncScore])
+def test_gating_excludes_stale_scores_in_sync_and_async_paths(score_cls):
+    rubric = Sequential(Gate(score_cls(), threshold=0.5), score_cls())
+
+    def score(value):
+        result = rubric(value, None)
+        return asyncio.run(result) if asyncio.iscoroutine(result) else result
+
+    score(1.0)
+    before = rubric_counts(rubric)
+    assert score(0.2) == 0.0
+    nodes = {node.name: node for node in rubric_snapshot(rubric, before)}
+    assert nodes["root"].evaluated and nodes["root"].score == 0.0
+    assert nodes["root.rubric_0"].aggregation == "gate"
+    assert nodes["root.rubric_0"].config == {"threshold": 0.5}
+    assert nodes["root.rubric_0.rubric"].score == 0.2
+    assert nodes["root.rubric_1"].evaluated is False
+    assert nodes["root.rubric_1"].score is None
+    assert rubric.rubric_1.last_score == 1.0
+
+
+def test_weighted_semantics_and_private_configuration_are_explicit():
+    class PrivateScore(Rubric):
+        def __init__(self):
+            super().__init__()
+            self.api_key = "must-not-appear"
+
+        def forward(self, action, observation):
+            return 0.5
+
+        def state_dict(self):
+            return {"secret": self.api_key}
+
+    rubric = WeightedSum([PublicScore(), PrivateScore()], [0.2, 0.8])
+    before = rubric_counts(rubric)
+    assert rubric(1.0, None) == pytest.approx(0.6)
+    nodes = rubric_snapshot(rubric, before)
+    assert nodes[0].aggregation == "weighted_sum"
+    assert nodes[0].config == {"weights": [0.2, 0.8]}
+    assert nodes[0].children == ["root.rubric_0", "root.rubric_1"]
+    assert nodes[2].config_available is False
+    assert "must-not-appear" not in str([node.model_dump() for node in nodes])
+
+
+def test_records_are_detached_and_action_and_byte_limits_are_explicit(monkeypatch):
+    subject = SessionTelemetry()
+    response = {"type": "observation", "data": {"observation": {"counter": 1}}}
+    subject.append("step", {"type": "step", "data": {}}, response)
+    response["data"]["observation"]["counter"] = 999
+    assert subject.snapshot.trajectory.records[0].response["data"]["observation"] == {
+        "counter": 1
+    }
+    for _ in range(100):
+        subject.append("step", {"type": "step", "data": {}}, response)
+    assert len(subject.snapshot.trajectory.records) == 100
+    assert subject.snapshot.trajectory.complete is False
+    assert "budget" in subject.snapshot.trajectory.reason
+
+    monkeypatch.setattr(session_telemetry, "MAX_TELEMETRY_BYTES", 4200)
+    subject = SessionTelemetry()
+    subject.append(
+        "state", {"type": "state"}, {"type": "state", "data": {"x": "x" * 200}}
+    )
+    assert subject.snapshot.trajectory.complete is False
+    assert subject.snapshot.trajectory.records == []
+
+
+def test_non_json_or_cyclic_rubrics_cannot_claim_complete_evidence():
+    subject = SessionTelemetry()
+    subject.append("step", {"type": "step", "data": {}}, {"reward": float("nan")})
+    assert subject.snapshot.trajectory.complete is False
+    rubric = PublicScore()
+    rubric.child = rubric
+    with pytest.raises(ValueError, match="cyclic"):
+        rubric_snapshot(rubric)
diff --git a/tests/validation_runtime/README.md b/tests/validation_runtime/README.md
index d54e6f3a16..d99d076f2a 100644
--- a/tests/validation_runtime/README.md
+++ b/tests/validation_runtime/README.md
@@ -20,6 +20,13 @@ is imported by the validator.
 Docker test execution has a twenty-minute budget for the full fault/control
 inventory; fast and protocol suites retain their ten-minute budgets.
 
+The protocol suite also exercises the installed wheel over real loopback HTTP
+and WebSocket connections using a test-only process provider. It verifies
+authorized session telemetry, full collection/grading/reporting, and process
+cleanup on CPU-only Hugging Face Jobs. Its evidence explicitly records process
+isolation; it does not qualify Docker, resource or network isolation. No Hub or
+GitHub credentials are needed by those tests.
+
 The Docker suite snapshots the current source, builds its exact OpenEnv wheel,
 and installs that wheel into this dedicated non-editable test environment. It
 downloads only binary dependencies selected from the committed lock for the
diff --git a/tests/validation_runtime/acceptance.json b/tests/validation_runtime/acceptance.json
index 6569f0f432..78916b3939 100644
--- a/tests/validation_runtime/acceptance.json
+++ b/tests/validation_runtime/acceptance.json
@@ -32,7 +32,27 @@
       "tests.test_validation.integration.test_oracle_disclosures_protocol::test_real_runner[unicode_ws]",
       "tests.test_validation.integration.test_oracle_disclosures_protocol::test_real_runner[json_state]",
       "tests.test_validation.integration.test_oracle_disclosures_protocol::test_real_runner[common_line]",
-      "tests.test_validation.integration.test_oracle_disclosures_protocol::test_real_runner[state_only]"
+      "tests.test_validation.integration.test_oracle_disclosures_protocol::test_real_runner[state_only]",
+      "tests.test_validation.integration.test_session_telemetry_protocol::test_factory_error_survives_real_telemetry_handshake[False]",
+      "tests.test_validation.integration.test_session_telemetry_protocol::test_factory_error_survives_real_telemetry_handshake[True]",
+      "tests.test_validation.integration.test_session_telemetry_protocol::test_same_session_seed_scores_and_subject_record[SessionEnv-True]",
+      "tests.test_validation.integration.test_session_telemetry_protocol::test_same_session_seed_scores_and_subject_record[DropsSeed-False]",
+      "tests.test_validation.integration.test_session_telemetry_protocol::test_same_session_seed_scores_and_subject_record[AsyncSeed-True]",
+      "tests.test_validation.integration.test_session_telemetry_protocol::test_same_session_seed_scores_and_subject_record[AsyncDropsSeed-False]",
+      "tests.test_validation.integration.test_session_telemetry_protocol::test_same_session_seed_scores_and_subject_record[KwargSeed-True]",
+      "tests.test_validation.integration.test_session_telemetry_protocol::test_validation_is_opt_in_and_simulation_only[False-simulation]",
+      "tests.test_validation.integration.test_session_telemetry_protocol::test_validation_is_opt_in_and_simulation_only[True-production]",
+      "tests.test_validation.integration.test_session_telemetry_protocol::test_missing_wrong_malformed_cross_socket_and_expired_capabilities",
+      "tests.test_validation.integration.test_session_telemetry_protocol::test_late_open_cannot_discard_prior_replay_operations",
+      "tests.test_validation.integration.test_session_telemetry_protocol::test_production_mcp_has_no_telemetry_or_reset_tools",
+      "tests.test_validation.integration.test_session_telemetry_protocol::test_absent_or_uninspectable_rubric_does_not_hide_subject_record[False]",
+      "tests.test_validation.integration.test_session_telemetry_protocol::test_absent_or_uninspectable_rubric_does_not_hide_subject_record[True]",
+      "tests.test_validation.integration.test_runtime_process::test_installed_server_collector_and_graders_over_loopback[good-None]",
+      "tests.test_validation.integration.test_runtime_process::test_installed_server_collector_and_graders_over_loopback[bad_reward-runtime.reward_well_formed]",
+      "tests.test_validation.integration.test_runtime_process::test_installed_server_collector_and_graders_over_loopback[bad_observation-runtime.observation_schema]",
+      "tests.test_validation.integration.test_runtime_process::test_installed_server_collector_and_graders_over_loopback[missing_done-runtime.observation_schema]",
+      "tests.test_validation.integration.test_runtime_process::test_installed_server_collector_and_graders_over_loopback[bad_state-runtime.state_contract]",
+      "tests.test_validation.integration.test_runtime_process::test_installed_server_collector_and_graders_over_loopback[startup_failure-runtime.startup]"
     ],
     "docker": [
       "tests.test_validation.integration.test_docker_lifecycle::test_docker_lifecycle_effective_limits_and_owned_cleanup",