diff --git a/rfcs/008-environment-auto-validation.md b/rfcs/008-environment-auto-validation.md index 5989163f06..4ace79b6b8 100644 --- a/rfcs/008-environment-auto-validation.md +++ b/rfcs/008-environment-auto-validation.md @@ -197,9 +197,10 @@ configuration (model, version, params) in the manifest; the oracle check becomes bit-exact. RFC 004 rubrics are **leveraged, not required**: the contract stays spec-neutral (graders read the manifest), but for the served OpenEnv format the rubric tree is the native satisfaction path — `LLMJudge` is the in-repo `llm_judged` implementation, and the -introspectability and reward-attribution graders read `named_rubrics()` / `state_dict()` / -per-child scores. Judge pinning stays a *manifest* declaration because the rubric object does not -serialize model/version/params today. +introspectability and reward-attribution graders read `named_rubrics()`, explicit +`validation_config()` and fresh per-child scores. `state_dict()` is never serialized +as validation configuration. Judge pinning stays a *manifest* declaration because +the rubric object does not serialize model/version/params today. Tolerances, margins, and variance bounds are author-declared in the manifest, **bounded by the versioned severity policy**, and carried verbatim in reports so hubs can apply stricter ceilings. @@ -624,7 +625,35 @@ attached to the **same** replay connection, reject unauthorized/cross-session reads and never expose telemetry as agent MCP tools. A second WebSocket creates another environment and cannot supply evidence for the measured instance. A validator transcript alone cannot pass subject-emitted trajectory recording. -These authorization requirements do not add new public wire messages in this slice. +The initial contracts slice added no public wire messages. PR4 implements the +following opt-in protocol. + +#### Session telemetry protocol (PR4) + +The server opts in only when `OPENENV_VALIDATION_TOKEN` is provisioned explicitly +by the validation supervisor. On its existing simulation `/ws` connection, the +collector sends `validation_open` with `data: {schema_version: 1, token: ...}`. +The reply returns a random capability bound to that connection. Subsequent +`validation_read` messages carry that capability and return a `validation` +snapshot. Disabled, unauthorized and cross-connection requests fail without +echoing credentials. Closing the connection destroys the capability. MCP and +production endpoints never expose these operations. Authentication exchanges +are excluded from persisted evidence. + +Snapshots identify requested and actually forwarded seed arguments; successful +reset alone is not acceptance. They include a named rubric tree rooted at `root`, +explicit safe configuration, and per-step attribution with operation-local +evaluation flags. Unevaluated gated children never reuse an earlier score. +Stock container aggregation is named explicitly; custom rubrics may supply +`validation_config()` to expose public JSON configuration. Arbitrary attributes +and `state_dict()` are never serialized as configuration. + +The subject server also emits a bounded record of the reset/step/state request +and response envelopes it executed. The validator captures the wire separately +and later compares the two. The record is bound to the authenticated session, +limited to 100 steps/202 operations and 8 MiB, and marks truncation explicitly. +Missing configuration, attribution or records cannot be inferred from other +successful operations. This transport introduces no new passing grader by itself. Applicability predicates must distinguish empty declared sets from absent capabilities. Missing subject features, missing provider support and checks whose diff --git a/src/openenv/core/env_server/http_server.py b/src/openenv/core/env_server/http_server.py index 92b40d642a..f11bb891f1 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, @@ -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: 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 842e4df68f..c7a66c26b5 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 @@ -163,6 +164,7 @@ def _runtime(subject, graders, *, skip_build, provider): resources=manifest.resources, network=manifest.network, run_id="validation-" + uuid.uuid4().hex, + env_vars={"OPENENV_VALIDATION_TOKEN": secrets.token_urlsafe(32)}, ) attempted = True image_ref = provider.build(subject.root, manifest.execution) @@ -182,6 +184,7 @@ def _runtime(subject, graders, *, skip_build, provider): running.base_url, plan, episode_timeout_s=manifest.resources.episode_timeout_s, + validation_token=spec.env_vars["OPENENV_VALIDATION_TOKEN"], ) # Protocol collection must finish before its dependent contract checks run. if evidence.failure_reason: diff --git a/src/openenv/validation/runtime/artifacts.py b/src/openenv/validation/runtime/artifacts.py index e69f9fcfd2..d315afe68a 100644 --- a/src/openenv/validation/runtime/artifacts.py +++ b/src/openenv/validation/runtime/artifacts.py @@ -113,10 +113,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 reset_schema_parse_failed 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 4fd853094f..21b264fbd8 100644 --- a/src/openenv/validation/runtime/collector.py +++ b/src/openenv/validation/runtime/collector.py @@ -6,6 +6,7 @@ from urllib.parse import urlsplit, urlunsplit import httpx +from websockets.exceptions import ConnectionClosed from websockets.sync.client import connect from ...core.env_server.types import WSErrorCode @@ -16,6 +17,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") @@ -60,6 +71,7 @@ def collect_runtime_evidence( *, episode_timeout_s: float, request_timeout_s: float | None = None, + validation_token: str | None = None, ) -> RuntimeEvidence: """ Preserve schema and reset/step/state responses without model coercion. @@ -78,6 +90,8 @@ def collect_runtime_evidence( request_timeout_s (`float`, *optional*): Optional per-operation cap. By default, each operation may use the remaining declared episode budget. + validation_token (`str`, *optional*): + Run-scoped telemetry authorization; never retained in evidence. Returns: [`~openenv.validation.runtime.contracts.RuntimeEvidence`]: raw evidence. @@ -87,6 +101,8 @@ def collect_runtime_evidence( schema_json = reset_schema_json = None phase = "schema" trace_bytes = 0 + telemetry_json = None + telemetry_error = None server_code = None def remaining() -> float: @@ -127,20 +143,26 @@ 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 if "reset_observation" in schema: - reset_schema_json = json.dumps( + reset_schema_payload = json.dumps( schema["reset_observation"], allow_nan=False, ensure_ascii=False, separators=(",", ":"), ) + if validation_token and validation_token in reset_schema_payload: + raise ValueError("schema contains validation credentials") + reset_schema_json = reset_schema_payload endpoint = urlsplit(base_url) ws_url = urlunsplit( @@ -158,13 +180,94 @@ 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 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 = json.loads(raw) + if not isinstance(response, dict): + raise ValueError("invalid telemetry envelope") + return response + + capability = None + if validation_token: + 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): + 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, server_code phase = operation @@ -172,17 +275,29 @@ def exchange(operation: str, data: dict | None = None) -> dict: if data is not None: request["data"] = data request_json = json.dumps(request, allow_nan=False) - _bounded_call( - connection, lambda: connection.send(request_json), remaining() - ) - raw = connection.recv(timeout=remaining()) + raw = receive_response(request_json) if not isinstance(raw, str): raise ValueError("binary response is not the JSON protocol") + if len(raw.encode("utf-8")) > MAX_MESSAGE_BYTES: + raise ValueError("response exceeds size bound") trace_bytes += len(raw.encode("utf-8")) + len( request_json.encode("utf-8") ) if trace_bytes > MAX_TRACE_BYTES: raise ValueError("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): + # Malformed JSON cannot be checked for escaped credentials. + if validation_token: + raise + parsed = None + if contains_credential(parsed): + raise ValueError("response contains validation credentials") exchanges.append( WireExchange( operation=operation, @@ -192,14 +307,7 @@ def exchange(operation: str, data: dict | None = None) -> dict: ) response = json.loads(raw) if isinstance(response, dict) and response.get("type") == "error": - data = response.get("data") - code = data.get("code") if isinstance(data, dict) else None - # Only protocol constants are safe diagnostics; never echo an - # arbitrary server message or a subject-defined error code. - if isinstance(code, str) and code in { - item.value for item in WSErrorCode - }: - server_code = code + server_code = _server_error_code(response) raise ValueError("server returned an error") expected = "state" if operation == "state" else "observation" if ( @@ -219,6 +327,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 @@ -239,6 +370,8 @@ def exchange(operation: str, data: dict | None = None) -> dict: return RuntimeEvidence( exchanges=tuple(exchanges), observation_schema_json=schema_json, + telemetry_json=telemetry_json, + telemetry_error=telemetry_error, reset_observation_schema_json=reset_schema_json, ) except KeyboardInterrupt: @@ -249,6 +382,8 @@ def exchange(operation: str, data: dict | None = None) -> dict: reset_observation_schema_json=reset_schema_json, failure_phase=phase, failure_reason=f"{phase} failed (KeyboardInterrupt)", + telemetry_json=telemetry_json, + telemetry_error=telemetry_error, ) ) from None except Exception as exc: @@ -259,4 +394,6 @@ def exchange(operation: str, data: dict | None = None) -> dict: reset_observation_schema_json=reset_schema_json, failure_phase=phase, failure_reason=f"{phase} failed ({server_code or type(exc).__name__})", + telemetry_json=telemetry_json, + telemetry_error=telemetry_error, ) diff --git a/src/openenv/validation/runtime/contracts.py b/src/openenv/validation/runtime/contracts.py index 9440783f96..c886edb8ce 100644 --- a/src/openenv/validation/runtime/contracts.py +++ b/src/openenv/validation/runtime/contracts.py @@ -265,6 +265,10 @@ class RuntimeEvidence: Collector phase that failed; a truncated transcript cannot pass silently. failure_reason (`str`, *optional*): Credential-safe explanation of the collection failure. + telemetry_json (`str`, *optional*): + Subject-emitted session snapshot, independent of the wire transcript. + telemetry_error (`str`, *optional*): + Bounded telemetry failure without invalidating completed wire evidence. reset_observation_schema_json (`str`, *optional*): Explicit `/schema` reset_observation value; absent means use the step schema. """ @@ -273,4 +277,6 @@ class RuntimeEvidence: observation_schema_json: str | None = None failure_phase: str | None = None failure_reason: str | None = None + telemetry_json: str | None = None + telemetry_error: str | None = None reset_observation_schema_json: str | None = None diff --git a/tests/fixtures/validation/runtime/served_probe/app.py b/tests/fixtures/validation/runtime/served_probe/app.py index 586e433f36..f864ccff51 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..f7e985e50b --- /dev/null +++ b/tests/test_validation/integration/test_runtime_process.py @@ -0,0 +1,220 @@ +"""Real HTTP/WebSocket evidence with installed OpenEnv and a test-only process. + +This covers protocol and grading on CPU Jobs. It does not establish Docker, +resource-limit, network-isolation or fresh-container acceptance. +""" + +import hashlib +import importlib.metadata +import json +import os +import shutil +import signal +import socket +import subprocess +import sys +import sysconfig +import time +from pathlib import Path + +import httpx +import pytest +from openenv.validation.providers import StartupError, UnsupportedCapability +from openenv.validation.runner import run_validation +from openenv.validation.types import CheckStatus, Level, ProviderCapability + +FIXTURE = Path(__file__).parents[2] / "fixtures/validation/runtime/served_probe" +SERVER = """ +import importlib.util, os, sys, uvicorn +spec = importlib.util.spec_from_file_location('process_probe', sys.argv[1]) +module = importlib.util.module_from_spec(spec) +spec.loader.exec_module(module) +uvicorn.run(module.make_app(os.environ['VALIDATION_FAULT']), + fd=int(sys.argv[2]), log_level='warning') +""" + + +class ProcessSubject: + def __init__(self, process, port, log_path, record_sha256): + self.process = process + self.port = port + self.log_path = log_path + self.record_sha256 = record_sha256 + self.base_url = f"http://127.0.0.1:{port}" + + def inspect(self): + return { + "test_only": True, + "isolation": "process", + "pid": self.process.pid, + "installed_record_sha256": self.record_sha256, + "container_build_exercised": False, + "resource_limits_enforced": False, + "network_policy_enforced": False, + } + + def logs(self, max_bytes=65536): + with self.log_path.open("rb") as stream: + stream.seek(max(0, self.log_path.stat().st_size - max_bytes)) + return stream.read(max_bytes).decode("utf-8", "replace") + + def exec(self, argv, timeout_s): + raise UnsupportedCapability("test process provider has no sandbox exec") + + def stop(self): + if self.process.poll() is None: + os.killpg(self.process.pid, signal.SIGTERM) + try: + self.process.wait(timeout=3) + except subprocess.TimeoutExpired: + os.killpg(self.process.pid, signal.SIGKILL) + self.process.wait(timeout=3) + assert self.process.poll() is not None + with socket.socket() as probe: + probe.settimeout(1) + assert probe.connect_ex(("127.0.0.1", self.port)) != 0 + + +class ProcessProvider: + """Substitute only process launch; all wire collection and grading are real. + + IMAGE_BUILD satisfies the runner's lifecycle seam in this test only. The + reference identifies installed wheel metadata, never a purported Docker image. + """ + + name = "test-process" + capabilities = frozenset({ProviderCapability.IMAGE_BUILD}) + supported_network_modes = frozenset({"public"}) + + def __init__(self, work, mode="good"): + self.work = work + self.work.mkdir(parents=True) + self.mode = mode + self.subjects = [] + installed = next( + item + for item in importlib.metadata.distributions( + path=[sysconfig.get_path("purelib")] + ) + if item.metadata["Name"] == "openenv" + ) + self.record_sha256 = hashlib.sha256( + installed.read_text("RECORD").encode() + ).hexdigest() + + def build(self, root, execution): + self.package = self.work / "fixture" + shutil.copytree( + root, self.package, ignore=shutil.ignore_patterns("__pycache__") + ) + return "sha256:" + self.record_sha256 + + def start(self, spec): + assert spec.network.mode == "public" and spec.resources.gpus == 0 + log_path = self.work / f"server-{len(self.subjects)}.log" + with socket.socket() as listener, log_path.open("wb") as log: + listener.bind(("127.0.0.1", 0)) + port = listener.getsockname()[1] + process = subprocess.Popen( + [ + sys.executable, + "-I", + "-B", + "-c", + SERVER, + str(self.package / "app.py"), + str(listener.fileno()), + ], + stdin=subprocess.DEVNULL, + stdout=log, + stderr=subprocess.STDOUT, + env={ + "PATH": os.defpath, + "GRADIO_ANALYTICS_ENABLED": "False", + "VALIDATION_FAULT": self.mode, + **spec.env_vars, + }, + pass_fds=(listener.fileno(),), + start_new_session=True, + ) + subject = ProcessSubject(process, port, log_path, self.record_sha256) + self.subjects.append(subject) + deadline = time.monotonic() + min(spec.startup_timeout_s, 10) + try: + with httpx.Client(trust_env=False, timeout=0.2) as client: + while process.poll() is None and time.monotonic() < deadline: + try: + if client.get(subject.base_url + "/health").status_code == 200: + return subject + except httpx.HTTPError: + pass # The process may still be starting its HTTP listener. + time.sleep(0.02) + raise StartupError("test process failed readiness") + except BaseException: + subject.stop() + raise + + +@pytest.mark.parametrize( + "mode,failed_check", + [ + ("good", None), + ("bad_reward", "runtime.reward_well_formed"), + ("bad_observation", "runtime.observation_schema"), + ("missing_done", "runtime.observation_schema"), + ("bad_state", "runtime.state_contract"), + ("startup_failure", "runtime.startup"), + ], +) +def test_installed_server_collector_and_graders_over_loopback( + tmp_path, mode, failed_check +): + artifacts = ( + Path(os.environ.get("OPENENV_VALIDATION_ARTIFACTS", tmp_path)) + / "process" + / mode + ) + provider = ProcessProvider(artifacts / "subject", mode) + try: + report = run_validation( + FIXTURE, + max_level=Level.RUNTIME, + provider=provider, + artifacts_dir=artifacts / "report", + ) + finally: + for subject in provider.subjects: + subject.stop() + checks = {result.check_id: result for result in report.results} + assert checks["static.manifest"].status is CheckStatus.PASS + if failed_check: + assert checks[failed_check].status is CheckStatus.FAIL + else: + for check in ( + "startup", + "reward_well_formed", + "observation_schema", + "state_contract", + ): + assert checks[f"runtime.{check}"].status is CheckStatus.PASS + if mode != "startup_failure": + trace = json.loads((artifacts / "report/collector-trace.json").read_text()) + assert sum(row["operation"] == "step" for row in trace) == 2 + manifest = json.loads((artifacts / "report/run-manifest.json").read_text()) + assert manifest["provider"]["isolation"] == "process" + assert manifest["provider"]["container_build_exercised"] is False + telemetry_path = artifacts / "report/session-telemetry.json" + telemetry = json.loads(telemetry_path.read_text()) + assert telemetry["seed"]["accepted"] is True + assert len(telemetry["trajectory"]["records"]) == len(trace) + assert len(telemetry["attribution"]) == 2 + assert telemetry["trajectory"]["complete"] is True + for line in (artifacts / "report/SHA256SUMS").read_text().splitlines(): + checksum, name = line.split(" ", 1) + assert ( + hashlib.sha256((artifacts / "report" / name).read_bytes()).hexdigest() + == checksum + ) + assert provider.subjects and all( + subject.process.poll() is not None for subject in provider.subjects + ) diff --git a/tests/test_validation/integration/test_session_telemetry_protocol.py b/tests/test_validation/integration/test_session_telemetry_protocol.py new file mode 100644 index 0000000000..e20c30d3a9 --- /dev/null +++ b/tests/test_validation/integration/test_session_telemetry_protocol.py @@ -0,0 +1,339 @@ +"""Exercise the real replay WebSocket and its separate production MCP boundary.""" + +import json +import socket +import threading +import time + +import pytest +import uvicorn +from fastapi import FastAPI +from fastapi.testclient import TestClient +from fastmcp import FastMCP +from openenv.core.env_server.http_server import create_app, HTTPEnvServer +from openenv.core.env_server.interfaces import Environment +from openenv.core.env_server.types import Action, Observation, State +from openenv.core.rubrics import Rubric, WeightedSum +from openenv.validation.runtime import collector +from openenv.validation.runtime.contracts import RuntimePlan + +TOKEN = "per-run-validation-test-token-000000000000" + + +class ValueAction(Action): + value: float + + +class ValueScore(Rubric): + def forward(self, action, observation): + return action.value + + def validation_config(self): + return {} + + +class SessionEnv(Environment): + SUPPORTS_CONCURRENT_SESSIONS = True + + def __init__(self): + super().__init__(rubric=WeightedSum([ValueScore(), ValueScore()], [0.25, 0.75])) + self.count = 0 + self.seed = None + self.mcp_server = FastMCP("telemetry-test") + + def reset(self, seed=None): + self.seed, self.count = seed, 0 + return Observation(reward=0.0, metadata={"seed": seed}) + + def step(self, action): + self.count += 1 + observation = Observation(metadata={"count": self.count}) + observation.reward = self.rubric(action, observation) + return observation + + @property + def state(self): + return State(episode_id="same-session", step_count=self.count) + + +class DropsSeed(SessionEnv): + def reset(self): + return super().reset() + + +class KwargSeed(SessionEnv): + def reset(self, **kwargs): + return super().reset(seed=kwargs.get("seed")) + + +class AsyncSeed(SessionEnv): + async def reset_async(self, seed=None): + return super().reset(seed=seed) + + +class AsyncDropsSeed(SessionEnv): + async def reset_async(self): + return super().reset() + + +@pytest.mark.parametrize("closed_before_send", [False, True]) +def test_factory_error_survives_real_telemetry_handshake( + monkeypatch, closed_before_send +): + class BrokenFactory(SessionEnv): + def __init__(self): + # Fail before base initialization to exercise factory errors. + raise RuntimeError(TOKEN) + + monkeypatch.setenv("OPENENV_VALIDATION_TOKEN", TOKEN) + monkeypatch.setenv("ENABLE_WEB_INTERFACE", "false") + app = create_app(BrokenFactory, ValueAction, Observation) + server = uvicorn.Server(uvicorn.Config(app, log_level="critical")) + with socket.socket() as listener: + listener.bind(("127.0.0.1", 0)) + port = listener.getsockname()[1] + thread = threading.Thread(target=server.run, kwargs={"sockets": [listener]}) + thread.start() + try: + deadline = time.monotonic() + 5 + while not server.started: + assert thread.is_alive() and time.monotonic() < deadline + time.sleep(0.01) + if closed_before_send: + real_connect = collector.connect + + def connect_after_close(*args, **kwargs): + connection = real_connect(*args, **kwargs) + deadline = time.monotonic() + 2 + while connection.close_code is None: + assert time.monotonic() < deadline + time.sleep(0.01) + return connection + + monkeypatch.setattr(collector, "connect", connect_after_close) + plan = RuntimePlan.model_validate( + { + "plan_schema_version": "1", + "reset": {"episode_id": "factory-failure", "seed": 1}, + "actions": [{"value": 0.5}], + } + ) + evidence = collector.collect_runtime_evidence( + f"http://127.0.0.1:{port}", + plan, + episode_timeout_s=3, + validation_token=TOKEN, + ) + assert evidence.failure_reason == "validation_open failed (FACTORY_ERROR)" + assert evidence.exchanges == () + assert evidence.telemetry_json is None + assert TOKEN not in repr(evidence) + finally: + server.should_exit = True + thread.join(timeout=5) + assert not thread.is_alive() + + +def app_for(monkeypatch, env=SessionEnv, *, enabled=True, mode="simulation"): + if enabled: + monkeypatch.setenv("OPENENV_VALIDATION_TOKEN", TOKEN) + else: + monkeypatch.delenv("OPENENV_VALIDATION_TOKEN", raising=False) + app = FastAPI() + HTTPEnvServer(env, ValueAction, Observation, max_concurrent_envs=4).register_routes( + app, mode=mode + ) + return app + + +def exchange(ws, request): + ws.send_json(request) + return ws.receive_json() + + +def authorize(ws): + response = exchange( + ws, {"type": "validation_open", "data": {"schema_version": 1, "token": TOKEN}} + ) + assert response["type"] == "validation_open", response + return response["data"]["capability"] + + +def read(ws, capability): + return exchange( + ws, + { + "type": "validation_read", + "data": {"schema_version": 1, "capability": capability}, + }, + ) + + +@pytest.mark.parametrize( + "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 d54af80eba..126f800480 100644 --- a/tests/test_validation/test_runtime_artifacts.py +++ b/tests/test_validation/test_runtime_artifacts.py @@ -230,6 +230,19 @@ def test_malformed_wire_omission_is_visible_in_metadata(tmp_path): ] +@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() + ) + + def test_redaction_preserves_token_metadata_and_filters_secret_keys(): from openenv.validation.runtime.artifacts import _redact diff --git a/tests/test_validation/test_runtime_collector.py b/tests/test_validation/test_runtime_collector.py index 6c9ee8f2f1..53dd8b7ae0 100644 --- a/tests/test_validation/test_runtime_collector.py +++ b/tests/test_validation/test_runtime_collector.py @@ -112,6 +112,27 @@ def test_uncompressed_schema_still_obeys_total_byte_budget(monkeypatch, plan): assert evidence.observation_schema_json is None +@pytest.mark.parametrize("schema_key", ["observation", "reset_observation"]) +@pytest.mark.parametrize("escaped", [False, True]) +def test_schema_cannot_persist_validation_credential( + monkeypatch, plan, schema_key, escaped +): + token = "validation-secret-value-" * 2 + schemas = {"observation": {"type": "object"}} + schemas[schema_key] = {"properties": {token: {"description": token}}} + payload = json.dumps(schemas) + if escaped: + payload = payload.replace(token, "".join(f"\\u{ord(c):04x}" for c in token)) + stream = TrackedStream(payload.encode()) + schema_transport(monkeypatch, stream) + evidence = collector.collect_runtime_evidence( + "http://127.0.0.1:8000", plan, episode_timeout_s=2, validation_token=token + ) + assert getattr(evidence, f"{schema_key}_schema_json") is None + assert evidence.failure_phase == "schema" + assert token not in str(evidence) + + @pytest.mark.parametrize("slow_phase", ["headers", "body"]) def test_schema_deadline_aborts_trickling_http_server(plan, slow_phase): body = json.dumps({"observation": {"type": "object"}}).encode() 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..cf16e60c77 --- /dev/null +++ b/tests/test_validation/test_runtime_telemetry_collector.py @@ -0,0 +1,263 @@ +"""Telemetry cannot expand its byte budget or retain replay credentials.""" + +import json + +import httpx +import pytest +from openenv.validation.runtime import collector +from openenv.validation.runtime.contracts import RuntimePlan +from websockets.exceptions import ConnectionClosedError + +TOKEN = "run-authorization-" + "x" * 32 +CAPABILITY = "socket-capability-" + "y" * 32 + + +def collect( + monkeypatch, + *, + snapshot=None, + leaked_response=None, + leak_operation="reset", + escaped=False, + malformed=False, + opening_reply=None, + opening_error=None, + send_error=None, +): + original = httpx.Client + monkeypatch.setattr( + collector.httpx, + "Client", + lambda **kwargs: original( + transport=httpx.MockTransport( + lambda request: httpx.Response(200, json={"observation": {}}) + ), + **kwargs, + ), + ) + + class Connection: + def send(self, raw): + self.request = json.loads(raw) + if self.request["type"] == "validation_open" and send_error is not None: + raise send_error + + def recv(self, timeout): + operation = self.request["type"] + if operation == "validation_open": + if opening_error is not None: + raise opening_error + if opening_reply is not None: + return opening_reply + response = { + "type": operation, + "data": {"schema_version": 1, "capability": CAPABILITY}, + } + elif operation == "validation_read": + 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 + if operation == leak_operation + else None + }, + "done": False, + "reward": 0.0, + }, + } + raw = json.dumps(response, ensure_ascii=False, separators=(",", ":")) + if escaped and leaked_response: + raw = raw.replace( + leaked_response, + "".join(f"\\u{ord(c):04x}" for c in leaked_response), + ) + return raw[:-1] if malformed and operation == leak_operation else raw + + def close(self): + pass + + monkeypatch.setattr(collector, "connect", lambda *args, **kwargs: Connection()) + plan = RuntimePlan.model_validate( + { + "plan_schema_version": "1", + "reset": {"episode_id": "test", "seed": 1}, + "actions": [{"increment": 1}], + } + ) + return collector.collect_runtime_evidence( + "http://127.0.0.1:8000", plan, episode_timeout_s=2, validation_token=TOKEN + ) + + +def test_utf8_telemetry_stays_inside_received_budget(monkeypatch): + monkeypatch.setattr(collector, "MAX_TRACE_BYTES", 2048) + result = collect(monkeypatch, snapshot={"schema_version": 1, "text": "😀" * 400}) + assert result.telemetry_error is None + assert len(result.telemetry_json.encode()) <= 2048 + assert "😀" in result.telemetry_json + + +@pytest.mark.parametrize("credential", [TOKEN, CAPABILITY]) +def test_telemetry_cannot_persist_credentials_under_arbitrary_keys( + monkeypatch, credential +): + result = collect(monkeypatch, snapshot={"schema_version": 1, "debug": [credential]}) + assert result.telemetry_json is None + assert result.telemetry_error == "session telemetry failed (ValueError)" + assert result.failure_reason is None + assert len(result.exchanges) == 4 + assert credential not in repr(result) + + +@pytest.mark.parametrize( + "credential,escaped", + [(TOKEN, False), (CAPABILITY, False), (TOKEN, True), (CAPABILITY, True)], +) +def test_normal_wire_cannot_persist_plain_or_escaped_credentials( + monkeypatch, credential, escaped +): + result = collect(monkeypatch, leaked_response=credential, escaped=escaped) + assert result.exchanges == () + assert result.failure_reason == "reset failed (ValueError)" + assert credential not in repr(result) + + +@pytest.mark.parametrize( + "reply,error_name", + [ + (TOKEN.encode(), "ValueError"), + ("not-json-" + TOKEN, "JSONDecodeError"), + (json.dumps([TOKEN]), "ValueError"), + ], + ids=["binary", "non-json", "list"], +) +def test_malformed_opening_reply_keeps_the_ordinary_session( + monkeypatch, reply, error_name +): + result = collect(monkeypatch, opening_reply=reply) + assert result.telemetry_error == f"session telemetry failed ({error_name})" + assert result.telemetry_json is None + assert result.failure_reason is None + assert [row.operation for row in result.exchanges] == [ + "reset", + "state", + "step", + "state", + ] + assert TOKEN not in repr(result) + + +@pytest.mark.parametrize( + "error", [ConnectionClosedError(None, None), TimeoutError(TOKEN)] +) +def test_opening_transport_failure_still_aborts_collection(monkeypatch, error): + result = collect(monkeypatch, opening_error=error) + assert result.failure_phase == "validation_open" + assert result.failure_reason == f"validation_open failed ({type(error).__name__})" + assert result.telemetry_json is None + assert not result.exchanges + assert TOKEN not in repr(result) + + +def test_opening_cancellation_is_not_downgraded_to_optional_telemetry(monkeypatch): + with pytest.raises(collector.RuntimeCollectionInterrupted) as error: + collect(monkeypatch, opening_error=KeyboardInterrupt(TOKEN)) + assert error.value.evidence.failure_phase == "validation_open" + assert not error.value.evidence.exchanges + assert TOKEN not in repr(error.value.evidence) + + +@pytest.mark.parametrize("code", ["FACTORY_ERROR", "CAPACITY_REACHED", "SESSION_ERROR"]) +@pytest.mark.parametrize("closed_before_send", [False, True]) +def test_terminal_opening_error_preserves_only_its_safe_code( + monkeypatch, code, closed_before_send +): + result = collect( + monkeypatch, + opening_reply=json.dumps( + {"type": "error", "data": {"code": code, "message": TOKEN}} + ), + send_error=ConnectionClosedError(None, None) if closed_before_send else None, + ) + assert result.failure_reason == f"validation_open failed ({code})" + assert result.exchanges == () + assert result.telemetry_json is None + assert TOKEN not in repr(result) + + +@pytest.mark.parametrize("code", ["UNKNOWN_TYPE", "VALIDATION_ERROR", TOKEN]) +def test_optional_opening_refusal_does_not_fail_the_episode(monkeypatch, code): + result = collect( + monkeypatch, + opening_reply=json.dumps( + {"type": "error", "data": {"code": code, "message": TOKEN}} + ), + ) + assert result.failure_reason is None + assert result.telemetry_json is None + assert ( + result.telemetry_error + == "session telemetry unavailable or authorization refused" + ) + assert [row.operation for row in result.exchanges] == [ + "reset", + "state", + "step", + "state", + ] + assert TOKEN not in repr(result) + + +@pytest.mark.parametrize( + "reply", + [ + "not-json-" + TOKEN, + json.dumps( + { + "type": "validation_open", + "data": {"schema_version": 1, "capability": CAPABILITY}, + } + ), + ], +) +def test_closed_send_cannot_become_an_optional_or_successful_handshake( + monkeypatch, reply +): + result = collect( + monkeypatch, + opening_reply=reply, + send_error=ConnectionClosedError(None, None), + ) + assert result.failure_reason == "validation_open failed (ConnectionClosedError)" + assert result.exchanges == () + assert result.telemetry_json is None + assert TOKEN not in repr(result) + + +@pytest.mark.parametrize("credential", [TOKEN, CAPABILITY]) +@pytest.mark.parametrize("operation", ["reset", "step"]) +def test_malformed_wire_cannot_retain_escaped_credentials( + monkeypatch, credential, operation +): + result = collect( + monkeypatch, + leaked_response=credential, + leak_operation=operation, + escaped=True, + malformed=True, + ) + assert [row.operation for row in result.exchanges] == ( + [] if operation == "reset" else ["reset", "state"] + ) + assert result.failure_reason == f"{operation} failed (JSONDecodeError)" 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 6803bb4e77..3612f913e0 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 f68cc85901..3c4cae5759 100644 --- a/tests/validation_runtime/acceptance.json +++ b/tests/validation_runtime/acceptance.json @@ -9,7 +9,27 @@ "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_real_echo_reset_schema_and_unscored_rewards[False]", - "tests.test_validation.integration.test_served_probe::test_real_echo_reset_schema_and_unscored_rewards[True]" + "tests.test_validation.integration.test_served_probe::test_real_echo_reset_schema_and_unscored_rewards[True]", + "tests.test_validation.integration.test_session_telemetry_protocol::test_factory_error_survives_real_telemetry_handshake[False]", + "tests.test_validation.integration.test_session_telemetry_protocol::test_factory_error_survives_real_telemetry_handshake[True]", + "tests.test_validation.integration.test_session_telemetry_protocol::test_same_session_seed_scores_and_subject_record[SessionEnv-True]", + "tests.test_validation.integration.test_session_telemetry_protocol::test_same_session_seed_scores_and_subject_record[DropsSeed-False]", + "tests.test_validation.integration.test_session_telemetry_protocol::test_same_session_seed_scores_and_subject_record[AsyncSeed-True]", + "tests.test_validation.integration.test_session_telemetry_protocol::test_same_session_seed_scores_and_subject_record[AsyncDropsSeed-False]", + "tests.test_validation.integration.test_session_telemetry_protocol::test_same_session_seed_scores_and_subject_record[KwargSeed-True]", + "tests.test_validation.integration.test_session_telemetry_protocol::test_validation_is_opt_in_and_simulation_only[False-simulation]", + "tests.test_validation.integration.test_session_telemetry_protocol::test_validation_is_opt_in_and_simulation_only[True-production]", + "tests.test_validation.integration.test_session_telemetry_protocol::test_missing_wrong_malformed_cross_socket_and_expired_capabilities", + "tests.test_validation.integration.test_session_telemetry_protocol::test_late_open_cannot_discard_prior_replay_operations", + "tests.test_validation.integration.test_session_telemetry_protocol::test_production_mcp_has_no_telemetry_or_reset_tools", + "tests.test_validation.integration.test_session_telemetry_protocol::test_absent_or_uninspectable_rubric_does_not_hide_subject_record[False]", + "tests.test_validation.integration.test_session_telemetry_protocol::test_absent_or_uninspectable_rubric_does_not_hide_subject_record[True]", + "tests.test_validation.integration.test_runtime_process::test_installed_server_collector_and_graders_over_loopback[good-None]", + "tests.test_validation.integration.test_runtime_process::test_installed_server_collector_and_graders_over_loopback[bad_reward-runtime.reward_well_formed]", + "tests.test_validation.integration.test_runtime_process::test_installed_server_collector_and_graders_over_loopback[bad_observation-runtime.observation_schema]", + "tests.test_validation.integration.test_runtime_process::test_installed_server_collector_and_graders_over_loopback[missing_done-runtime.observation_schema]", + "tests.test_validation.integration.test_runtime_process::test_installed_server_collector_and_graders_over_loopback[bad_state-runtime.state_contract]", + "tests.test_validation.integration.test_runtime_process::test_installed_server_collector_and_graders_over_loopback[startup_failure-runtime.startup]" ], "docker": [ "tests.test_validation.integration.test_docker_lifecycle::test_docker_lifecycle_effective_limits_and_owned_cleanup",