[burtenshaw]

Summary

Add bounded, opt-in telemetry on the existing validation WebSocket session: requested/forwarded seeds, rubric snapshots, per-step attribution and a subject-emitted trajectory.

reset/step/state failure → incomplete subject record → explicit evidence metadata

Execution and response-validation errors mark the record incomplete. The same run credential is scrubbed from primary, sequential-isolation and fresh-isolation evidence. Later sessions keep telemetry disabled and ordinary receive limits; raw schemas enter evidence only after raw and decoded checks succeed. Persistent-session shutdown cleanup and recursive rubric reset behavior are preserved. This is the telemetry prerequisite for #1321 and #1322 and lands after #1524.

Evidence

• Before: an omitted failed operation could leave the subject trajectory marked complete.
• After: failed operations retain incompleteness metadata; collector and real-WebSocket regressions cover same-session telemetry, bounded records, credential omission and saved evidence.

Test Plan

The combined token-scrubbing candidate passed 326 focused source/loopback tests. Independent review passed 325 targeted tests, 19 additional controls and both retained external regressions, and reproduced the seven baseline privacy failures and eight fail-closed retention failures.

The follow-on full stack passed 1,721 source/CLI tests at 47240bda632f273c7805d93dd430b10b5001aeb0. Its test-only privacy-fixture integration at clean a3023550f52217ef15a5932f48eb230c10939ac1 passed 126 fresh non-editable-wheel protocol tests, with all artifact hashes verified and 185 installed Python files matching source. These macOS arm64 results are combined-stack evidence, not individual-PR or Linux/Docker/HF qualification.

Current-head required checks and reference Linux runtime acceptance: PR checks.

RFC Status

RFC 008 / #778 remains In Review. This PR adds no grader: the #1524 baseline remains 9/16 runtime implementations. Telemetry supports later checks; its presence does not establish determinism, rubric correctness or full RFC qualification.

Merge Danger

Door: Two-way.

Blast Radius: Telemetry.

Changes the opt-in server/collector evidence path and runtime bundles.

[bot-ci-comment]

The docs for this PR live here. All of your documentation changes will be reflected on that endpoint. The docs are available until 30 days after the last update.

[cursor]

Alignment Review Report

Automated Checks

• Lint: PASS — usort check, ruff format --check, and ruff check are all clean on every file this PR touches (src/openenv/core/env_server/http_server.py, session_telemetry.py, src/openenv/core/rubrics/base.py, src/openenv/validation/runner.py, src/openenv/validation/runtime/{artifacts,collector,contracts}.py, and the new/changed test and fixture files). A full-tree lint.sh run does flag ~58 unrelated files (e.g. envs/wildfire_env, src/openenv/core/harness/README.md) — pre-existing debt, not introduced by this diff.
• Debug code: CLEAN — no print()/TODO/FIXME/breakpoint() in any file this PR touches. check-debug.sh does flag print()/TODO elsewhere in src/ (e.g. cli/commands/*.py, auto/auto_action.py), all pre-existing and outside this diff.
• Tests: PYTHONPATH=src:envs uv run pytest tests/test_validation/ -q → 521 passed, 22 skipped. Related MCP/WS/production-mode suites under tests/core/ also pass (264 passed, 2 skipped).

Open RFCs Context

• RFC 008 — Environment Auto-Validation (Status: In Review, @zkwentz) is directly implemented/extended by this PR. The diff adds a new "Session telemetry protocol (PR4)" subsection to the RFC text itself and the code matches it closely: opt-in only via OPENENV_VALIDATION_TOKEN, capability bound to one /ws connection, validation_open/validation_read message types, 100-step/202-record/8 MiB bounds, credential redaction. This is RFC-conformant delivery rather than a conflict, but since the governing RFC is still In Review (not Accepted) this is exactly the kind of change worth a design sign-off, which matches the PR's own note ("Leave open for human review; do not merge").
• No other open RFC (001, 002, 003, 005, 006, 010, 011, 012 — all Draft/In Review) touches this code area.

Tier 1: Fixes Required

None — no lint failures, debug code, uninitialized variables, missing imports, or credential-exposure issues found in the touched files.

Tier 2: Alignment Discussion

Principle Conflicts

None identified. Checked against .claude/docs/INVARIANTS.md:

• Agent isolation (src/openenv/core/env_server/http_server.py:891-897): the new validation_open/validation_read handlers are gated by mode == ServerMode.SIMULATION and an explicit OPENENV_VALIDATION_TOKEN env var, so a production-mode server never exposes them. tests/test_validation/integration/test_session_telemetry_protocol.py::test_production_mcp_has_no_telemetry_or_reset_tools and test_validation_is_opt_in_and_simulation_only cover this directly, including that the operations are absent from the /mcp JSON-RPC tool list.
• No credential exposure: the token/capability are pydantic.SecretStr, compared with secrets.compare_digest (constant-time), never echoed back on failure (WSErrorResponse returns a generic "Validation unavailable or unauthorized" message), and the collector (src/openenv/validation/runtime/collector.py) actively scans every wire response/schema for the token or capability substring and aborts collection if found.
• Client-server separation / rewards in environment: no changes to client code or reward computation; Rubric.validation_config() (src/openenv/core/rubrics/base.py:213) is a new, opt-in, server-side-only introspection hook that explicitly excludes state_dict() and defaults to None (not configurable) unless a rubric author opts in.

RFC Conflicts

None identified beyond the context note above — the implementation matches the RFC 008 PR4 amendment it ships alongside.

Summary

• 0 mechanical issues to fix
• 1 alignment point for human review (RFC 008 is still In Review; recommend @zkwentz confirm the PR4 protocol text matches intent before merge, per the PR's own "leave open for human review" note)
• 0 RFC conflicts to discuss

  

_{Sent by Cursor Automation: Pre-review}

[rycerzes]

Reviewed at 215859d9 and ran the pinned lab (Python 3.12.13, uv 0.9.3, --require-complete, clean tree) on two engines:

Suite	Docker 29.7.2, linux/amd64	Podman 5.8.4, macOS arm64
fast	501 passed	501 passed
protocol	28 passed	pass
docker	14/14	14/14

The artifacts verify on every run. I also probed a real echo_env server built from this branch over /ws, on both machines. The following were all refused with the same generic error: a wrong token, a wrong capability, opening after an operation, a second open, an attached (non-owning) socket, and production mode. Seed acceptance was recorded correctly.

The gating looks right to me: simulation-only, owner socket, open before any operation, constant-time comparisons, no MCP exposure, and the token never ends up in evidence. The token is inherited by subprocesses in the subject container, but it only authorizes telemetry on the caller's own socket, so that seems acceptable.

One small comment inline. Separately, test_reproduction::test_command_timeout_retains_partial_log failed once for me under load and passed 5/5 in isolation; its 0.2 s budget has to cover interpreter startup. It isn't touched here, but it may flake in CI.

[cursor]

Cursor Bugbot has reviewed your changes using default effort and found 2 potential issues.

Bugbot Autofix prepared fixes for both issues found in the latest run.

• ✅ Fixed: Isolation paths skip token scrubbing
  • Isolation and fresh collections now receive the validation token for credential scrubbing while telemetry requests remain disabled.
• ✅ Fixed: Schema scrub skips the raw body
  • Schema collection now rejects and discards raw response bodies containing the validation token before evidence can be retained.

Preview (be2d9bc04e)

diff --git a/rfcs/008-environment-auto-validation.md b/rfcs/008-environment-auto-validation.md
--- a/rfcs/008-environment-auto-validation.md
+++ b/rfcs/008-environment-auto-validation.md
@@ -197,9 +197,10 @@
 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,8 +625,36 @@
 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
 implementation has not shipped are distinct outcomes. Independent graders must

diff --git a/src/openenv/core/env_server/http_server.py b/src/openenv/core/env_server/http_server.py
--- 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 @@
                     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 @@
             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 @@
 
                         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 @@
                                             )
                                         )
 
+                                    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 @@
                                         }
                                     )
 
+                            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 @@
                             )
                             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
--- /dev/null
+++ b/src/openenv/core/env_server/session_telemetry.py
@@ -1,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
--- a/src/openenv/core/rubrics/base.py
+++ b/src/openenv/core/rubrics/base.py
@@ -46,6 +46,7 @@
         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 _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 @@
     ) -> 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 @@
         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
--- 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 @@
             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 @@
             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 @@
                         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 @@
                         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
--- a/src/openenv/validation/runtime/artifacts.py
+++ b/src/openenv/validation/runtime/artifacts.py
@@ -106,6 +106,7 @@
             "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 @@
         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
--- 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 @@
     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,10 @@
             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 telemetry authorization; never retained in evidence.
+        request_telemetry (`bool`, *optional*, defaults to `True`):
+            Whether to request session telemetry when authorization is provided.
 
     Returns:
         [`~openenv.validation.runtime.contracts.RuntimeEvidence`]: raw evidence.
@@ -103,6 +120,8 @@
     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:
@@ -141,13 +160,25 @@
                         raise ValueError("schema exceeds size bound")
                     payload.extend(chunk)
             schema_response_json = payload.decode("utf-8")
-            schema = loads(payload)
+            try:
+                schema = loads(payload)
+                if validation_token and (
+                    validation_token in schema_response_json
+                    or validation_token in dumps(schema)
+                ):
+                    raise ValueError("schema contains validation credentials")
+            except (ValueError, ArithmeticError, RecursionError):
+                if validation_token:
+                    schema_response_json = None
+                raise
             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 +196,100 @@
             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 +297,29 @@
                 if data is not None:
... diff truncated: showing 800 of 2250 lines

_{You can send follow-ups to the cloud agent here.}

^{Reviewed by Cursor Bugbot for commit 85b2acc. Configure here.}