Add session telemetry

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

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
@@ -61,6 +62,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,
@@ -879,6 +891,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 idle-session reaper lifecycle via app events
         server_ref = self
 
@@ -1753,6 +1771,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")
@@ -1817,6 +1837,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":
@@ -1848,6 +1926,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(
@@ -1925,6 +2009,30 @@ 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: