From 434bb1dcf892119b1d7344c7fc1c1459a7da390a Mon Sep 17 00:00:00 2001 From: burtenshaw Date: Thu, 24 Sep 2026 12:09:30 +0200 Subject: [PATCH 1/3] feat: add session validation telemetry --- rfcs/008-environment-auto-validation.md | 30 +- src/openenv/core/env_server/http_server.py | 108 +++++++ .../core/env_server/session_telemetry.py | 241 +++++++++++++++ src/openenv/core/rubrics/base.py | 12 + src/openenv/validation/runner.py | 3 + src/openenv/validation/runtime/artifacts.py | 12 + src/openenv/validation/runtime/collector.py | 102 ++++++- src/openenv/validation/runtime/contracts.py | 6 + .../validation/runtime/served_probe/app.py | 14 +- .../integration/test_runtime_process.py | 219 ++++++++++++++ .../test_session_telemetry_protocol.py | 275 ++++++++++++++++++ .../test_validation/test_runtime_artifacts.py | 14 + .../test_validation/test_runtime_collector.py | 12 + .../test_runtime_telemetry_collector.py | 111 +++++++ .../test_validation/test_session_telemetry.py | 102 +++++++ tests/validation_runtime/README.md | 7 + tests/validation_runtime/acceptance.json | 20 +- 17 files changed, 1282 insertions(+), 6 deletions(-) create mode 100644 src/openenv/core/env_server/session_telemetry.py create mode 100644 tests/test_validation/integration/test_runtime_process.py create mode 100644 tests/test_validation/integration/test_session_telemetry_protocol.py create mode 100644 tests/test_validation/test_runtime_telemetry_collector.py create mode 100644 tests/test_validation/test_session_telemetry.py diff --git a/rfcs/008-environment-auto-validation.md b/rfcs/008-environment-auto-validation.md index 63405af7e6..6ad227f5f0 100644 --- a/rfcs/008-environment-auto-validation.md +++ b/rfcs/008-environment-auto-validation.md @@ -590,7 +590,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 50dd5b1746..2e80677fd7 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 @@ -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, @@ -667,6 +679,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 @@ -1534,6 +1552,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") @@ -1598,6 +1618,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": @@ -1629,6 +1707,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( @@ -1706,6 +1790,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: 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 1ba184bfda..fc0786ccc1 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 @@ -95,6 +96,7 @@ async def _run_forward_pre_hooks_async(self, action: Any, observation: Any) -> N 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: @@ -107,6 +109,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: @@ -207,6 +210,15 @@ def reset(self) -> None: """Reset any internal state. Override in subclasses if needed.""" pass + 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 a3c96e0fed..8151dbc133 100644 --- a/src/openenv/validation/runner.py +++ b/src/openenv/validation/runner.py @@ -2,6 +2,7 @@ import hashlib import os +import secrets import stat import time import uuid @@ -140,6 +141,7 @@ def _runtime(subject, *, 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) @@ -159,6 +161,7 @@ def _runtime(subject, *, skip_build, provider): running.base_url, plan, episode_timeout_s=manifest.resources.episode_timeout_s, + validation_token=spec.env_vars["OPENENV_VALIDATION_TOKEN"], ) # A health endpoint without a functioning protocol isn't a startup success. if not evidence.exchanges and evidence.failure_reason: diff --git a/src/openenv/validation/runtime/artifacts.py b/src/openenv/validation/runtime/artifacts.py index d2d8e340bb..44585100d4 100644 --- a/src/openenv/validation/runtime/artifacts.py +++ b/src/openenv/validation/runtime/artifacts.py @@ -101,10 +101,22 @@ def write_runtime_bundle( "failure_reason": evidence.failure_reason, "complete": evidence.failure_phase is None and evidence.failure_reason is None, + "telemetry_error": evidence.telemetry_error, } + telemetry = None + telemetry_parse_failed = False + if evidence.telemetry_json is not None: + try: + telemetry = json.loads(evidence.telemetry_json) + files["session-telemetry.json"] = telemetry + except (ValueError, RecursionError): + telemetry_parse_failed = True + collector_metadata["telemetry_error"] = "malformed telemetry omitted" collector_metadata["redacted"] = ( bool(omitted_fields) or schema_parse_failed + or telemetry_parse_failed + or _redact(telemetry) != telemetry or _redact(trace) != trace or _redact(collector_metadata) != collector_metadata ) diff --git a/src/openenv/validation/runtime/collector.py b/src/openenv/validation/runtime/collector.py index f9d9b32aaa..5f1095e7a2 100644 --- a/src/openenv/validation/runtime/collector.py +++ b/src/openenv/validation/runtime/collector.py @@ -72,6 +72,7 @@ def collect_runtime_evidence( *, episode_timeout_s: float, request_timeout_s: float = 5.0, + validation_token: str | None = None, ) -> RuntimeEvidence: """ Preserve schema and reset/step/state responses without model coercion. @@ -89,6 +90,8 @@ def collect_runtime_evidence( Deadline for the complete collection, including schema retrieval. request_timeout_s (`float`, *optional*, defaults to `5.0`): Per-operation deadline, capped by the remaining episode budget. + validation_token (`str`, *optional*): + Run-scoped telemetry authorization; never retained in evidence. Returns: [`~openenv.validation.runtime.contracts.RuntimeEvidence`]: raw evidence. @@ -98,6 +101,8 @@ def collect_runtime_evidence( schema_json = None phase = "schema" trace_bytes = 0 + telemetry_json = None + telemetry_error = None def remaining() -> float: value = min(request_timeout_s, deadline - time.monotonic()) @@ -131,12 +136,15 @@ def remaining() -> float: schema = json.loads(payload) if not isinstance(schema, dict) or "observation" not in schema: raise ValueError("missing observation schema") - schema_json = json.dumps( + schema_payload = json.dumps( schema["observation"], allow_nan=False, ensure_ascii=False, separators=(",", ":"), ) + if validation_token and validation_token in schema_payload: + raise ValueError("schema contains validation credentials") + schema_json = schema_payload endpoint = urlsplit(base_url) ws_url = urlunsplit( @@ -154,13 +162,59 @@ def remaining() -> float: proxy=None, open_timeout=remaining(), close_timeout=1, - max_size=MAX_MESSAGE_BYTES, + max_size=MAX_TRACE_BYTES if validation_token else MAX_MESSAGE_BYTES, max_queue=1, compression=None, ) complete = False try: + def telemetry_request(operation, data): + request = json.dumps({"type": operation, "data": data}) + _bounded_call(connection, lambda: connection.send(request), remaining()) + raw = connection.recv(timeout=remaining()) + if not isinstance(raw, str) or len(raw.encode()) > MAX_TRACE_BYTES: + raise ValueError("invalid telemetry response") + response = json.loads(raw) + if not isinstance(response, dict): + raise ValueError("invalid telemetry envelope") + return response + + capability = None + if validation_token: + phase = "validation_open" + response = telemetry_request( + phase, {"schema_version": 1, "token": validation_token} + ) + 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): + if isinstance(value, str): + return any( + secret and secret in value + for secret in (validation_token, capability) + ) + if isinstance(value, dict): + return any( + contains_credential(key) or contains_credential(child) + for key, child in value.items() + ) + if isinstance(value, list): + return any(contains_credential(child) for child in value) + return False + def exchange(operation: str, data: dict | None = None) -> dict: nonlocal phase, trace_bytes phase = operation @@ -174,11 +228,23 @@ def exchange(operation: str, data: dict | None = None) -> dict: raw = connection.recv(timeout=remaining()) 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("trace exceeds total 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): + parsed = None + if contains_credential(parsed): + raise ValueError("response contains validation credentials") exchanges.append( WireExchange( operation=operation, @@ -205,6 +271,29 @@ 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 = json.dumps( + response["data"], + allow_nan=False, + ensure_ascii=False, + separators=(",", ":"), + ) + 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 @@ -223,7 +312,10 @@ def exchange(operation: str, data: dict | None = None) -> dict: except (Exception, KeyboardInterrupt): _abort_transport(connection) return RuntimeEvidence( - exchanges=tuple(exchanges), observation_schema_json=schema_json + exchanges=tuple(exchanges), + observation_schema_json=schema_json, + telemetry_json=telemetry_json, + telemetry_error=telemetry_error, ) except KeyboardInterrupt: raise RuntimeCollectionInterrupted( @@ -232,6 +324,8 @@ def exchange(operation: str, data: dict | None = None) -> dict: observation_schema_json=schema_json, failure_phase=phase, failure_reason=f"{phase} failed (KeyboardInterrupt)", + telemetry_json=telemetry_json, + telemetry_error=telemetry_error, ) ) from None except Exception as exc: @@ -241,4 +335,6 @@ def exchange(operation: str, data: dict | None = None) -> dict: observation_schema_json=schema_json, failure_phase=phase, failure_reason=f"{phase} failed ({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 3d84e7be68..9df4a3eb1b 100644 --- a/src/openenv/validation/runtime/contracts.py +++ b/src/openenv/validation/runtime/contracts.py @@ -265,9 +265,15 @@ 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. """ exchanges: tuple[WireExchange, ...] = () 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 diff --git a/tests/fixtures/validation/runtime/served_probe/app.py b/tests/fixtures/validation/runtime/served_probe/app.py index 91999cb91c..edaf208448 100644 --- a/tests/fixtures/validation/runtime/served_probe/app.py +++ b/tests/fixtures/validation/runtime/served_probe/app.py @@ -9,6 +9,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 @@ -20,6 +21,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 @@ -27,6 +36,7 @@ def __init__(self): super().__init__() self._state = State(episode_id="uninitialized", step_count=0) self.counter = 0 + self.rubric = WeightedSum([CounterRubric(), CounterRubric()], [0.5, 0.5]) def reset(self, seed=None, episode_id=None, **kwargs): self.counter = 0 @@ -36,11 +46,13 @@ def reset(self, seed=None, episode_id=None, **kwargs): def step(self, action, timeout_s=None, **kwargs): 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_runtime_process.py b/tests/test_validation/integration/test_runtime_process.py new file mode 100644 index 0000000000..bf6b68c31a --- /dev/null +++ b/tests/test_validation/integration/test_runtime_process.py @@ -0,0 +1,219 @@ +"""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", + "-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 + 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..d4586a2665 --- /dev/null +++ b/tests/test_validation/integration/test_session_telemetry_protocol.py @@ -0,0 +1,275 @@ +"""Exercise the real replay WebSocket and its separate production MCP boundary.""" + +import json + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient +from fastmcp import FastMCP +from openenv.core.env_server.http_server import 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 + +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() + + +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( + "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 3921447f8a..490d28f6b0 100644 --- a/tests/test_validation/test_runtime_artifacts.py +++ b/tests/test_validation/test_runtime_artifacts.py @@ -2,6 +2,7 @@ import json from dataclasses import replace +import pytest from conftest import load_fixture_manifest from openenv.validation.graders import Subject from openenv.validation.graders.runtime import ( @@ -222,3 +223,16 @@ def test_malformed_wire_omission_is_visible_in_metadata(tmp_path): assert metadata["omitted_trace_fields"] == [ {"exchange_index": 0, "field": "response_json"} ] + + +@pytest.mark.parametrize( + "telemetry", ['{"rubric":[{"config":{"api_key":"private-value"}}]}', "{broken"] +) +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() + ) diff --git a/tests/test_validation/test_runtime_collector.py b/tests/test_validation/test_runtime_collector.py index 0741e2e8a5..c6ab1b36bc 100644 --- a/tests/test_validation/test_runtime_collector.py +++ b/tests/test_validation/test_runtime_collector.py @@ -105,3 +105,15 @@ def test_uncompressed_schema_still_obeys_total_byte_budget(monkeypatch, plan): assert evidence.failure_phase == "schema" assert evidence.failure_reason == "schema failed (ValueError)" assert evidence.observation_schema_json is None + + +def test_schema_cannot_persist_validation_credential(monkeypatch, plan): + token = "validation-secret-value-" * 2 + stream = TrackedStream(json.dumps({"observation": {"description": token}}).encode()) + schema_transport(monkeypatch, stream) + evidence = collector.collect_runtime_evidence( + "http://127.0.0.1:8000", plan, episode_timeout_s=2, validation_token=token + ) + assert evidence.observation_schema_json is None + assert evidence.failure_phase == "schema" + assert token not in str(evidence) 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..754c8792c9 --- /dev/null +++ b/tests/test_validation/test_runtime_telemetry_collector.py @@ -0,0 +1,111 @@ +"""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 + +TOKEN = "run-authorization-" + "x" * 32 +CAPABILITY = "socket-capability-" + "y" * 32 + + +def collect(monkeypatch, *, snapshot=None, leaked_response=None, escaped=False): + 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) + + def recv(self, timeout): + operation = self.request["type"] + if operation == "validation_open": + response = { + "type": operation, + "data": {"schema_version": 1, "capability": CAPABILITY}, + } + elif operation == "validation_read": + response = { + "type": "validation", + "data": snapshot or {"schema_version": 1}, + } + elif operation == "state": + response = { + "type": "state", + "data": {"episode_id": "test", "step_count": 0}, + } + else: + response = { + "type": "observation", + "data": { + "observation": {"message": leaked_response}, + "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 + + 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 + ) + + +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) 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 75ddafb443..4b27945737 100644 --- a/tests/validation_runtime/README.md +++ b/tests/validation_runtime/README.md @@ -17,6 +17,13 @@ failure. The reference job uses Linux x86-64; Docker Desktop arm64 uses the same recipe but records its different platform. No host Python code from the subject is imported by the validator. +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 d8a3d1d1c3..75ad404610 100644 --- a/tests/validation_runtime/acceptance.json +++ b/tests/validation_runtime/acceptance.json @@ -7,7 +7,25 @@ "tests.test_validation.integration.test_served_probe::test_fault_is_visible_in_raw_wire_response[bad_observation]", "tests.test_validation.integration.test_served_probe::test_fault_is_visible_in_raw_wire_response[missing_done]", "tests.test_validation.integration.test_served_probe::test_fault_is_visible_in_raw_wire_response[bad_state]", - "tests.test_validation.integration.test_served_probe::test_failed_start_is_explicit" + "tests.test_validation.integration.test_served_probe::test_failed_start_is_explicit", + "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", From f1575e9856aa022b7c0c89fe79e41f2d6cf2906d Mon Sep 17 00:00:00 2001 From: burtenshaw Date: Thu, 24 Sep 2026 12:16:14 +0200 Subject: [PATCH 2/3] fix: keep process evidence free of bytecode --- tests/test_validation/integration/test_runtime_process.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/test_validation/integration/test_runtime_process.py b/tests/test_validation/integration/test_runtime_process.py index bf6b68c31a..efe4464a24 100644 --- a/tests/test_validation/integration/test_runtime_process.py +++ b/tests/test_validation/integration/test_runtime_process.py @@ -119,6 +119,7 @@ def start(self, spec): [ sys.executable, "-I", + "-B", "-c", SERVER, str(self.package / "app.py"), From 317983cc4b55be8fa9a8fc97e00acefd554056c0 Mon Sep 17 00:00:00 2001 From: burtenshaw Date: Thu, 24 Sep 2026 13:59:14 +0200 Subject: [PATCH 3/3] fix: preserve runtime evidence on telemetry errors --- rfcs/008-environment-auto-validation.md | 7 ++- src/openenv/validation/runtime/collector.py | 36 ++++++----- .../integration/test_runtime_process.py | 2 +- .../test_runtime_telemetry_collector.py | 60 ++++++++++++++++++- 4 files changed, 85 insertions(+), 20 deletions(-) diff --git a/rfcs/008-environment-auto-validation.md b/rfcs/008-environment-auto-validation.md index 6ad227f5f0..5f0171a868 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. diff --git a/src/openenv/validation/runtime/collector.py b/src/openenv/validation/runtime/collector.py index 5f1095e7a2..c542855a87 100644 --- a/src/openenv/validation/runtime/collector.py +++ b/src/openenv/validation/runtime/collector.py @@ -183,22 +183,28 @@ def telemetry_request(operation, data): capability = None if validation_token: phase = "validation_open" - response = telemetry_request( - phase, {"schema_version": 1, "token": validation_token} - ) - 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" + 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: + 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): if isinstance(value, str): diff --git a/tests/test_validation/integration/test_runtime_process.py b/tests/test_validation/integration/test_runtime_process.py index efe4464a24..f7e985e50b 100644 --- a/tests/test_validation/integration/test_runtime_process.py +++ b/tests/test_validation/integration/test_runtime_process.py @@ -147,7 +147,7 @@ def start(self, spec): if client.get(subject.base_url + "/health").status_code == 200: return subject except httpx.HTTPError: - pass + pass # The process may still be starting its HTTP listener. time.sleep(0.02) raise StartupError("test process failed readiness") except BaseException: diff --git a/tests/test_validation/test_runtime_telemetry_collector.py b/tests/test_validation/test_runtime_telemetry_collector.py index 754c8792c9..df1dd57146 100644 --- a/tests/test_validation/test_runtime_telemetry_collector.py +++ b/tests/test_validation/test_runtime_telemetry_collector.py @@ -6,12 +6,21 @@ 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, leaked_response=None, escaped=False): +def collect( + monkeypatch, + *, + snapshot=None, + leaked_response=None, + escaped=False, + opening_reply=None, + opening_error=None, +): original = httpx.Client monkeypatch.setattr( collector.httpx, @@ -31,6 +40,10 @@ def send(self, raw): 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}, @@ -109,3 +122,48 @@ def test_normal_wire_cannot_persist_plain_or_escaped_credentials( assert result.exchanges == () assert result.failure_reason == "reset failed (ValueError)" assert credential not in repr(result) + + +@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)