From 071ffd287f2af9120c56eab0b7c1b15ffff4dc96 Mon Sep 17 00:00:00 2001 From: burtenshaw Date: Thu, 24 Sep 2026 12:20:24 +0200 Subject: [PATCH 01/11] feat: validate runtime repeatability --- rfcs/008-environment-auto-validation.md | 15 +- .../graders/runtime/repeatability.py | 305 +++++++++++++++++ src/openenv/validation/providers/docker.py | 5 +- src/openenv/validation/runner.py | 44 ++- src/openenv/validation/runtime/artifacts.py | 50 +++ src/openenv/validation/runtime/contracts.py | 12 + src/openenv/validation/runtime/replay.py | 228 +++++++++++++ src/openenv/validation/types.py | 1 + .../validation/runtime/served_probe/app.py | 57 +++- .../integration/test_runtime_cli.py | 83 ++++- .../integration/test_runtime_process.py | 26 +- .../test_validation/test_runtime_artifacts.py | 131 +++++++- .../test_validation/test_runtime_execution.py | 122 ++++++- .../test_runtime_repeatability.py | 288 ++++++++++++++++ tests/test_validation/test_runtime_replay.py | 307 ++++++++++++++++++ tests/validation_runtime/README.md | 12 +- tests/validation_runtime/acceptance.json | 13 +- 17 files changed, 1661 insertions(+), 38 deletions(-) create mode 100644 src/openenv/validation/graders/runtime/repeatability.py create mode 100644 src/openenv/validation/runtime/replay.py create mode 100644 tests/test_validation/test_runtime_repeatability.py create mode 100644 tests/test_validation/test_runtime_replay.py diff --git a/rfcs/008-environment-auto-validation.md b/rfcs/008-environment-auto-validation.md index 6ad227f5f0..f49732d083 100644 --- a/rfcs/008-environment-auto-validation.md +++ b/rfcs/008-environment-auto-validation.md @@ -571,7 +571,7 @@ limitations. Alternate DNS, unmatched addresses and unsupported address families must fail closed. This requires a separate reviewed enforcement implementation; parsing these declarations does not claim they are enforced. -### Later runtime evidence contracts +### Runtime replay evidence (PR5) Seed acceptance and empirical determinism are separate findings. A reset that silently drops its seed does not establish seed control, while a deterministic @@ -583,9 +583,18 @@ replays and population reward variance in reward-squared units, compared to the declared bound. The total run budget bounds all samples; fewer than 20 is incomplete. This procedure is a runtime check, not a statistical confidence claim. +The initial implementation compares the baseline against a new session and an +independently inspected new container, and separately requests a different seed. +The judged sample count includes the completed baseline. Container identity must +change while image identity remains fixed. There are no volatile-field exclusions +in this version. Collection shares a 300-second deadline and retains at most +32 MiB across baseline and replay evidence. Missing samples, missing provider +capabilities and failed cleanup cannot produce a passing determinism finding. +`replays.json` preserves completed traces, telemetry, identity and cleanup outcomes. + Session telemetry for seed handling, named rubric/configuration, child attribution -and subject-emitted record references is orchestrator-only. A future protocol -slice must authorize access with an opt-in, random per-run/per-session capability +and subject-emitted record references is orchestrator-only. The protocol +authorizes access with an opt-in, random per-run/per-session capability 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. diff --git a/src/openenv/validation/graders/runtime/repeatability.py b/src/openenv/validation/graders/runtime/repeatability.py new file mode 100644 index 0000000000..2812d28f71 --- /dev/null +++ b/src/openenv/validation/graders/runtime/repeatability.py @@ -0,0 +1,305 @@ +"""Seed acceptance, independent records and bounded fresh replay comparisons.""" + +import json +import math +import statistics +import time + +from ...report import CheckResult +from ...types import CheckStatus +from .basic import _RuntimeGrader + +JUDGED_REPLAYS = 20 + + +def _trace(evidence): + trace = [ + { + "operation": row.operation, + "request": json.loads(row.request_json), + "response": json.loads(row.response_json), + } + for row in evidence.exchanges + ] + for row in trace: + json.dumps(row, allow_nan=False) + operation, request, response = ( + row["operation"], + row["request"], + row["response"], + ) + if ( + operation not in {"reset", "step", "state"} + or not isinstance(request, dict) + or request.get("type") != operation + or not isinstance(response, dict) + or response.get("type") + != ("state" if operation == "state" else "observation") + or not isinstance(response.get("data"), dict) + ): + raise ValueError("malformed trace envelope") + return trace + + +def _difference(expected, observed, path="$"): + """First differing JSON path, without disclosing subject-controlled values.""" + if type(expected) is not type(observed): + return path + if isinstance(expected, dict): + for key in sorted(expected.keys() | observed.keys()): + child = f"{path}.{key[:80]}" + if key not in expected or key not in observed: + return child + mismatch = _difference(expected[key], observed[key], child) + if mismatch: + return mismatch + elif isinstance(expected, list): + for index, (left, right) in enumerate(zip(expected, observed)): + mismatch = _difference(left, right, f"{path}[{index}]") + if mismatch: + return mismatch + if len(expected) != len(observed): + return f"{path}[{min(len(expected), len(observed))}]" + elif expected != observed: + return path + return None + + +def _telemetry(evidence): + value = json.loads(evidence.telemetry_json) + if not isinstance(value, dict) or type(value.get("schema_version")) is not int: + raise ValueError("malformed telemetry") + if value["schema_version"] != 1: + raise ValueError("unsupported telemetry version") + return value + + +def _missing_telemetry(grader, evidence): + if evidence is not None and evidence.telemetry_json is None: + failure = evidence.failure_reason or evidence.telemetry_error + return CheckResult( + check_id=grader.check_id, + status=CheckStatus.FAIL if failure else CheckStatus.SKIP, + evidence=[failure or "session telemetry is unavailable"], + duration_s=0, + ) + return None + + +class SeedControlGrader(_RuntimeGrader): + """Require observed seed forwarding for the original and scheduled resets.""" + + check_id = "runtime.seed_control" + + def run(self, subject): + evidence = subject.runtime_evidence + result = _missing_telemetry(self, evidence) or super().run(subject) + if result.status is CheckStatus.PASS: + if not any(replay.scope == "seed" for replay in evidence.replays): + return result.model_copy( + update={ + "status": CheckStatus.SKIP, + "evidence": [ + evidence.replay_failure_reason + or "different-seed reset experiment is unavailable" + ], + } + ) + return result + + def check(self, subject, evidence): + problems = [] + original_seed = None + for index, sample in enumerate( + [evidence] + [replay.evidence for replay in evidence.replays] + ): + if sample.failure_reason or sample.telemetry_error: + problems.append(f"replay {index}: reset or telemetry collection failed") + continue + if sample.telemetry_json is None: + problems.append(f"replay {index}: seed telemetry unavailable") + continue + resets = [row for row in sample.exchanges if row.operation == "reset"] + if len(resets) != 1: + problems.append(f"replay {index}: expected one measured reset") + continue + seed = json.loads(resets[0].request_json)["data"]["seed"] + if type(seed) is not int: + problems.append(f"replay {index}: requested seed is not an integer") + continue + if index == 0: + original_seed = seed + elif evidence.replays[index - 1].scope == "seed" and seed == original_seed: + problems.append(f"replay {index}: scheduled seed was not changed") + observed = _telemetry(sample).get("seed") + if ( + not isinstance(observed, dict) + or observed.get("requested") is not True + or observed.get("accepted") is not True + or type(observed.get("value")) is not int + or observed["value"] != seed + ): + problems.append(f"replay {index}: seed was not observed as forwarded") + return problems + + +class TrajectoryRecordGrader(_RuntimeGrader): + """Compare a subject-emitted record with the independent collector transcript.""" + + check_id = "runtime.trajectory_record" + + def run(self, subject): + return _missing_telemetry(self, subject.runtime_evidence) or super().run( + subject + ) + + def check(self, subject, evidence): + if evidence.telemetry_error: + return [evidence.telemetry_error] + record = _telemetry(evidence).get("trajectory") + if not isinstance(record, dict): + return ["subject-emitted trajectory record is missing"] + if ( + type(record.get("schema_version")) is not int + or record["schema_version"] != 1 + or record.get("source") != "openenv-server" + ): + return ["subject-emitted trajectory record metadata is invalid"] + if record.get("complete") is not True: + return ["subject-emitted trajectory record is incomplete"] + if not isinstance(record.get("records"), list): + return ["subject-emitted trajectory records are malformed"] + mismatch = _difference(_trace(evidence), record["records"]) + return [f"subject record differs at {mismatch}"] if mismatch else [] + + +class EpisodeDeterminismGrader(_RuntimeGrader): + """Compare full traces, or per-step reward population variance for a judge.""" + + check_id = "runtime.episode_determinism" + + def run(self, subject): + started = time.monotonic() + evidence = subject.runtime_evidence + if evidence is None: + return super().run(subject) + measured = {} + try: + status, messages, measured = self._grade(subject, evidence) + except (ValueError, TypeError, KeyError, RecursionError, OverflowError): + status, messages = CheckStatus.FAIL, ["malformed replay evidence"] + return CheckResult( + check_id=self.check_id, + status=status, + evidence=messages, + measured=measured, + duration_s=time.monotonic() - started, + ) + + def _grade(self, subject, evidence): + if any( + row.scope not in {"session", "container", "seed"} + for row in evidence.replays + ): + return CheckStatus.FAIL, ["invalid replay scope"], {} + replays = [row for row in evidence.replays if row.scope != "seed"] + samples = [evidence] + [row.evidence for row in replays] + measured = { + "completed_replays": sum( + not row.failure_reason + and any(exchange.operation == "step" for exchange in row.exchanges) + for row in samples + ) + } + for index, sample in enumerate(samples): + if sample.failure_reason: + return ( + CheckStatus.FAIL, + [f"replay {index}: collection failed"], + measured, + ) + if not any(row.operation == "step" for row in sample.exchanges): + return ( + CheckStatus.SKIP, + [ + evidence.replay_failure_reason + or f"replay {index}: no step was observed" + ], + measured, + ) + judged = subject.manifest.capabilities.llm_judged + required = JUDGED_REPLAYS if judged else 3 + if len(samples) < required or {row.scope for row in replays} != { + "session", + "container", + }: + return ( + CheckStatus.SKIP, + [ + evidence.replay_failure_reason + or f"requires {required} completed replays across fresh sessions and containers" + ], + measured, + ) + if judged and len(samples) != JUDGED_REPLAYS: + return ( + CheckStatus.FAIL, + ["judged procedure requires exactly 20 samples"], + measured, + ) + traces = [_trace(sample) for sample in samples] + rewards = [] + for trace in traces: + sample_rewards = [] + for row in trace: + if row["operation"] != "step": + continue + if judged: + reward = row["response"]["data"]["reward"] + low, high = subject.manifest.reward.range + if ( + type(reward) not in (int, float) + or not math.isfinite(reward) + or not low <= reward <= high + ): + return ( + CheckStatus.FAIL, + ["invalid judged replay reward"], + measured, + ) + sample_rewards.append(reward) + row["response"]["data"]["reward"] = None + rewards.append(sample_rewards) + # Policy-owned volatile exclusions are deliberately empty. All replays + # use the same episode identity; no author field can suppress a difference. + for index, trace in enumerate(traces[1:], 1): + mismatch = _difference(traces[0], trace) + if mismatch: + return ( + CheckStatus.FAIL, + [f"replay {index}: first divergence at {mismatch}"], + measured, + ) + if judged: + variances = [statistics.pvariance(values) for values in zip(*rewards)] + measured.update( + reward_population_variance=variances, + variance_units="reward_squared", + ) + bound = subject.manifest.reward.variance_tolerance + if type(bound) not in (int, float) or not math.isfinite(bound) or bound < 0: + return CheckStatus.FAIL, ["invalid declared variance bound"], measured + for index, variance in enumerate(variances): + if variance > bound: + return ( + CheckStatus.FAIL, + [ + f"step {index}: reward population variance exceeds declared bound" + ], + measured, + ) + return ( + CheckStatus.PASS, + ["fresh-session and fresh-container replays agree"], + measured, + ) diff --git a/src/openenv/validation/providers/docker.py b/src/openenv/validation/providers/docker.py index 9bb976d1f7..8c103928fc 100644 --- a/src/openenv/validation/providers/docker.py +++ b/src/openenv/validation/providers/docker.py @@ -209,7 +209,10 @@ class DockerValidationProvider: """Build isolated source snapshots and launch owned, resource-bounded subjects.""" name = "docker-local" - capabilities = frozenset({ProviderCapability.IMAGE_BUILD, ProviderCapability.EXEC}) + capabilities = frozenset({ + ProviderCapability.IMAGE_BUILD, ProviderCapability.EXEC, + ProviderCapability.FRESH_CONTAINER, + }) supported_network_modes = frozenset({"public"}) def __init__(self, *, build_timeout_s: float = 600, max_context_bytes: int = 2**30): diff --git a/src/openenv/validation/runner.py b/src/openenv/validation/runner.py index 8151dbc133..180e77d858 100644 --- a/src/openenv/validation/runner.py +++ b/src/openenv/validation/runner.py @@ -15,6 +15,11 @@ RewardWellFormedGrader, StateContractGrader, ) +from .graders.runtime.repeatability import ( + EpisodeDeterminismGrader, + SeedControlGrader, + TrajectoryRecordGrader, +) from .graders.static import StaticManifestGrader from .manifest import ManifestError, NormalizedManifest, NormalizedManifestV2 from .parsers import ParserRegistry @@ -25,6 +30,7 @@ from .runtime.artifacts import write_runtime_bundle from .runtime.collector import collect_runtime_evidence, RuntimeCollectionInterrupted from .runtime.contracts import LaunchSpec, load_runtime_plan, RuntimePlanError +from .runtime.replay import collect_replays, REPLAY_BUDGET_SECONDS from .runtime.scheduler import execute_graders from .signature import detect_signature from .types import CheckStatus, Lane, Level, ProviderCapability @@ -145,9 +151,8 @@ def _runtime(subject, *, skip_build, provider): ) attempted = True image_ref = provider.build(subject.root, manifest.execution) - running = provider.start( - LaunchSpec.model_validate({**spec.model_dump(), "image_ref": image_ref}) - ) + spec = LaunchSpec.model_validate({**spec.model_dump(), "image_ref": image_ref}) + running = provider.start(spec) cleanup = {"required": True, "completed": False} inspection = running.inspect() result = _outcome( @@ -157,10 +162,13 @@ def _runtime(subject, *, skip_build, provider): started=started, measured={"provider": provider.name, "image_ref": image_ref}, ) + replay_deadline = time.monotonic() + REPLAY_BUDGET_SECONDS evidence = collect_runtime_evidence( running.base_url, plan, - episode_timeout_s=manifest.resources.episode_timeout_s, + episode_timeout_s=min( + manifest.resources.episode_timeout_s, REPLAY_BUDGET_SECONDS + ), validation_token=spec.env_vars["OPENENV_VALIDATION_TOKEN"], ) # A health endpoint without a functioning protocol isn't a startup success. @@ -171,6 +179,18 @@ def _runtime(subject, *, skip_build, provider): evidence.failure_reason, started=started, ) + evidence = collect_replays( + provider, + running, + spec, + plan, + evidence, + capabilities=manifest.capabilities, + deadline=replay_deadline, + ) + if any(replay.cleanup_complete is False for replay in evidence.replays): + result.status = CheckStatus.ERROR + result.evidence.append("replay subject teardown failed") subject = replace( subject, image_ref=image_ref, running=running, runtime_evidence=evidence ) @@ -179,6 +199,9 @@ def _runtime(subject, *, skip_build, provider): RewardWellFormedGrader(), ObservationSchemaGrader(), StateContractGrader(), + SeedControlGrader(), + EpisodeDeterminismGrader(), + TrajectoryRecordGrader(), ], subject, provider_capabilities=provider.capabilities, @@ -222,7 +245,12 @@ def _runtime(subject, *, skip_build, provider): if running is not None: try: running.stop() - cleanup["completed"] = True + cleanup["completed"] = not ( + evidence + and any( + replay.cleanup_complete is False for replay in evidence.replays + ) + ) except (Exception, KeyboardInterrupt): cleanup["completed"] = False if result is None: @@ -358,6 +386,9 @@ def run_validation( "runtime.reward_well_formed", "runtime.observation_schema", "runtime.state_contract", + "runtime.seed_control", + "runtime.episode_determinism", + "runtime.trajectory_record", }: reason = "unmet dependency: runtime.startup" results.append(_outcome(entry.check_id, CheckStatus.SKIP, reason)) @@ -387,6 +418,9 @@ def run_validation( "runtime.reward_well_formed", "runtime.observation_schema", "runtime.state_contract", + "runtime.seed_control", + "runtime.episode_determinism", + "runtime.trajectory_record", } else r for r in results diff --git a/src/openenv/validation/runtime/artifacts.py b/src/openenv/validation/runtime/artifacts.py index 44585100d4..4bf43e3956 100644 --- a/src/openenv/validation/runtime/artifacts.py +++ b/src/openenv/validation/runtime/artifacts.py @@ -122,6 +122,56 @@ def write_runtime_bundle( ) files["collector-trace.json"] = trace files["collector-evidence.json"] = collector_metadata + if evidence and (evidence.replays or evidence.replay_failure_reason): + samples = [] + for replay in evidence.replays: + sample = replay.evidence + row = { + "scope": replay.scope, + "cleanup_complete": replay.cleanup_complete, + "failure_phase": sample.failure_phase, + "failure_reason": sample.failure_reason, + "telemetry_error": sample.telemetry_error, + "trace": [], + "omitted_trace_fields": [], + "omitted_evidence_fields": [], + } + for index, exchange in enumerate(sample.exchanges): + item = {"operation": exchange.operation} + for key in ("request_json", "response_json"): + try: + item[key] = json.loads(getattr(exchange, key)) + except (ValueError, RecursionError): + item[key] = "[malformed response omitted]" + row["omitted_trace_fields"].append( + {"exchange_index": index, "field": key} + ) + row["trace"].append(item) + for key, raw in ( + ("telemetry", sample.telemetry_json), + ("schema", sample.observation_schema_json), + ("provider", replay.provider_json), + ): + row[f"{key}_available"] = raw is not None + try: + row[key] = json.loads(raw) if raw is not None else None + except (ValueError, RecursionError): + row[key] = "[malformed evidence omitted]" + row["omitted_evidence_fields"].append(key) + row["redacted"] = ( + bool(row["omitted_trace_fields"]) + or bool(row["omitted_evidence_fields"]) + or _redact(row) != row + ) + samples.append(row) + files["replays.json"] = { + "schema_version": "1", + "failure_reason": evidence.replay_failure_reason, + "samples": samples, + } + for name in ("runtime-plan.json", "session-telemetry.json", "replays.json"): + if name not in files: + (directory / name).unlink(missing_ok=True) digests = [] for name, value in files.items(): payload = ( diff --git a/src/openenv/validation/runtime/contracts.py b/src/openenv/validation/runtime/contracts.py index 9df4a3eb1b..6d116f9987 100644 --- a/src/openenv/validation/runtime/contracts.py +++ b/src/openenv/validation/runtime/contracts.py @@ -277,3 +277,15 @@ class RuntimeEvidence: failure_reason: str | None = None telemetry_json: str | None = None telemetry_error: str | None = None + replays: tuple["ReplayEvidence", ...] = () + replay_failure_reason: str | None = None + + +@dataclass(frozen=True) +class ReplayEvidence: + """A fresh replay and, for containers, inspected identity and cleanup evidence.""" + + scope: Literal["session", "container", "seed"] + evidence: RuntimeEvidence + provider_json: str | None = None + cleanup_complete: bool | None = None diff --git a/src/openenv/validation/runtime/replay.py b/src/openenv/validation/runtime/replay.py new file mode 100644 index 0000000000..fca6700cb2 --- /dev/null +++ b/src/openenv/validation/runtime/replay.py @@ -0,0 +1,228 @@ +"""Bounded fresh-session and independently verified fresh-container experiments.""" + +import json +import secrets +import time +import uuid +from dataclasses import replace + +from ..types import ProviderCapability +from .collector import collect_runtime_evidence, RuntimeCollectionInterrupted +from .contracts import ReplayEvidence, RuntimeEvidence + +REPLAY_BUDGET_SECONDS = 300.0 +MAX_REPLAY_BYTES = 32 * 1024 * 1024 +JUDGED_REPLAYS = 20 + + +def _evidence_bytes(evidence): + return sum( + len(value.encode("utf-8")) + for value in ( + evidence.observation_schema_json or "", + evidence.telemetry_json or "", + *( + value + for row in evidence.exchanges + for value in (row.request_json, row.response_json) + ), + ) + ) + + +def collect_replays( + provider, running, spec, plan, evidence, *, capabilities, deadline=None +): + """ + Collect the fixed replay schedule without replacing the primary transcript. + + Args: + provider (`ValidationProvider`): + Provider used for the original subject and optional fresh container. + running (`RunningSubject`): + The original subject; its lifetime remains owned by the caller. + spec (`LaunchSpec`): + Original immutable image, resource limits and explicit authorization. + plan (`RuntimePlan`): + Fixed episode identity, seed, options and public action sequence. + evidence (`RuntimeEvidence`): + Independently collected primary episode. + capabilities (`CapabilitiesSpec`): + Manifest capabilities; judged rewards require 20 identical-input samples. + deadline (`float`, *optional*): + Absolute monotonic deadline set before primary collection. Defaults to + 300 seconds from this call; every collection and launch shares it. + + Returns: + `RuntimeEvidence`: the original episode plus immutable replay evidence. + """ + retained_bytes = _evidence_bytes(evidence) + budget_failure = f"total retained replay evidence exceeds {MAX_REPLAY_BYTES} bytes" + if retained_bytes > MAX_REPLAY_BYTES: + # Normal collector bounds keep a primary episode below this limit. A + # malformed oversized input must not bypass the final artifact budget. + return RuntimeEvidence( + failure_phase="replay evidence budget", + failure_reason=budget_failure, + replay_failure_reason=budget_failure, + ) + if evidence.failure_reason: + return evidence + deadline = min( + deadline if deadline is not None else float("inf"), + time.monotonic() + REPLAY_BUDGET_SECONDS, + ) + replays = [] + phase = "fresh session" + failure = None + + def remaining(): + budget = deadline - time.monotonic() + if budget <= 0: + raise TimeoutError("total replay deadline exceeded") + return budget + + def collect(target, reset_plan, token): + return collect_runtime_evidence( + target.base_url, + reset_plan, + episode_timeout_s=min(spec.resources.episode_timeout_s, remaining()), + validation_token=token, + ) + + def retain(scope, sample, **metadata): + nonlocal retained_bytes, failure + size = _evidence_bytes(sample) + if retained_bytes + size > MAX_REPLAY_BYTES: + failure = budget_failure + if scope == "container": + # Keep inspected identity and cleanup even when the transcript + # cannot be retained. An empty sample cannot pass determinism. + replays.append(ReplayEvidence(scope, RuntimeEvidence(), **metadata)) + return False + retained_bytes += size + replays.append(ReplayEvidence(scope, sample, **metadata)) + return True + + def session(scope, reset_plan): + try: + sample = collect( + running, reset_plan, spec.env_vars.get("OPENENV_VALIDATION_TOKEN") + ) + except RuntimeCollectionInterrupted as exc: + retain(scope, exc.evidence) + raise + if not retain(scope, sample): + raise ValueError("replay evidence byte budget exceeded") + if sample.failure_reason: + raise RuntimeError("replay collection failed") + + try: + session("session", plan) + phase = "different-seed session" + changed_seed = plan.reset.model_copy( + update={"seed": (plan.reset.seed + 1) % 2**32} + ) + session("seed", plan.model_copy(update={"reset": changed_seed})) + if ProviderCapability.FRESH_CONTAINER not in provider.capabilities: + failure = "missing provider capability: fresh_container" + else: + phase = "original container inspection" + remaining() + original = running.inspect() + if ( + not isinstance(original.get("container_id"), str) + or not original["container_id"] + or original.get("image_id") != spec.image_ref + ): + raise ValueError("original container identity unavailable") + phase = "fresh container start" + fresh_spec = spec.model_copy( + update={ + "run_id": "validation-replay-" + uuid.uuid4().hex, + "startup_timeout_s": min(spec.startup_timeout_s, remaining()), + "env_vars": { + **spec.env_vars, + "OPENENV_VALIDATION_TOKEN": secrets.token_urlsafe(32), + }, + } + ) + try: + fresh = provider.start(fresh_spec) + except (Exception, KeyboardInterrupt): + # No returned handle means this layer cannot verify teardown, + # including when the provider's startup cleanup itself failed. + retain( + "container", + RuntimeEvidence( + failure_phase=phase, + failure_reason="fresh container start failed; teardown could not be confirmed", + ), + cleanup_complete=False, + ) + raise + inspection_json = None + sample = None + cleanup = False + try: + phase = "fresh container inspection" + remaining() + inspection = fresh.inspect() + inspection_json = json.dumps(inspection, allow_nan=False) + if ( + not isinstance(inspection.get("container_id"), str) + or not inspection["container_id"] + or inspection["container_id"] == original["container_id"] + or inspection.get("image_id") != spec.image_ref + ): + raise ValueError("fresh container identity did not match") + phase = "fresh container replay" + sample = collect( + fresh, plan, fresh_spec.env_vars["OPENENV_VALIDATION_TOKEN"] + ) + except RuntimeCollectionInterrupted as exc: + sample = exc.evidence + raise + finally: + cleanup_interrupted = False + try: + fresh.stop() + cleanup = True + except KeyboardInterrupt: + cleanup_interrupted = True + failure = "fresh container cleanup interrupted" + except Exception: + failure = "fresh container cleanup failed" + if sample is None: + sample = RuntimeEvidence( + failure_phase=phase, failure_reason=f"{phase} did not complete" + ) + retained = retain( + "container", + sample, + provider_json=inspection_json, + cleanup_complete=cleanup, + ) + if cleanup_interrupted: + raise KeyboardInterrupt + if not cleanup: + raise RuntimeError("fresh container cleanup failed") + if not retained: + raise ValueError("replay evidence byte budget exceeded") + if sample.failure_reason: + raise RuntimeError("fresh container replay failed") + phase = "judged replay sampling" + if capabilities.llm_judged: + # Primary + session + container are already three identical inputs. + for _ in range(JUDGED_REPLAYS - 3): + session("session", plan) + except KeyboardInterrupt: + interrupted = replace( + evidence, + replays=tuple(replays), + replay_failure_reason=failure or f"{phase} interrupted", + ) + raise RuntimeCollectionInterrupted(interrupted) from None + except Exception as exc: + failure = failure or f"{phase} failed ({type(exc).__name__})" + return replace(evidence, replays=tuple(replays), replay_failure_reason=failure) diff --git a/src/openenv/validation/types.py b/src/openenv/validation/types.py index bb775596f1..fec440fae9 100644 --- a/src/openenv/validation/types.py +++ b/src/openenv/validation/types.py @@ -86,3 +86,4 @@ class ProviderCapability(str, Enum): IMAGE_BUILD = "image_build" GPU = "gpu" REMOTE = "remote" + FRESH_CONTAINER = "fresh_container" diff --git a/tests/fixtures/validation/runtime/served_probe/app.py b/tests/fixtures/validation/runtime/served_probe/app.py index edaf208448..e89b715002 100644 --- a/tests/fixtures/validation/runtime/served_probe/app.py +++ b/tests/fixtures/validation/runtime/served_probe/app.py @@ -29,11 +29,28 @@ def validation_config(self): return {"threshold": 2} +class ControlledJudge(Rubric): + """A fixed public test judge; this fixture performs no model inference.""" + + def __init__(self, score): + super().__init__() + self.score = score + + def forward(self, action, observation): + return self.score + + def validation_config(self): + return {"judge": "controlled-test-v1", "score": self.score} + + class ProbeEnvironment(Environment): SUPPORTS_CONCURRENT_SESSIONS = True + _reset_ordinal = 0 - def __init__(self): + def __init__(self, mode="good"): super().__init__() + self.mode = mode + self.ordinal = 0 self._state = State(episode_id="uninitialized", step_count=0) self.counter = 0 self.rubric = WeightedSum([CounterRubric(), CounterRubric()], [0.5, 0.5]) @@ -41,7 +58,21 @@ def __init__(self): def reset(self, seed=None, episode_id=None, **kwargs): self.counter = 0 self._state = State(episode_id=episode_id or "probe", step_count=0) - return ProbeObservation(counter=0, reward=0.0, done=False) + observation = ProbeObservation(counter=0, reward=0.0, done=False) + if self.mode in { + "nondeterministic", + "ignored_seed", + "judged_stable", + "judged_noisy", + }: + ProbeEnvironment._reset_ordinal += 1 + self.ordinal = ProbeEnvironment._reset_ordinal + if self.mode in {"nondeterministic", "ignored_seed"}: + observation.metadata["session_ordinal"] = self.ordinal + if self.mode in {"judged_stable", "judged_noisy"}: + score = 0.5 if self.mode == "judged_stable" else float(self.ordinal % 2) + self.rubric = ControlledJudge(score) + return observation def step(self, action, timeout_s=None, **kwargs): self.counter += action.increment @@ -59,6 +90,12 @@ def state(self): return self._state +class IgnoredSeedEnvironment(ProbeEnvironment): + # Deliberately exclude seed and **kwargs so framework filtering is observable. + def reset(self, episode_id=None): + return super().reset(episode_id=episode_id) + + class WireFault: """Corrupt real server responses after serialization, preserving raw defects.""" @@ -99,6 +136,13 @@ async def fault_send(message: dict[str, Any]): payload.pop("done", None) elif data.get("type") == "state" and self.mode == "bad_state": data["data"]["episode_id"] = "wrong-episode" + elif data.get("type") == "validation": + if self.mode == "missing_record": + data["data"].pop("trajectory", None) + elif self.mode == "trace_mismatch": + data["data"]["trajectory"]["records"][0]["response"]["data"][ + "observation" + ]["counter"] = 999 message = {**message, "text": json.dumps(data)} await send(message) @@ -117,11 +161,18 @@ def make_app(mode="good"): "missing_done", "bad_state", "hung_step", + "ignored_seed", + "nondeterministic", + "missing_record", + "trace_mismatch", + "judged_stable", + "judged_noisy", }: raise ValueError(f"Unknown fixture mode: {mode}") + environment = IgnoredSeedEnvironment if mode == "ignored_seed" else ProbeEnvironment return WireFault( create_app( - ProbeEnvironment, + lambda: environment(mode), ProbeAction, ProbeObservation, env_name="validation_probe", diff --git a/tests/test_validation/integration/test_runtime_cli.py b/tests/test_validation/integration/test_runtime_cli.py index daee42a28e..bae6f2f50a 100644 --- a/tests/test_validation/integration/test_runtime_cli.py +++ b/tests/test_validation/integration/test_runtime_cli.py @@ -13,6 +13,7 @@ import jsonschema import pytest +import yaml pytestmark = pytest.mark.docker @@ -22,12 +23,12 @@ "runtime.reward_well_formed", "runtime.observation_schema", "runtime.state_contract", + "runtime.seed_control", + "runtime.episode_determinism", + "runtime.trajectory_record", } PENDING = { - "runtime.trajectory_record", "runtime.tool_declaration_accuracy", - "runtime.seed_control", - "runtime.episode_determinism", "runtime.network_policy", "runtime.host_containment", "runtime.resource_bounds", @@ -285,6 +286,23 @@ def _invoke_cli( return result, report, checks, artifacts +def _assert_fresh_replays(artifacts, identical_samples=3): + replay = artifacts["replays.json"] + assert replay["failure_reason"] is None + samples = replay["samples"] + assert len(samples) == identical_samples # One extra different-seed control. + assert sum(row["scope"] != "seed" for row in samples) == identical_samples - 1 + assert sum(row["scope"] == "seed" for row in samples) == 1 + assert all(row["failure_reason"] is None for row in samples) + assert all(len(row["trace"]) == 6 for row in samples) + fresh = [row for row in samples if row["scope"] == "container"] + assert len(fresh) == 1 and fresh[0]["cleanup_complete"] is True + primary = artifacts["run-manifest.json"]["provider"] + assert fresh[0]["provider"]["container_id"] != primary["container_id"] + assert fresh[0]["provider"]["image_id"] == primary["image_id"] + assert fresh[0]["provider"]["container_id"] + + @pytest.mark.parametrize( "mode,failed_check", [ @@ -293,6 +311,10 @@ def _invoke_cli( ("bad_observation", "runtime.observation_schema"), ("missing_done", "runtime.observation_schema"), ("bad_state", "runtime.state_contract"), + ("ignored_seed", "runtime.seed_control"), + ("nondeterministic", "runtime.episode_determinism"), + ("missing_record", "runtime.trajectory_record"), + ("trace_mismatch", "runtime.trajectory_record"), ], ) def test_cli_runtime_contract_findings(cli_context, tmp_path, mode, failed_check): @@ -306,6 +328,7 @@ def test_cli_runtime_contract_findings(cli_context, tmp_path, mode, failed_check assert artifacts["cleanup.json"]["required"] is True assert artifacts["run-manifest.json"]["provider"]["container_id"] assert artifacts["run-manifest.json"]["source_digest"] == report["source_digest"] + _assert_fresh_replays(artifacts) trace = artifacts["collector-trace.json"] assert [row["operation"] for row in trace] == [ "reset", @@ -316,8 +339,9 @@ def test_cli_runtime_contract_findings(cli_context, tmp_path, mode, failed_check "state", ] if failed_check: - assert result.returncode == 1 - assert report["verdict"] == "fail" + warning_only = failed_check == "runtime.trajectory_record" + assert result.returncode == (0 if warning_only else 1) + assert report["verdict"] == ("warn" if warning_only else "fail") assert checks[failed_check]["status"] == "fail" assert checks[failed_check]["evidence"] else: @@ -361,7 +385,14 @@ def test_cli_hung_step_times_out_with_partial_evidence(cli_context, tmp_path): assert report["manifest"]["resources"]["episode_timeout_s"] == 3.0 assert checks["runtime.startup"]["status"] == "pass" assert all( - checks[key]["status"] == "fail" for key in IMPLEMENTED - {"runtime.startup"} + checks[key]["status"] == "fail" + for key in { + "runtime.reward_well_formed", + "runtime.observation_schema", + "runtime.state_contract", + "runtime.seed_control", + "runtime.trajectory_record", + } ) _assert_partial_episode(artifacts, "TimeoutError") @@ -414,3 +445,43 @@ def test_cli_skip_build_does_not_invoke_docker(cli_context, tmp_path): assert artifacts["cleanup.json"]["required"] is False assert artifacts["run-manifest.json"]["provider"] == {} assert artifacts["collector-trace.json"] == [] + + +@pytest.mark.parametrize( + "mode,expected", [("judged_stable", "pass"), ("judged_noisy", "fail")] +) +def test_cli_controlled_judge_requires_twenty_samples( + cli_context, tmp_path, mode, expected +): + with (cli_context / "Dockerfile").open("a") as stream: + stream.write(f"\nENV VALIDATION_FAULT={mode}\n") + manifest_path = cli_context / "openenv.yaml" + manifest = yaml.safe_load(manifest_path.read_text()) + validation = manifest["validation"] + validation["capabilities"]["llm_judged"] = True + validation["judge"] = { + "model": "controlled-test-judge", + "version": "1", + "params": {"mode": mode}, + } + validation["reward"]["variance_tolerance"] = 0.01 + manifest_path.unlink() + manifest_path.write_text(yaml.safe_dump(manifest, sort_keys=False)) + result, report, checks, artifacts = _invoke_cli(cli_context, tmp_path, mode) + assert checks["runtime.startup"]["status"] == "pass" + assert checks["runtime.seed_control"]["status"] == "pass" + assert checks["runtime.trajectory_record"]["status"] == "pass" + determinism = checks["runtime.episode_determinism"] + assert determinism["status"] == expected + assert determinism["measured"]["completed_replays"] == 20 + _assert_fresh_replays(artifacts, identical_samples=20) + assert determinism["measured"]["variance_units"] == "reward_squared" + variances = determinism["measured"]["reward_population_variance"] + assert len(variances) == 2 + if mode == "judged_stable": + assert variances == [0.0, 0.0] + assert result.returncode == 0 and report["verdict"] == "warn" + else: + assert all(value > 0.01 for value in variances) + assert result.returncode == 1 and report["verdict"] == "fail" + assert artifacts["cleanup.json"] == {"required": True, "completed": True} diff --git a/tests/test_validation/integration/test_runtime_process.py b/tests/test_validation/integration/test_runtime_process.py index bf6b68c31a..8afe549ea3 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"), @@ -163,6 +164,9 @@ def start(self, spec): ("missing_done", "runtime.observation_schema"), ("bad_state", "runtime.state_contract"), ("startup_failure", "runtime.startup"), + ("ignored_seed", "runtime.seed_control"), + ("missing_record", "runtime.trajectory_record"), + ("trace_mismatch", "runtime.trajectory_record"), ], ) def test_installed_server_collector_and_graders_over_loopback( @@ -194,6 +198,8 @@ def test_installed_server_collector_and_graders_over_loopback( "reward_well_formed", "observation_schema", "state_contract", + "seed_control", + "trajectory_record", ): assert checks[f"runtime.{check}"].status is CheckStatus.PASS if mode != "startup_failure": @@ -204,10 +210,24 @@ def test_installed_server_collector_and_graders_over_loopback( 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 telemetry["seed"]["accepted"] is (mode != "ignored_seed") + if mode != "missing_record": + assert len(telemetry["trajectory"]["records"]) == len(trace) + assert telemetry["trajectory"]["complete"] is True + else: + assert "trajectory" not in telemetry assert len(telemetry["attribution"]) == 2 - assert telemetry["trajectory"]["complete"] is True + assert checks["runtime.episode_determinism"].status is CheckStatus.SKIP + assert "fresh_container" in " ".join( + checks["runtime.episode_determinism"].evidence + ) + replay = json.loads((artifacts / "report/replays.json").read_text()) + assert ( + replay["failure_reason"] == "missing provider capability: fresh_container" + ) + assert [row["scope"] for row in replay["samples"]] == ["session", "seed"] + assert all(row["provider"] is None for row in replay["samples"]) + assert all(row["cleanup_complete"] is None for row in replay["samples"]) for line in (artifacts / "report/SHA256SUMS").read_text().splitlines(): checksum, name = line.split(" ", 1) assert ( diff --git a/tests/test_validation/test_runtime_artifacts.py b/tests/test_validation/test_runtime_artifacts.py index 490d28f6b0..ddcb124baa 100644 --- a/tests/test_validation/test_runtime_artifacts.py +++ b/tests/test_validation/test_runtime_artifacts.py @@ -13,7 +13,11 @@ from openenv.validation.manifest import NormalizedManifest from openenv.validation.report import ValidationReportV2 from openenv.validation.runtime.artifacts import write_runtime_bundle -from openenv.validation.runtime.contracts import RuntimeEvidence, WireExchange +from openenv.validation.runtime.contracts import ( + ReplayEvidence, + RuntimeEvidence, + WireExchange, +) from openenv.validation.types import Lane, Level, SignatureKind, Verdict from support.runtime import evidence, exchange @@ -236,3 +240,128 @@ def test_telemetry_redaction_or_omission_marks_bundle_modified(tmp_path, telemet assert "private-value" not in "".join( path.read_text() for path in tmp_path.iterdir() ) + + +def test_replay_artifact_retains_transcript_telemetry_identity_and_cleanup(tmp_path): + sample = replace( + measured(), telemetry_json='{"schema_version":1,"seed":{"value":42}}' + ) + original = replace( + measured(), + replays=( + ReplayEvidence("session", sample), + ReplayEvidence( + "container", + sample, + '{"container_id":"second","image_id":"sha256:abc"}', + True, + ), + ), + replay_failure_reason="judged replay sampling exceeded total budget", + ) + write_runtime_bundle(tmp_path, report(), evidence=original) + artifact = json.loads((tmp_path / "replays.json").read_text()) + assert artifact["failure_reason"] == original.replay_failure_reason + assert [row["scope"] for row in artifact["samples"]] == ["session", "container"] + for row in artifact["samples"]: + assert row["trace"] == [ + { + "operation": exchange.operation, + "request_json": json.loads(exchange.request_json), + "response_json": json.loads(exchange.response_json), + } + for exchange in sample.exchanges + ] + assert row["schema"] == json.loads(sample.observation_schema_json) + assert row["telemetry"] == json.loads(sample.telemetry_json) + assert row["redacted"] is False + container = artifact["samples"][1] + assert container["cleanup_complete"] is True + assert container["provider"] == {"container_id": "second", "image_id": "sha256:abc"} + sums = (tmp_path / "SHA256SUMS").read_text() + digest = hashlib.sha256((tmp_path / "replays.json").read_bytes()).hexdigest() + assert f"{digest} replays.json" in sums + + +def test_replay_artifact_redacts_each_independent_evidence_source(tmp_path): + sample = RuntimeEvidence( + exchanges=( + exchange( + "step", {"token": "private-request"}, {"value": "hf_notarealtoken12345"} + ), + ), + observation_schema_json='{"description":"Bearer private-schema"}', + telemetry_json='{"rubric":{"api_key":"private-telemetry"}}', + failure_reason="Authorization: Bearer private-failure", + ) + original = replace( + measured(), + replays=( + ReplayEvidence("container", sample, '{"secret":"private-provider"}', False), + ), + replay_failure_reason="Bearer private-schedule", + ) + write_runtime_bundle(tmp_path, report(), evidence=original) + text = (tmp_path / "replays.json").read_text() + assert "private-" not in text + assert "hf_notarealtoken" not in text + row = json.loads(text)["samples"][0] + assert row["redacted"] is True + assert row["cleanup_complete"] is False + assert row["provider"]["secret"] == "[REDACTED]" + + +def test_malformed_replay_omissions_are_explicit_and_do_not_leak_raw_data(tmp_path): + malformed = "not-json-private-value" + sample = RuntimeEvidence( + exchanges=(WireExchange("step", malformed, malformed),), + observation_schema_json=malformed, + telemetry_json=malformed, + ) + original = replace( + measured(), replays=(ReplayEvidence("container", sample, malformed, True),) + ) + write_runtime_bundle(tmp_path, report(), evidence=original) + text = (tmp_path / "replays.json").read_text() + assert malformed not in text + row = json.loads(text)["samples"][0] + assert row["redacted"] is True + assert row["omitted_trace_fields"] == [ + {"exchange_index": 0, "field": "request_json"}, + {"exchange_index": 0, "field": "response_json"}, + ] + assert row["omitted_evidence_fields"] == ["telemetry", "schema", "provider"] + + +def test_replay_artifact_distinguishes_absent_fields_from_json_null(tmp_path): + samples = tuple( + ReplayEvidence( + "session", + RuntimeEvidence(observation_schema_json=value, telemetry_json=value), + value, + ) + for value in (None, "null") + ) + write_runtime_bundle( + tmp_path, report(), evidence=replace(measured(), replays=samples) + ) + rows = json.loads((tmp_path / "replays.json").read_text())["samples"] + for key in ("schema", "telemetry", "provider"): + assert rows[0][key] is rows[1][key] is None + assert rows[0][f"{key}_available"] is False + assert rows[1][f"{key}_available"] is True + + +def test_rewriting_bundle_removes_stale_optional_evidence(tmp_path): + original = replace( + measured(), + telemetry_json="{}", + replays=(ReplayEvidence("session", measured()),), + ) + write_runtime_bundle(tmp_path, report(), evidence=original) + assert (tmp_path / "replays.json").exists() + assert (tmp_path / "session-telemetry.json").exists() + write_runtime_bundle(tmp_path, report(), evidence=measured()) + assert not (tmp_path / "replays.json").exists() + assert not (tmp_path / "session-telemetry.json").exists() + assert "replays.json" not in (tmp_path / "SHA256SUMS").read_text() diff --git a/tests/test_validation/test_runtime_execution.py b/tests/test_validation/test_runtime_execution.py index 29b0b0b7a5..f6f792b15e 100644 --- a/tests/test_validation/test_runtime_execution.py +++ b/tests/test_validation/test_runtime_execution.py @@ -12,7 +12,9 @@ from openenv.validation.report import CheckResult from openenv.validation.runner import run_validation, source_digest from openenv.validation.runtime.artifacts import write_runtime_bundle -from openenv.validation.runtime.contracts import RuntimeEvidence +from openenv.validation.runtime.collector import RuntimeCollectionInterrupted +from openenv.validation.runtime.contracts import ReplayEvidence, RuntimeEvidence +from openenv.validation.runtime.replay import REPLAY_BUDGET_SECONDS from openenv.validation.runtime.scheduler import execute_graders, order_graders from openenv.validation.types import CheckStatus, Level, ProviderCapability from support.runtime import evidence, exchange, FakeRuntimeProvider @@ -28,14 +30,28 @@ def package(tmp_path): return root -def measured_episode(): +@pytest.fixture +def baseline_only(monkeypatch): + """Isolate original-subject failure handling from separately tested replays.""" + monkeypatch.setattr( + "openenv.validation.runner.collect_replays", + lambda provider, running, spec, plan, evidence, **kwargs: evidence, + ) + + +def measured_episode(seed=42): rows = [] for step in (0, 1): operation = "reset" if step == 0 else "step" rows.append( exchange( operation, - {"data": {"episode_id": "validation-probe"}}, + { + "type": operation, + "data": {"episode_id": "validation-probe", "seed": seed} + if operation == "reset" + else {"increment": 1}, + }, { "type": "observation", "data": { @@ -56,7 +72,7 @@ def measured_episode(): }, ) ) - return evidence( + result = evidence( *rows, observation_schema={ "type": "object", @@ -68,9 +84,31 @@ def measured_episode(): }, }, ) + return replace( + result, + telemetry_json=json.dumps( + { + "schema_version": 1, + "seed": {"requested": True, "accepted": True, "value": seed}, + "trajectory": { + "schema_version": 1, + "source": "openenv-server", + "complete": True, + "records": [ + { + "operation": row.operation, + "request": json.loads(row.request_json), + "response": json.loads(row.response_json), + } + for row in rows + ], + }, + } + ), + ) -def test_runtime_collects_once_cleans_up_and_marks_remaining_work( +def test_runtime_collects_primary_and_session_replays_then_cleans_up( package, monkeypatch, tmp_path ): provider = FakeRuntimeProvider() @@ -78,15 +116,19 @@ def test_runtime_collects_once_cleans_up_and_marks_remaining_work( def collect(*args, **kwargs): calls.append((args, kwargs)) - return measured_episode() + return measured_episode(seed=args[1].reset.seed) monkeypatch.setattr("openenv.validation.runner.collect_runtime_evidence", collect) + monkeypatch.setattr( + "openenv.validation.runtime.replay.collect_runtime_evidence", collect + ) bundle = tmp_path / "bundle" report = run_validation( package, max_level=Level.RUNTIME, provider=provider, artifacts_dir=bundle ) results = {r.check_id: r.status for r in report.results} - assert len(calls) == 1 + assert len(calls) == 3 + assert [args[1].reset.seed for args, _ in calls] == [42, 42, 43] assert len(provider.builds) == len(provider.launches) == 1 assert provider.subject.stopped assert report.report_schema_version == "2" @@ -97,14 +139,25 @@ def collect(*args, **kwargs): "reward_well_formed", "observation_schema", "state_contract", + "seed_control", + "trajectory_record", ): assert results[f"runtime.{name}"] is CheckStatus.PASS assert results["runtime.network_policy"] is CheckStatus.SKIP + assert results["runtime.episode_determinism"] is CheckStatus.SKIP + assert any( + "fresh_container" in reason + for result in report.results + if result.check_id == "runtime.episode_determinism" + for reason in result.evidence + ) assert json.loads((bundle / "cleanup.json").read_text())["completed"] is True assert json.loads((bundle / "runtime-plan.json").read_text())["reset"]["seed"] == 42 -def test_runtime_collection_uses_declared_episode_timeout(package, monkeypatch): +def test_runtime_collection_caps_declared_episode_timeout_at_total_budget( + package, monkeypatch, baseline_only +): path = package / "openenv.yaml" path.write_text( path.read_text().replace("episode_timeout_s: 30.0", "episode_timeout_s: 600.0") @@ -118,7 +171,7 @@ def collect(*args, **kwargs): monkeypatch.setattr("openenv.validation.runner.collect_runtime_evidence", collect) run_validation(package, max_level=Level.RUNTIME, provider=FakeRuntimeProvider()) - assert calls[0]["episode_timeout_s"] == 600.0 + assert calls[0]["episode_timeout_s"] == REPLAY_BUDGET_SECONDS def test_skip_build_has_no_provider_side_effects(package): @@ -210,7 +263,7 @@ def failed_build(*args): @pytest.mark.parametrize("mutation", ["content", "symlink", "unreadable"]) def test_source_change_withdraws_dependent_runtime_results( - package, monkeypatch, tmp_path, mutation + package, monkeypatch, tmp_path, mutation, baseline_only ): provider = FakeRuntimeProvider() original_digest = source_digest(package) @@ -300,7 +353,7 @@ def failed_stop(): @pytest.mark.parametrize("collection_state", ["complete", "failed", "partial"]) @pytest.mark.parametrize("teardown_error", [RuntimeError, KeyboardInterrupt]) def test_teardown_failure_preserves_collection_outcome( - package, monkeypatch, tmp_path, collection_state, teardown_error + package, monkeypatch, tmp_path, collection_state, teardown_error, baseline_only ): provider = FakeRuntimeProvider() collected = measured_episode() @@ -356,7 +409,9 @@ def failed_stop(): assert "must-not-appear" not in report.model_dump_json() -def test_bad_static_bounds_do_not_suppress_independent_runtime(package, monkeypatch): +def test_bad_static_bounds_do_not_suppress_independent_runtime( + package, monkeypatch, baseline_only +): path = package / "openenv.yaml" path.write_text(path.read_text().replace("floor_margin: 0.5", "floor_margin: 0.01")) provider = FakeRuntimeProvider() @@ -372,6 +427,49 @@ def test_bad_static_bounds_do_not_suppress_independent_runtime(package, monkeypa ) +@pytest.mark.parametrize("interrupted", [False, True]) +def test_bundle_cleanup_includes_failed_replay_teardown( + package, monkeypatch, tmp_path, interrupted +): + provider = FakeRuntimeProvider() + combined = replace( + measured_episode(), + replays=( + ReplayEvidence( + "container", + RuntimeEvidence(failure_reason="fresh container teardown unconfirmed"), + cleanup_complete=False, + ), + ), + ) + + def replays(*args, **kwargs): + if interrupted: + raise RuntimeCollectionInterrupted(combined) + return combined + + monkeypatch.setattr( + "openenv.validation.runner.collect_runtime_evidence", + lambda *a, **k: measured_episode(), + ) + monkeypatch.setattr("openenv.validation.runner.collect_replays", replays) + bundle = tmp_path / "bundle" + result = run_validation( + package, max_level=Level.RUNTIME, provider=provider, artifacts_dir=bundle + ) + assert provider.subject.stopped + assert ( + next(row for row in result.results if row.check_id == "runtime.startup").status + is CheckStatus.ERROR + ) + assert json.loads((bundle / "cleanup.json").read_text()) == { + "required": True, + "completed": False, + } + replay = json.loads((bundle / "replays.json").read_text())["samples"][0] + assert replay["cleanup_complete"] is False + + def test_semantic_ceiling_does_not_claim_semantic_execution(package): report = run_validation(package, max_level=Level.SEMANTIC, skip_build=True) assert report.levels_run == [Level.STATIC] diff --git a/tests/test_validation/test_runtime_repeatability.py b/tests/test_validation/test_runtime_repeatability.py new file mode 100644 index 0000000000..c8eae8fe76 --- /dev/null +++ b/tests/test_validation/test_runtime_repeatability.py @@ -0,0 +1,288 @@ +import json +from dataclasses import replace + +import pytest +from openenv.validation.graders.runtime.repeatability import ( + EpisodeDeterminismGrader, + SeedControlGrader, + TrajectoryRecordGrader, +) +from openenv.validation.manifest import JudgePin +from openenv.validation.runtime.contracts import ReplayEvidence +from openenv.validation.types import CheckStatus +from test_runtime_grading import good_rows, mutate_response, subject_with + +GRADERS = [SeedControlGrader, EpisodeDeterminismGrader, TrajectoryRecordGrader] + + +def sample(subject, *, seed=42, reward=1.0): + rows = good_rows() + request = json.loads(rows[0].request_json) + request["data"]["seed"] = seed + rows[0] = replace(rows[0], request_json=json.dumps(request)) + mutate_response(rows, 2, lambda response: response["data"].update(reward=reward)) + telemetry = { + "schema_version": 1, + "seed": {"requested": True, "accepted": True, "value": seed}, + "trajectory": { + "schema_version": 1, + "source": "openenv-server", + "complete": True, + "reason": None, + "records": [ + { + "operation": row.operation, + "request": json.loads(row.request_json), + "response": json.loads(row.response_json), + } + for row in rows + ], + }, + } + return replace( + subject.runtime_evidence, + exchanges=tuple(rows), + telemetry_json=json.dumps(telemetry), + replays=(), + ) + + +def subject_with_replays(tmp_path, *, judged=False): + subject = subject_with(tmp_path) + count = 20 if judged else 3 + replays = tuple( + ReplayEvidence("container" if index == 1 else "session", sample(subject)) + for index in range(1, count) + ) + (ReplayEvidence("seed", sample(subject, seed=43)),) + if judged: + subject.manifest.capabilities.llm_judged = True + subject.manifest.judge = JudgePin(model="test-judge", version="1") + subject.manifest.reward.variance_tolerance = 0.2 + return replace(subject, runtime_evidence=replace(sample(subject), replays=replays)) + + +def alter_telemetry(evidence, mutate): + value = json.loads(evidence.telemetry_json) + mutate(value) + return replace(evidence, telemetry_json=json.dumps(value)) + + +@pytest.mark.parametrize("grader", GRADERS) +def test_seed_invariant_fresh_replays_and_independent_record_pass(tmp_path, grader): + assert grader().run(subject_with_replays(tmp_path)).status is CheckStatus.PASS + + +@pytest.mark.parametrize("grader", GRADERS) +def test_missing_runtime_evidence_is_incomplete(tmp_path, grader): + subject = replace(subject_with_replays(tmp_path), runtime_evidence=None) + assert grader().run(subject).status is CheckStatus.SKIP + + +@pytest.mark.parametrize("grader", [SeedControlGrader, TrajectoryRecordGrader]) +def test_unavailable_telemetry_differs_from_failed_telemetry(tmp_path, grader): + subject = subject_with_replays(tmp_path) + evidence = replace(subject.runtime_evidence, telemetry_json=None) + assert ( + grader().run(replace(subject, runtime_evidence=evidence)).status + is CheckStatus.SKIP + ) + evidence = replace(evidence, telemetry_error="telemetry failed (PermissionError)") + result = grader().run(replace(subject, runtime_evidence=evidence)) + assert result.status is CheckStatus.FAIL + assert "PermissionError" in result.evidence[0] + + +@pytest.mark.parametrize( + "field,value", + [("accepted", False), ("value", 99), ("requested", False), ("value", True)], +) +def test_rejected_dropped_or_wrong_seed_fails(tmp_path, field, value): + subject = subject_with_replays(tmp_path) + evidence = alter_telemetry( + subject.runtime_evidence, lambda data: data["seed"].update({field: value}) + ) + result = SeedControlGrader().run(replace(subject, runtime_evidence=evidence)) + assert result.status is CheckStatus.FAIL + assert "seed was not observed as forwarded" in result.evidence[0] + + +def test_different_seed_schedule_is_required_but_output_need_not_differ(tmp_path): + subject = subject_with_replays(tmp_path) + evidence = replace( + subject.runtime_evidence, replays=subject.runtime_evidence.replays[:-1] + ) + assert ( + SeedControlGrader().run(replace(subject, runtime_evidence=evidence)).status + is CheckStatus.SKIP + ) + evidence = replace( + evidence, replays=evidence.replays + (ReplayEvidence("seed", sample(subject)),) + ) + assert ( + SeedControlGrader().run(replace(subject, runtime_evidence=evidence)).status + is CheckStatus.FAIL + ) + + +@pytest.mark.parametrize( + "field,value", [("trajectory", None), ("schema_version", True)] +) +def test_missing_or_malformed_record_fails(tmp_path, field, value): + subject = subject_with_replays(tmp_path) + evidence = alter_telemetry( + subject.runtime_evidence, lambda data: data.update({field: value}) + ) + assert ( + TrajectoryRecordGrader().run(replace(subject, runtime_evidence=evidence)).status + is CheckStatus.FAIL + ) + + +@pytest.mark.parametrize("mutation", ["truncated", "changed", "incomplete"]) +def test_subject_record_cannot_pass_by_using_validator_transcript_alone( + tmp_path, mutation +): + subject = subject_with_replays(tmp_path) + + def mutate(data): + record = data["trajectory"] + if mutation == "truncated": + record["records"].pop() + elif mutation == "changed": + record["records"][2]["response"]["data"]["reward"] = 0.0 + else: + record["complete"] = False + + evidence = alter_telemetry(subject.runtime_evidence, mutate) + result = TrajectoryRecordGrader().run(replace(subject, runtime_evidence=evidence)) + assert result.status is CheckStatus.FAIL + + +@pytest.mark.parametrize( + "field,value", + [("reward", 0.0), ("done", True), ("observation", {"counter": "private-value"})], +) +def test_divergence_reports_first_path_without_values(tmp_path, field, value): + subject = subject_with_replays(tmp_path) + evidence = subject.runtime_evidence + replay = evidence.replays[0] + rows = mutate_response( + list(replay.evidence.exchanges), + 2, + lambda data: data["data"].update({field: value}), + ) + replay = replace(replay, evidence=replace(replay.evidence, exchanges=tuple(rows))) + evidence = replace(evidence, replays=(replay,) + evidence.replays[1:]) + result = EpisodeDeterminismGrader().run(replace(subject, runtime_evidence=evidence)) + assert result.status is CheckStatus.FAIL + assert f"$[2].response.data.{field}" in result.evidence[0] + assert "private-value" not in result.model_dump_json() + + +def test_fresh_sessions_without_fresh_container_are_incomplete(tmp_path): + subject = subject_with_replays(tmp_path) + evidence = replace( + subject.runtime_evidence, + replays=tuple( + replace(row, scope="session") for row in subject.runtime_evidence.replays + ), + ) + result = EpisodeDeterminismGrader().run(replace(subject, runtime_evidence=evidence)) + assert result.status is CheckStatus.SKIP + + +def test_partial_judged_sample_is_incomplete(tmp_path): + subject = subject_with_replays(tmp_path, judged=True) + evidence = replace( + subject.runtime_evidence, replays=subject.runtime_evidence.replays[:18] + ) + result = EpisodeDeterminismGrader().run(replace(subject, runtime_evidence=evidence)) + assert result.status is CheckStatus.SKIP + assert result.measured["completed_replays"] == 19 + + +@pytest.mark.parametrize( + "low,bound,status", [(0.5, 0.1, CheckStatus.PASS), (0.0, 0.2, CheckStatus.FAIL)] +) +def test_judged_population_variance_uses_reward_squared_units( + tmp_path, low, bound, status +): + subject = subject_with_replays(tmp_path, judged=True) + subject.manifest.reward.variance_tolerance = bound + evidence = subject.runtime_evidence + replays = tuple( + replace(row, evidence=sample(subject, reward=low if index < 10 else 1.0)) + for index, row in enumerate(evidence.replays[:-1]) + ) + (evidence.replays[-1],) + result = EpisodeDeterminismGrader().run( + replace(subject, runtime_evidence=replace(evidence, replays=replays)) + ) + assert result.status is status + assert result.measured["variance_units"] == "reward_squared" + assert result.measured["reward_population_variance"] == [ + pytest.approx((1.0 - low) ** 2 / 4) + ] + + +def test_judged_mode_still_checks_observations_and_state(tmp_path): + subject = subject_with_replays(tmp_path, judged=True) + evidence = subject.runtime_evidence + replay = evidence.replays[0] + rows = mutate_response( + list(replay.evidence.exchanges), + 3, + lambda data: data["data"].update(step_count=2), + ) + replay = replace(replay, evidence=replace(replay.evidence, exchanges=tuple(rows))) + evidence = replace(evidence, replays=(replay,) + evidence.replays[1:]) + result = EpisodeDeterminismGrader().run(replace(subject, runtime_evidence=evidence)) + assert result.status is CheckStatus.FAIL + assert "$[3].response.data.step_count" in result.evidence[0] + + +def test_judged_variance_does_not_pool_different_steps(tmp_path): + subject = subject_with_replays(tmp_path, judged=True) + subject.manifest.reward.variance_tolerance = 0.0 + + def two_steps(evidence): + rows = list(evidence.exchanges) + second = mutate_response( + [rows[2]], 0, lambda data: data["data"].update(reward=0.0) + )[0] + state = mutate_response( + [rows[3]], 0, lambda data: data["data"].update(step_count=2) + )[0] + return replace(evidence, exchanges=tuple(rows + [second, state])) + + evidence = two_steps(subject.runtime_evidence) + evidence = replace( + evidence, + replays=tuple( + replace(row, evidence=two_steps(row.evidence)) for row in evidence.replays + ), + ) + result = EpisodeDeterminismGrader().run(replace(subject, runtime_evidence=evidence)) + assert result.status is CheckStatus.PASS + assert result.measured["reward_population_variance"] == [0.0, 0.0] + + +@pytest.mark.parametrize("grader", [SeedControlGrader, TrajectoryRecordGrader]) +def test_malformed_telemetry_is_a_finding(tmp_path, grader): + subject = subject_with_replays(tmp_path) + evidence = replace(subject.runtime_evidence, telemetry_json="{not-json") + assert ( + grader().run(replace(subject, runtime_evidence=evidence)).status + is CheckStatus.FAIL + ) + + +@pytest.mark.parametrize("grader", GRADERS) +def test_truncated_primary_evidence_never_passes(tmp_path, grader): + subject = subject_with_replays(tmp_path) + evidence = replace( + subject.runtime_evidence, failure_reason="step failed (TimeoutError)" + ) + assert ( + grader().run(replace(subject, runtime_evidence=evidence)).status + is CheckStatus.FAIL + ) diff --git a/tests/test_validation/test_runtime_replay.py b/tests/test_validation/test_runtime_replay.py new file mode 100644 index 0000000000..47300b4d2a --- /dev/null +++ b/tests/test_validation/test_runtime_replay.py @@ -0,0 +1,307 @@ +import json +from dataclasses import replace +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest +from openenv.validation.runtime import replay +from openenv.validation.runtime.collector import RuntimeCollectionInterrupted +from openenv.validation.runtime.contracts import ( + LaunchSpec, + RuntimeEvidence, + RuntimePlan, + WireExchange, +) +from openenv.validation.types import ProviderCapability + + +def inputs(monkeypatch, *, judged=False, fresh_container=True): + image = "sha256:" + "a" * 64 + original = SimpleNamespace( + base_url="http://original", + inspect=Mock(return_value={"container_id": "original", "image_id": image}), + stop=Mock(), + ) + fresh = SimpleNamespace( + base_url="http://fresh", + inspect=Mock(return_value={"container_id": "fresh", "image_id": image}), + stop=Mock(), + ) + provider = SimpleNamespace( + capabilities={ProviderCapability.FRESH_CONTAINER} if fresh_container else set(), + start=Mock(return_value=fresh), + ) + spec = LaunchSpec( + image_ref=image, + resources={"cpu": 1, "memory_mb": 256, "disk_mb": 64, "episode_timeout_s": 30}, + network={"mode": "public"}, + run_id="validation-original", + env_vars={"OPENENV_VALIDATION_TOKEN": "original-run-authorization"}, + ) + plan = RuntimePlan.model_validate( + { + "plan_schema_version": "1", + "reset": {"episode_id": "same-episode", "seed": 42}, + "actions": [{"increment": 1}], + } + ) + collect = Mock(return_value=RuntimeEvidence()) + monkeypatch.setattr(replay, "collect_runtime_evidence", collect) + return ( + provider, + original, + fresh, + spec, + plan, + RuntimeEvidence(), + SimpleNamespace(llm_judged=judged), + collect, + ) + + +def run(values, **kwargs): + provider, original, _, spec, plan, evidence, capabilities, _ = values + return replay.collect_replays( + provider, original, spec, plan, evidence, capabilities=capabilities, **kwargs + ) + + +def test_schedule_preserves_inputs_verifies_identity_and_cleans_fresh_container( + monkeypatch, +): + values = inputs(monkeypatch) + provider, original, fresh, spec, plan, evidence, _, collect = values + result = run(values) + assert [row.scope for row in result.replays] == ["session", "seed", "container"] + assert result.exchanges == evidence.exchanges + assert result.replay_failure_reason is None + assert [call.args[0] for call in collect.call_args_list] == [ + "http://original", + "http://original", + "http://fresh", + ] + assert [call.args[1].reset.seed for call in collect.call_args_list] == [42, 43, 42] + assert all( + call.args[1].reset.episode_id == plan.reset.episode_id + for call in collect.call_args_list + ) + assert all(call.args[1].actions == plan.actions for call in collect.call_args_list) + new_spec = provider.start.call_args.args[0] + assert new_spec.image_ref == spec.image_ref + assert new_spec.run_id != spec.run_id + assert ( + new_spec.env_vars["OPENENV_VALIDATION_TOKEN"] + != spec.env_vars["OPENENV_VALIDATION_TOKEN"] + ) + assert ( + collect.call_args_list[-1].kwargs["validation_token"] + == new_spec.env_vars["OPENENV_VALIDATION_TOKEN"] + ) + assert json.loads(result.replays[-1].provider_json)["container_id"] == "fresh" + assert result.replays[-1].cleanup_complete is True + fresh.stop.assert_called_once() + original.stop.assert_not_called() + + +def test_judged_schedule_has_exactly_twenty_identical_inputs_plus_changed_seed( + monkeypatch, +): + values = inputs(monkeypatch, judged=True) + result = run(values) + assert 1 + sum(row.scope != "seed" for row in result.replays) == 20 + assert sum(row.scope == "seed" for row in result.replays) == 1 + assert values[-1].call_count == 20 + assert ( + len({call.args[1].model_dump_json() for call in values[-1].call_args_list}) == 2 + ) + + +def test_provider_without_fresh_containers_collects_only_session_experiments( + monkeypatch, +): + values = inputs(monkeypatch, fresh_container=False, judged=True) + result = run(values) + assert [row.scope for row in result.replays] == ["session", "seed"] + assert ( + result.replay_failure_reason == "missing provider capability: fresh_container" + ) + values[0].start.assert_not_called() + + +@pytest.mark.parametrize( + "inspection", + [ + {"container_id": "original"}, + {"container_id": "fresh", "image_id": "wrong-image"}, + {"image_id": "missing-container-id"}, + ], +) +def test_identity_mismatch_cannot_supply_a_completed_fresh_container_sample( + monkeypatch, inspection +): + values = inputs(monkeypatch) + fresh = values[2] + fresh.inspect.return_value = {"image_id": values[3].image_ref, **inspection} + result = run(values) + assert result.replays[-1].evidence.failure_reason + assert result.replays[-1].cleanup_complete is True + assert "inspection failed" in result.replay_failure_reason + assert values[-1].call_count == 2 + fresh.stop.assert_called_once() + + +def test_container_cleanup_failure_is_preserved_even_after_good_replay(monkeypatch): + values = inputs(monkeypatch) + values[2].stop.side_effect = RuntimeError("private-provider-message") + result = run(values) + assert result.replays[-1].cleanup_complete is False + assert result.replay_failure_reason == "fresh container cleanup failed" + assert "private-provider-message" not in result.replay_failure_reason + + +@pytest.mark.parametrize("failure", [RuntimeError, KeyboardInterrupt]) +def test_failed_start_retains_unconfirmed_cleanup_without_a_returned_handle( + monkeypatch, failure +): + values = inputs(monkeypatch) + values[0].start.side_effect = failure("private-provider-message") + if failure is KeyboardInterrupt: + with pytest.raises(RuntimeCollectionInterrupted) as error: + run(values) + result = error.value.evidence + else: + result = run(values) + sample = result.replays[-1] + assert sample.scope == "container" + assert sample.cleanup_complete is False + assert sample.provider_json is None + assert not sample.evidence.exchanges + assert "teardown could not be confirmed" in sample.evidence.failure_reason + assert "private-provider-message" not in result.replay_failure_reason + values[2].stop.assert_not_called() + + +def test_total_deadline_caps_episode_and_startup_budgets(monkeypatch): + values = inputs(monkeypatch) + monkeypatch.setattr(replay.time, "monotonic", lambda: 100.0) + result = run(values, deadline=102.5) + assert result.replay_failure_reason is None + assert all( + call.kwargs["episode_timeout_s"] == 2.5 for call in values[-1].call_args_list + ) + assert values[0].start.call_args.args[0].startup_timeout_s == 2.5 + + +def test_expired_total_budget_retains_baseline_without_starting_experiments( + monkeypatch, +): + values = inputs(monkeypatch) + monkeypatch.setattr(replay.time, "monotonic", lambda: 100.0) + result = run(values, deadline=99.0) + assert result.exchanges == values[5].exchanges + assert not result.replays + assert "TimeoutError" in result.replay_failure_reason + values[-1].assert_not_called() + values[0].start.assert_not_called() + + +def test_collection_failure_retains_prefix_and_stops_replaying(monkeypatch): + values = inputs(monkeypatch) + values[-1].side_effect = [ + RuntimeEvidence(), + RuntimeEvidence(failure_reason="step failed (TimeoutError)"), + ] + result = run(values) + assert len(result.replays) == 2 + assert result.replays[-1].evidence.failure_reason == "step failed (TimeoutError)" + values[0].start.assert_not_called() + + +@pytest.mark.parametrize("interrupted_call", [1, 3]) +def test_interruption_keeps_partial_replay_and_cleans_fresh_container( + monkeypatch, interrupted_call +): + values = inputs(monkeypatch) + partial = RuntimeEvidence( + failure_phase="step", failure_reason="step failed (KeyboardInterrupt)" + ) + values[-1].side_effect = [RuntimeEvidence()] * (interrupted_call - 1) + [ + RuntimeCollectionInterrupted(partial) + ] + with pytest.raises(RuntimeCollectionInterrupted) as error: + run(values) + result = error.value.evidence + assert len(result.replays) == interrupted_call + assert result.replays[-1].evidence is partial + assert "interrupted" in result.replay_failure_reason + if interrupted_call == 3: + values[2].stop.assert_called_once() + assert result.replays[-1].cleanup_complete is True + + +def test_failed_baseline_is_returned_without_further_execution(monkeypatch): + values = list(inputs(monkeypatch)) + values[5] = replace(values[5], failure_reason="reset failed (ValueError)") + assert run(values) is values[5] + values[-1].assert_not_called() + + +@pytest.mark.parametrize("budget,retained_samples", [(12, 1), (22, 3), (32, 3)]) +def test_retained_byte_budget_counts_primary_and_all_raw_fields( + monkeypatch, budget, retained_samples +): + values = list(inputs(monkeypatch, judged=True)) + values[5] = RuntimeEvidence(observation_schema_json="é") # Two UTF-8 bytes. + sample = RuntimeEvidence( + exchanges=(WireExchange("step", "aa", "bbb"),), + observation_schema_json="é", + telemetry_json="xyz", + ) + assert replay._evidence_bytes(sample) == 10 + values[-1].return_value = sample + monkeypatch.setattr(replay, "MAX_REPLAY_BYTES", budget) + result = run(values) + assert len(result.replays) == retained_samples + assert "total retained replay evidence exceeds" in result.replay_failure_reason + retained = replay._evidence_bytes(result) + sum( + replay._evidence_bytes(row.evidence) for row in result.replays + ) + assert retained <= budget + assert result.observation_schema_json == "é" + if budget >= 22: + container = next(row for row in result.replays if row.scope == "container") + assert container.cleanup_complete is True + assert json.loads(container.provider_json)["container_id"] == "fresh" + values[2].stop.assert_called_once() + if budget == 22: + assert not container.evidence.exchanges + + +def test_oversized_primary_evidence_is_explicitly_incomplete_and_not_retained( + monkeypatch, +): + values = list(inputs(monkeypatch)) + values[5] = RuntimeEvidence(telemetry_json="x" * 20) + monkeypatch.setattr(replay, "MAX_REPLAY_BYTES", 10) + result = run(values) + assert result.failure_phase == "replay evidence budget" + assert result.failure_reason == result.replay_failure_reason + assert replay._evidence_bytes(result) == 0 + values[-1].assert_not_called() + + +def test_byte_budget_does_not_swallow_interruption_or_discard_completed_prefix( + monkeypatch, +): + values = inputs(monkeypatch) + sample = RuntimeEvidence(telemetry_json="x" * 10) + values[-1].side_effect = [sample, RuntimeCollectionInterrupted(sample)] + monkeypatch.setattr(replay, "MAX_REPLAY_BYTES", 15) + with pytest.raises(RuntimeCollectionInterrupted) as error: + run(values) + assert len(error.value.evidence.replays) == 1 + assert error.value.evidence.replays[0].evidence is sample + assert ( + "total retained replay evidence exceeds" + in error.value.evidence.replay_failure_reason + ) diff --git a/tests/validation_runtime/README.md b/tests/validation_runtime/README.md index 4b27945737..d81b4ea378 100644 --- a/tests/validation_runtime/README.md +++ b/tests/validation_runtime/README.md @@ -38,8 +38,16 @@ The image supports controlled `VALIDATION_FAULT` modes: `good`, `bad_reward`, switches and wire corruption remain inside test assets. They share one fixture and one public runtime plan, so a defect changes one property at a time. -The Docker suite contains 13 required cases: three provider lifecycle tests, -nine CLI fault/control cases, and one real `echo_env` canary. The hung-step case +Repeatability checks compare the original episode against a fresh session and an +independently inspected fresh container, then exercise a different seed. Controlled +judge fixtures use exactly 20 identical-input samples and per-step population +variance; they make no inference calls. `replays.json` retains each trace, subject +telemetry, container identity and cleanup outcome. A process-only provider explicitly +skips fresh-container determinism. Subject-emitted records are compared with the +independently collected wire trace. + +The Docker suite contains 19 required cases: three provider lifecycle tests, +15 CLI fault/control cases, and one real `echo_env` canary. The hung-step case checks the episode deadline; the interruption case sends SIGINT only after a container log confirms the second step has begun. Both must retain the completed reset/state/step/state prefix and remove their own containers. Each CLI case uses diff --git a/tests/validation_runtime/acceptance.json b/tests/validation_runtime/acceptance.json index 75ad404610..2a2eb4c531 100644 --- a/tests/validation_runtime/acceptance.json +++ b/tests/validation_runtime/acceptance.json @@ -25,7 +25,10 @@ "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]" + "tests.test_validation.integration.test_runtime_process::test_installed_server_collector_and_graders_over_loopback[startup_failure-runtime.startup]", + "tests.test_validation.integration.test_runtime_process::test_installed_server_collector_and_graders_over_loopback[ignored_seed-runtime.seed_control]", + "tests.test_validation.integration.test_runtime_process::test_installed_server_collector_and_graders_over_loopback[missing_record-runtime.trajectory_record]", + "tests.test_validation.integration.test_runtime_process::test_installed_server_collector_and_graders_over_loopback[trace_mismatch-runtime.trajectory_record]" ], "docker": [ "tests.test_validation.integration.test_docker_lifecycle::test_docker_lifecycle_effective_limits_and_owned_cleanup", @@ -40,7 +43,13 @@ "tests.test_validation.integration.test_runtime_cli::test_cli_hung_step_times_out_with_partial_evidence", "tests.test_validation.integration.test_runtime_cli::test_cli_sigint_retains_partial_evidence_and_cleans_up", "tests.test_validation.integration.test_runtime_cli::test_cli_startup_failure_is_a_finding_and_leaves_no_container", - "tests.test_validation.integration.test_runtime_cli::test_cli_skip_build_does_not_invoke_docker" + "tests.test_validation.integration.test_runtime_cli::test_cli_skip_build_does_not_invoke_docker", + "tests.test_validation.integration.test_runtime_cli::test_cli_runtime_contract_findings[ignored_seed-runtime.seed_control]", + "tests.test_validation.integration.test_runtime_cli::test_cli_runtime_contract_findings[nondeterministic-runtime.episode_determinism]", + "tests.test_validation.integration.test_runtime_cli::test_cli_runtime_contract_findings[missing_record-runtime.trajectory_record]", + "tests.test_validation.integration.test_runtime_cli::test_cli_runtime_contract_findings[trace_mismatch-runtime.trajectory_record]", + "tests.test_validation.integration.test_runtime_cli::test_cli_controlled_judge_requires_twenty_samples[judged_stable-pass]", + "tests.test_validation.integration.test_runtime_cli::test_cli_controlled_judge_requires_twenty_samples[judged_noisy-fail]" ] } } From 0d461e521f99b6e335a4c32f82c9eef29c77598e Mon Sep 17 00:00:00 2001 From: burtenshaw Date: Thu, 24 Sep 2026 12:25:07 +0200 Subject: [PATCH 02/11] style: format provider capabilities --- src/openenv/validation/providers/docker.py | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/src/openenv/validation/providers/docker.py b/src/openenv/validation/providers/docker.py index 8c103928fc..6fa4fd3a13 100644 --- a/src/openenv/validation/providers/docker.py +++ b/src/openenv/validation/providers/docker.py @@ -209,10 +209,13 @@ class DockerValidationProvider: """Build isolated source snapshots and launch owned, resource-bounded subjects.""" name = "docker-local" - capabilities = frozenset({ - ProviderCapability.IMAGE_BUILD, ProviderCapability.EXEC, - ProviderCapability.FRESH_CONTAINER, - }) + capabilities = frozenset( + { + ProviderCapability.IMAGE_BUILD, + ProviderCapability.EXEC, + ProviderCapability.FRESH_CONTAINER, + } + ) supported_network_modes = frozenset({"public"}) def __init__(self, *, build_timeout_s: float = 600, max_context_bytes: int = 2**30): From ddb3ddb8e5a760e2ce3e0e128da5962fd220ce97 Mon Sep 17 00:00:00 2001 From: burtenshaw Date: Thu, 24 Sep 2026 12:36:18 +0200 Subject: [PATCH 03/11] fix: distinguish incomplete replay evidence --- .../graders/runtime/repeatability.py | 80 ++++++++---------- .../test_runtime_repeatability.py | 82 ++++++++++++++++++- 2 files changed, 116 insertions(+), 46 deletions(-) diff --git a/src/openenv/validation/graders/runtime/repeatability.py b/src/openenv/validation/graders/runtime/repeatability.py index 2812d28f71..8774b4cf67 100644 --- a/src/openenv/validation/graders/runtime/repeatability.py +++ b/src/openenv/validation/graders/runtime/repeatability.py @@ -111,7 +111,8 @@ def check(self, subject, evidence): problems = [] original_seed = None for index, sample in enumerate( - [evidence] + [replay.evidence for replay in evidence.replays] + [evidence] + + [replay.evidence for replay in evidence.replays if replay.scope == "seed"] ): if sample.failure_reason or sample.telemetry_error: problems.append(f"replay {index}: reset or telemetry collection failed") @@ -129,7 +130,7 @@ def check(self, subject, evidence): continue if index == 0: original_seed = seed - elif evidence.replays[index - 1].scope == "seed" and seed == original_seed: + elif seed == original_seed: problems.append(f"replay {index}: scheduled seed was not changed") observed = _telemetry(sample).get("seed") if ( @@ -204,49 +205,17 @@ def _grade(self, subject, evidence): return CheckStatus.FAIL, ["invalid replay scope"], {} replays = [row for row in evidence.replays if row.scope != "seed"] samples = [evidence] + [row.evidence for row in replays] - measured = { - "completed_replays": sum( - not row.failure_reason - and any(exchange.operation == "step" for exchange in row.exchanges) - for row in samples - ) - } - for index, sample in enumerate(samples): - if sample.failure_reason: - return ( - CheckStatus.FAIL, - [f"replay {index}: collection failed"], - measured, - ) - if not any(row.operation == "step" for row in sample.exchanges): - return ( - CheckStatus.SKIP, - [ - evidence.replay_failure_reason - or f"replay {index}: no step was observed" - ], - measured, - ) + completed = [ + not row.failure_reason + and not row.failure_phase + and any(exchange.operation == "step" for exchange in row.exchanges) + for row in samples + ] + measured = {"completed_replays": sum(completed)} judged = subject.manifest.capabilities.llm_judged required = JUDGED_REPLAYS if judged else 3 - if len(samples) < required or {row.scope for row in replays} != { - "session", - "container", - }: - return ( - CheckStatus.SKIP, - [ - evidence.replay_failure_reason - or f"requires {required} completed replays across fresh sessions and containers" - ], - measured, - ) - if judged and len(samples) != JUDGED_REPLAYS: - return ( - CheckStatus.FAIL, - ["judged procedure requires exactly 20 samples"], - measured, - ) + # Inspect retained content before reporting an incomplete schedule: a + # timeout cannot conceal malformed wire data or an observed divergence. traces = [_trace(sample) for sample in samples] rewards = [] for trace in traces: @@ -273,13 +242,36 @@ def _grade(self, subject, evidence): # Policy-owned volatile exclusions are deliberately empty. All replays # use the same episode identity; no author field can suppress a difference. for index, trace in enumerate(traces[1:], 1): - mismatch = _difference(traces[0], trace) + if completed[0] and completed[index]: + mismatch = _difference(traces[0], trace) + else: + shared = min(len(traces[0]), len(trace)) + mismatch = _difference(traces[0][:shared], trace[:shared]) if mismatch: return ( CheckStatus.FAIL, [f"replay {index}: first divergence at {mismatch}"], measured, ) + if ( + not all(completed) + or len(samples) < required + or {row.scope for row in replays} != {"session", "container"} + ): + return ( + CheckStatus.SKIP, + [ + evidence.replay_failure_reason + or f"requires {required} completed replays across fresh sessions and containers" + ], + measured, + ) + if judged and len(samples) != JUDGED_REPLAYS: + return ( + CheckStatus.FAIL, + ["judged procedure requires exactly 20 samples"], + measured, + ) if judged: variances = [statistics.pvariance(values) for values in zip(*rewards)] measured.update( diff --git a/tests/test_validation/test_runtime_repeatability.py b/tests/test_validation/test_runtime_repeatability.py index c8eae8fe76..2434e03cd6 100644 --- a/tests/test_validation/test_runtime_repeatability.py +++ b/tests/test_validation/test_runtime_repeatability.py @@ -184,7 +184,8 @@ def test_fresh_sessions_without_fresh_container_are_incomplete(tmp_path): evidence = replace( subject.runtime_evidence, replays=tuple( - replace(row, scope="session") for row in subject.runtime_evidence.replays + replace(row, scope="session") if row.scope != "seed" else row + for row in subject.runtime_evidence.replays ), ) result = EpisodeDeterminismGrader().run(replace(subject, runtime_evidence=evidence)) @@ -282,7 +283,84 @@ def test_truncated_primary_evidence_never_passes(tmp_path, grader): evidence = replace( subject.runtime_evidence, failure_reason="step failed (TimeoutError)" ) + expected = ( + CheckStatus.SKIP if grader is EpisodeDeterminismGrader else CheckStatus.FAIL + ) + assert grader().run(replace(subject, runtime_evidence=evidence)).status is expected + + +@pytest.mark.parametrize("scope", ["session", "container"]) +def test_seed_proof_is_independent_of_other_replay_collection_failures(tmp_path, scope): + subject = subject_with_replays(tmp_path) + evidence = subject.runtime_evidence + replays = tuple( + replace( + replay, + evidence=replace( + replay.evidence, + failure_reason="step failed (TimeoutError)", + telemetry_json=None, + telemetry_error="unrelated replay telemetry unavailable", + ), + ) + if replay.scope == scope + else replay + for replay in evidence.replays + ) + result = SeedControlGrader().run( + replace(subject, runtime_evidence=replace(evidence, replays=replays)) + ) + assert result.status is CheckStatus.PASS + + +def test_changed_seed_rejection_still_fails_seed_control(tmp_path): + subject = subject_with_replays(tmp_path) + evidence = subject.runtime_evidence + changed = evidence.replays[-1] + changed = replace( + changed, + evidence=alter_telemetry( + changed.evidence, lambda value: value["seed"].update(accepted=False) + ), + ) + evidence = replace(evidence, replays=evidence.replays[:-1] + (changed,)) assert ( - grader().run(replace(subject, runtime_evidence=evidence)).status + SeedControlGrader().run(replace(subject, runtime_evidence=evidence)).status is CheckStatus.FAIL ) + + +@pytest.mark.parametrize("judged", [False, True]) +@pytest.mark.parametrize("fault", ["timeout", "malformed", "diverged"]) +def test_incomplete_replay_preserves_observed_failures(tmp_path, judged, fault): + subject = subject_with_replays(tmp_path, judged=judged) + evidence = subject.runtime_evidence + replay = evidence.replays[-2] + rows = list(replay.evidence.exchanges[:3]) + if fault == "malformed": + rows[2] = replace(rows[2], response_json="{not-json") + elif fault == "diverged": + rows = mutate_response( + rows, 2, lambda value: value["data"]["observation"].update(counter=99) + ) + partial = replace( + replay.evidence, + exchanges=tuple(rows), + failure_phase="step", + failure_reason="step failed (TimeoutError)", + ) + replay = replace(replay, evidence=partial) + evidence = replace( + evidence, + replays=evidence.replays[:-2] + (replay, evidence.replays[-1]), + replay_failure_reason="total replay deadline exceeded", + ) + result = EpisodeDeterminismGrader().run(replace(subject, runtime_evidence=evidence)) + assert result.status is ( + CheckStatus.SKIP if fault == "timeout" else CheckStatus.FAIL + ) + if fault == "timeout": + assert result.measured["completed_replays"] == (19 if judged else 2) + assert "reward_population_variance" not in result.measured + elif fault == "diverged": + assert "$[2].response.data.observation.counter" in result.evidence[0] From 0d4443192d167e0019f6b437457299ff1326f3d0 Mon Sep 17 00:00:00 2001 From: burtenshaw Date: Thu, 24 Sep 2026 12:41:56 +0200 Subject: [PATCH 04/11] fix: isolate runtime fixture faults --- tests/fixtures/validation/runtime/served_probe/app.py | 3 +-- tests/test_validation/integration/test_runtime_process.py | 2 +- 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/tests/fixtures/validation/runtime/served_probe/app.py b/tests/fixtures/validation/runtime/served_probe/app.py index e89b715002..aa50c3d58d 100644 --- a/tests/fixtures/validation/runtime/served_probe/app.py +++ b/tests/fixtures/validation/runtime/served_probe/app.py @@ -61,13 +61,12 @@ def reset(self, seed=None, episode_id=None, **kwargs): observation = ProbeObservation(counter=0, reward=0.0, done=False) if self.mode in { "nondeterministic", - "ignored_seed", "judged_stable", "judged_noisy", }: ProbeEnvironment._reset_ordinal += 1 self.ordinal = ProbeEnvironment._reset_ordinal - if self.mode in {"nondeterministic", "ignored_seed"}: + if self.mode == "nondeterministic": observation.metadata["session_ordinal"] = self.ordinal if self.mode in {"judged_stable", "judged_noisy"}: score = 0.5 if self.mode == "judged_stable" else float(self.ordinal % 2) diff --git a/tests/test_validation/integration/test_runtime_process.py b/tests/test_validation/integration/test_runtime_process.py index 8afe549ea3..d600fbbae0 100644 --- a/tests/test_validation/integration/test_runtime_process.py +++ b/tests/test_validation/integration/test_runtime_process.py @@ -139,7 +139,7 @@ def start(self, spec): ) subject = ProcessSubject(process, port, log_path, self.record_sha256) self.subjects.append(subject) - deadline = time.monotonic() + min(spec.startup_timeout_s, 10) + deadline = time.monotonic() + min(spec.startup_timeout_s, 30) try: with httpx.Client(trust_env=False, timeout=0.2) as client: while process.poll() is None and time.monotonic() < deadline: From 9f9b780d33ea4d1253fe1ba578b8e619e35e0d9a Mon Sep 17 00:00:00 2001 From: burtenshaw Date: Thu, 24 Sep 2026 14:10:57 +0200 Subject: [PATCH 05/11] fix: preserve grading after replay cleanup failure --- src/openenv/validation/runner.py | 8 ++- .../test_validation/test_runtime_execution.py | 55 +++++++++++++++++++ 2 files changed, 60 insertions(+), 3 deletions(-) diff --git a/src/openenv/validation/runner.py b/src/openenv/validation/runner.py index 586229cfd1..298d77fd46 100644 --- a/src/openenv/validation/runner.py +++ b/src/openenv/validation/runner.py @@ -189,9 +189,6 @@ def _runtime(subject, *, skip_build, provider): capabilities=manifest.capabilities, deadline=replay_deadline, ) - if any(replay.cleanup_complete is False for replay in evidence.replays): - result.status = CheckStatus.ERROR - result.evidence.append("replay subject teardown failed") subject = replace( subject, image_ref=image_ref, running=running, runtime_evidence=evidence ) @@ -208,6 +205,11 @@ def _runtime(subject, *, skip_build, provider): provider_capabilities=provider.capabilities, prior=[result], ) + # Cleanup failure cannot invalidate evidence already collected from a + # healthy subject. Preserve its findings before failing the run closed. + if any(replay.cleanup_complete is False for replay in evidence.replays): + result.status = CheckStatus.ERROR + result.evidence.append("replay subject teardown failed") except UnsupportedCapability as exc: result = _outcome( "runtime.startup", CheckStatus.SKIP, str(exc), started=started diff --git a/tests/test_validation/test_runtime_execution.py b/tests/test_validation/test_runtime_execution.py index f6f792b15e..a33c30949a 100644 --- a/tests/test_validation/test_runtime_execution.py +++ b/tests/test_validation/test_runtime_execution.py @@ -470,6 +470,61 @@ def replays(*args, **kwargs): assert replay["cleanup_complete"] is False +@pytest.mark.parametrize( + "fault,failed_check", + [ + ("invalid_reward", "runtime.reward_well_formed"), + ("replay_divergence", "runtime.episode_determinism"), + ], +) +def test_replay_cleanup_failure_preserves_independent_runtime_findings( + package, monkeypatch, tmp_path, fault, failed_check +): + baseline = measured_episode() + changed_rows = list(baseline.exchanges) + response = json.loads(changed_rows[2].response_json) + if fault == "invalid_reward": + response["data"]["reward"] = 2.0 + else: + response["data"]["observation"]["counter"] = 99 + changed_rows[2] = replace(changed_rows[2], response_json=json.dumps(response)) + telemetry = json.loads(baseline.telemetry_json) + telemetry["trajectory"]["records"][2]["response"] = response + changed = replace( + baseline, exchanges=tuple(changed_rows), telemetry_json=json.dumps(telemetry) + ) + if fault == "invalid_reward": + baseline = changed + combined = replace( + baseline, + replays=( + ReplayEvidence("session", baseline), + ReplayEvidence("container", changed, cleanup_complete=False), + ReplayEvidence("seed", measured_episode(seed=43)), + ), + ) + monkeypatch.setattr( + "openenv.validation.runner.collect_runtime_evidence", lambda *a, **k: baseline + ) + monkeypatch.setattr( + "openenv.validation.runner.collect_replays", lambda *a, **k: combined + ) + bundle = tmp_path / "bundle" + report = run_validation( + package, + max_level=Level.RUNTIME, + provider=FakeRuntimeProvider(), + artifacts_dir=bundle, + ) + checks = {row.check_id: row for row in report.results} + assert checks["runtime.startup"].status is CheckStatus.ERROR + assert "replay subject teardown failed" in checks["runtime.startup"].evidence + assert checks[failed_check].status is CheckStatus.FAIL + assert checks["runtime.state_contract"].status is CheckStatus.PASS + assert report.verdict.value == "fail" + assert json.loads((bundle / "cleanup.json").read_text())["completed"] is False + + def test_semantic_ceiling_does_not_claim_semantic_execution(package): report = run_validation(package, max_level=Level.SEMANTIC, skip_build=True) assert report.levels_run == [Level.STATIC] From 26f6bb166b61306bfe4f03f3bf1cdd7c8469d06f Mon Sep 17 00:00:00 2001 From: burtenshaw Date: Fri, 25 Sep 2026 09:17:26 +0200 Subject: [PATCH 06/11] fix: conceal keys in divergence reports --- .../graders/runtime/repeatability.py | 6 +-- .../test_runtime_repeatability.py | 51 +++++++++++++++++-- 2 files changed, 51 insertions(+), 6 deletions(-) diff --git a/src/openenv/validation/graders/runtime/repeatability.py b/src/openenv/validation/graders/runtime/repeatability.py index 8774b4cf67..538f5d3736 100644 --- a/src/openenv/validation/graders/runtime/repeatability.py +++ b/src/openenv/validation/graders/runtime/repeatability.py @@ -42,12 +42,12 @@ def _trace(evidence): def _difference(expected, observed, path="$"): - """First differing JSON path, without disclosing subject-controlled values.""" + """First differing location, exposing indices and key ordinals only.""" if type(expected) is not type(observed): return path if isinstance(expected, dict): - for key in sorted(expected.keys() | observed.keys()): - child = f"{path}.{key[:80]}" + for index, key in enumerate(sorted(expected.keys() | observed.keys())): + child = f"{path}." if key not in expected or key not in observed: return child mismatch = _difference(expected[key], observed[key], child) diff --git a/tests/test_validation/test_runtime_repeatability.py b/tests/test_validation/test_runtime_repeatability.py index 2434e03cd6..9bdde4a05e 100644 --- a/tests/test_validation/test_runtime_repeatability.py +++ b/tests/test_validation/test_runtime_repeatability.py @@ -175,7 +175,8 @@ def test_divergence_reports_first_path_without_values(tmp_path, field, value): evidence = replace(evidence, replays=(replay,) + evidence.replays[1:]) result = EpisodeDeterminismGrader().run(replace(subject, runtime_evidence=evidence)) assert result.status is CheckStatus.FAIL - assert f"$[2].response.data.{field}" in result.evidence[0] + field_index = {"done": 0, "observation": 1, "reward": 2}[field] + assert f"$[2]..." in result.evidence[0] assert "private-value" not in result.model_dump_json() @@ -238,7 +239,7 @@ def test_judged_mode_still_checks_observations_and_state(tmp_path): evidence = replace(evidence, replays=(replay,) + evidence.replays[1:]) result = EpisodeDeterminismGrader().run(replace(subject, runtime_evidence=evidence)) assert result.status is CheckStatus.FAIL - assert "$[3].response.data.step_count" in result.evidence[0] + assert "$[3]..." in result.evidence[0] def test_judged_variance_does_not_pool_different_steps(tmp_path): @@ -363,4 +364,48 @@ def test_incomplete_replay_preserves_observed_failures(tmp_path, judged, fault): assert result.measured["completed_replays"] == (19 if judged else 2) assert "reward_population_variance" not in result.measured elif fault == "diverged": - assert "$[2].response.data.observation.counter" in result.evidence[0] + assert "$[2]...." in result.evidence[0] + + +@pytest.mark.parametrize("grader", [EpisodeDeterminismGrader, TrajectoryRecordGrader]) +@pytest.mark.parametrize("mismatch", ["extra_key", "changed_value"]) +def test_public_divergence_paths_never_disclose_subject_keys( + tmp_path, grader, mismatch +): + subject = subject_with_replays(tmp_path) + evidence = subject.runtime_evidence + private_key = "hf_privateSubjectSecret123\nUNTRUSTED\r\tkey" + if mismatch == "changed_value": + rows = mutate_response( + list(evidence.exchanges), + 2, + lambda response: response["data"]["observation"].update({private_key: 0}), + ) + evidence = replace(evidence, exchanges=tuple(rows)) + if grader is EpisodeDeterminismGrader: + replay = evidence.replays[0] + rows = mutate_response( + list(replay.evidence.exchanges), + 2, + lambda response: response["data"]["observation"].update({private_key: 1}), + ) + replay = replace( + replay, evidence=replace(replay.evidence, exchanges=tuple(rows)) + ) + evidence = replace(evidence, replays=(replay,) + evidence.replays[1:]) + else: + evidence = alter_telemetry( + evidence, + lambda value: value["trajectory"]["records"][2]["response"]["data"][ + "observation" + ].update({private_key: 1}), + ) + result = grader().run(replace(subject, runtime_evidence=evidence)) + assert result.status is CheckStatus.FAIL + assert "$[2]...." in result.evidence[0] + assert private_key not in "".join(result.evidence) + assert "privateSubjectSecret" not in result.model_dump_json() + assert "UNTRUSTED" not in result.model_dump_json() + assert "\n" not in result.evidence[0] + assert "\r" not in result.evidence[0] + assert "\t" not in result.evidence[0] From 7dc4d9671c280609b58ed4056011b1a723618e45 Mon Sep 17 00:00:00 2001 From: burtenshaw Date: Fri, 25 Sep 2026 10:20:17 +0200 Subject: [PATCH 07/11] fix: require confirmed replay cleanup --- .../graders/runtime/repeatability.py | 9 +++ .../test_validation/test_runtime_execution.py | 14 ++++- .../test_runtime_repeatability.py | 61 +++++++++++++++++-- 3 files changed, 77 insertions(+), 7 deletions(-) diff --git a/src/openenv/validation/graders/runtime/repeatability.py b/src/openenv/validation/graders/runtime/repeatability.py index 538f5d3736..a3be3d66d1 100644 --- a/src/openenv/validation/graders/runtime/repeatability.py +++ b/src/openenv/validation/graders/runtime/repeatability.py @@ -290,6 +290,15 @@ def _grade(self, subject, evidence): ], measured, ) + if any( + row.scope == "container" and row.cleanup_complete is not True + for row in replays + ): + return ( + CheckStatus.SKIP, + ["fresh container cleanup was not confirmed"], + measured, + ) return ( CheckStatus.PASS, ["fresh-session and fresh-container replays agree"], diff --git a/tests/test_validation/test_runtime_execution.py b/tests/test_validation/test_runtime_execution.py index a33c30949a..3ca02e4feb 100644 --- a/tests/test_validation/test_runtime_execution.py +++ b/tests/test_validation/test_runtime_execution.py @@ -473,11 +473,12 @@ def replays(*args, **kwargs): @pytest.mark.parametrize( "fault,failed_check", [ + (None, None), ("invalid_reward", "runtime.reward_well_formed"), ("replay_divergence", "runtime.episode_determinism"), ], ) -def test_replay_cleanup_failure_preserves_independent_runtime_findings( +def test_replay_cleanup_failure_prevents_pass_and_preserves_runtime_findings( package, monkeypatch, tmp_path, fault, failed_check ): baseline = measured_episode() @@ -485,7 +486,7 @@ def test_replay_cleanup_failure_preserves_independent_runtime_findings( response = json.loads(changed_rows[2].response_json) if fault == "invalid_reward": response["data"]["reward"] = 2.0 - else: + elif fault == "replay_divergence": response["data"]["observation"]["counter"] = 99 changed_rows[2] = replace(changed_rows[2], response_json=json.dumps(response)) telemetry = json.loads(baseline.telemetry_json) @@ -519,7 +520,14 @@ def test_replay_cleanup_failure_preserves_independent_runtime_findings( checks = {row.check_id: row for row in report.results} assert checks["runtime.startup"].status is CheckStatus.ERROR assert "replay subject teardown failed" in checks["runtime.startup"].evidence - assert checks[failed_check].status is CheckStatus.FAIL + if failed_check: + assert checks[failed_check].status is CheckStatus.FAIL + else: + determinism = checks["runtime.episode_determinism"] + assert determinism.status is CheckStatus.SKIP + assert determinism.measured["completed_replays"] == 3 + assert determinism.evidence == ["fresh container cleanup was not confirmed"] + assert checks["runtime.reward_well_formed"].status is CheckStatus.PASS assert checks["runtime.state_contract"].status is CheckStatus.PASS assert report.verdict.value == "fail" assert json.loads((bundle / "cleanup.json").read_text())["completed"] is False diff --git a/tests/test_validation/test_runtime_repeatability.py b/tests/test_validation/test_runtime_repeatability.py index 9bdde4a05e..735b9a64ce 100644 --- a/tests/test_validation/test_runtime_repeatability.py +++ b/tests/test_validation/test_runtime_repeatability.py @@ -51,7 +51,11 @@ def subject_with_replays(tmp_path, *, judged=False): subject = subject_with(tmp_path) count = 20 if judged else 3 replays = tuple( - ReplayEvidence("container" if index == 1 else "session", sample(subject)) + ReplayEvidence( + "container" if index == 1 else "session", + sample(subject), + cleanup_complete=True if index == 1 else None, + ) for index in range(1, count) ) + (ReplayEvidence("seed", sample(subject, seed=43)),) if judged: @@ -193,6 +197,45 @@ def test_fresh_sessions_without_fresh_container_are_incomplete(tmp_path): assert result.status is CheckStatus.SKIP +@pytest.mark.parametrize("cleanup_complete", [False, None]) +@pytest.mark.parametrize("judged", [False, True]) +@pytest.mark.parametrize("fault", [None, "malformed", "diverged"]) +def test_unconfirmed_container_cleanup_prevents_pass_without_hiding_failures( + tmp_path, cleanup_complete, judged, fault +): + subject = subject_with_replays(tmp_path, judged=judged) + evidence = subject.runtime_evidence + replay = evidence.replays[0] + rows = list(replay.evidence.exchanges) + if fault == "malformed": + rows[2] = replace(rows[2], response_json="{not-json") + elif fault == "diverged": + rows = mutate_response( + rows, 2, lambda value: value["data"]["observation"].update(counter=99) + ) + replay = replace( + replay, + evidence=replace(replay.evidence, exchanges=tuple(rows)), + cleanup_complete=cleanup_complete, + ) + evidence = replace( + evidence, + replays=(replay,) + evidence.replays[1:], + replay_failure_reason="fresh container cleanup failed" + if cleanup_complete is False + else None, + ) + result = EpisodeDeterminismGrader().run(replace(subject, runtime_evidence=evidence)) + assert result.status is (CheckStatus.FAIL if fault else CheckStatus.SKIP) + if fault is None: + assert result.measured["completed_replays"] == (20 if judged else 3) + assert result.evidence == ["fresh container cleanup was not confirmed"] + elif fault == "diverged": + assert "first divergence" in result.evidence[0] + else: + assert result.evidence == ["malformed replay evidence"] + + def test_partial_judged_sample_is_incomplete(tmp_path): subject = subject_with_replays(tmp_path, judged=True) evidence = replace( @@ -204,16 +247,26 @@ def test_partial_judged_sample_is_incomplete(tmp_path): @pytest.mark.parametrize( - "low,bound,status", [(0.5, 0.1, CheckStatus.PASS), (0.0, 0.2, CheckStatus.FAIL)] + "low,bound,status,cleanup_complete", + [ + (0.5, 0.1, CheckStatus.PASS, True), + (0.5, 0.1, CheckStatus.SKIP, False), + (0.0, 0.2, CheckStatus.FAIL, True), + (0.0, 0.2, CheckStatus.FAIL, False), + ], ) def test_judged_population_variance_uses_reward_squared_units( - tmp_path, low, bound, status + tmp_path, low, bound, status, cleanup_complete ): subject = subject_with_replays(tmp_path, judged=True) subject.manifest.reward.variance_tolerance = bound evidence = subject.runtime_evidence replays = tuple( - replace(row, evidence=sample(subject, reward=low if index < 10 else 1.0)) + replace( + row, + evidence=sample(subject, reward=low if index < 10 else 1.0), + cleanup_complete=cleanup_complete if row.scope == "container" else None, + ) for index, row in enumerate(evidence.replays[:-1]) ) + (evidence.replays[-1],) result = EpisodeDeterminismGrader().run( From b2ca1eb8c196d2d9426077ae1decfe68c7ede9b6 Mon Sep 17 00:00:00 2001 From: burtenshaw Date: Thu, 1 Oct 2026 12:09:26 +0200 Subject: [PATCH 08/11] fix: honor repeatability startup dependencies --- tests/test_validation/test_runtime_repeatability.py | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/tests/test_validation/test_runtime_repeatability.py b/tests/test_validation/test_runtime_repeatability.py index 735b9a64ce..105e5a5085 100644 --- a/tests/test_validation/test_runtime_repeatability.py +++ b/tests/test_validation/test_runtime_repeatability.py @@ -337,10 +337,9 @@ def test_truncated_primary_evidence_never_passes(tmp_path, grader): evidence = replace( subject.runtime_evidence, failure_reason="step failed (TimeoutError)" ) - expected = ( - CheckStatus.SKIP if grader is EpisodeDeterminismGrader else CheckStatus.FAIL - ) - assert grader().run(replace(subject, runtime_evidence=evidence)).status is expected + result = grader().run(replace(subject, runtime_evidence=evidence)) + assert result.status is CheckStatus.SKIP + assert any("runtime.startup" in reason for reason in result.evidence) @pytest.mark.parametrize("scope", ["session", "container"]) From 7d4b03025663d2cb2758441af4dbf37f35fef5fb Mon Sep 17 00:00:00 2001 From: burtenshaw Date: Thu, 1 Oct 2026 12:09:58 +0200 Subject: [PATCH 09/11] fix: accept incomplete replay skip reasons --- tests/test_validation/test_runtime_repeatability.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_validation/test_runtime_repeatability.py b/tests/test_validation/test_runtime_repeatability.py index 105e5a5085..184eb1c616 100644 --- a/tests/test_validation/test_runtime_repeatability.py +++ b/tests/test_validation/test_runtime_repeatability.py @@ -339,7 +339,7 @@ def test_truncated_primary_evidence_never_passes(tmp_path, grader): ) result = grader().run(replace(subject, runtime_evidence=evidence)) assert result.status is CheckStatus.SKIP - assert any("runtime.startup" in reason for reason in result.evidence) + assert result.evidence @pytest.mark.parametrize("scope", ["session", "container"]) From 274ebf1e60b4bc1def7bfdcadf76018ede23faf6 Mon Sep 17 00:00:00 2001 From: burtenshaw Date: Fri, 2 Oct 2026 13:33:18 +0200 Subject: [PATCH 10/11] fix: handle unscored replay steps --- rfcs/008-environment-auto-validation.md | 4 ++ .../graders/runtime/repeatability.py | 22 +++++++-- .../test_runtime_repeatability.py | 48 +++++++++++++++++++ 3 files changed, 70 insertions(+), 4 deletions(-) diff --git a/rfcs/008-environment-auto-validation.md b/rfcs/008-environment-auto-validation.md index 65ae63e414..cc6baedab2 100644 --- a/rfcs/008-environment-auto-validation.md +++ b/rfcs/008-environment-auto-validation.md @@ -582,6 +582,10 @@ policy-owned volatile metadata can be excluded. Authors cannot exclude fields. For `llm_judged`, the bounded variance path uses 20 completed identical-input fresh replays and population reward variance in reward-squared units, compared to the declared bound. The total run budget bounds all samples; fewer than 20 is incomplete. +An unscored non-terminal step contributes no reward variance. Its null position must +agree across replays; a null/numeric mismatch is a divergence. Variance is measured +separately at each numeric step position, and an entirely unscored episode is +incomplete rather than passing. Terminal null rewards remain invalid. This procedure is a runtime check, not a statistical confidence claim. The initial implementation compares the baseline against a new session and an diff --git a/src/openenv/validation/graders/runtime/repeatability.py b/src/openenv/validation/graders/runtime/repeatability.py index a3be3d66d1..afe0f2fbfc 100644 --- a/src/openenv/validation/graders/runtime/repeatability.py +++ b/src/openenv/validation/graders/runtime/repeatability.py @@ -224,7 +224,11 @@ def _grade(self, subject, evidence): if row["operation"] != "step": continue if judged: - reward = row["response"]["data"]["reward"] + data = row["response"]["data"] + reward = data["reward"] + if reward is None and data.get("done") is False: + sample_rewards.append(None) + continue low, high = subject.manifest.reward.range if ( type(reward) not in (int, float) @@ -237,7 +241,8 @@ def _grade(self, subject, evidence): measured, ) sample_rewards.append(reward) - row["response"]["data"]["reward"] = None + # Keep null versus numeric positions visible to trace comparison. + data["reward"] = 0.0 rewards.append(sample_rewards) # Policy-owned volatile exclusions are deliberately empty. All replays # use the same episode identity; no author field can suppress a difference. @@ -273,7 +278,10 @@ def _grade(self, subject, evidence): measured, ) if judged: - variances = [statistics.pvariance(values) for values in zip(*rewards)] + variances = [ + None if values[0] is None else statistics.pvariance(values) + for values in zip(*rewards) + ] measured.update( reward_population_variance=variances, variance_units="reward_squared", @@ -282,7 +290,7 @@ def _grade(self, subject, evidence): if type(bound) not in (int, float) or not math.isfinite(bound) or bound < 0: return CheckStatus.FAIL, ["invalid declared variance bound"], measured for index, variance in enumerate(variances): - if variance > bound: + if variance is not None and variance > bound: return ( CheckStatus.FAIL, [ @@ -290,6 +298,12 @@ def _grade(self, subject, evidence): ], measured, ) + if not any(variance is not None for variance in variances): + return ( + CheckStatus.SKIP, + ["missing prerequisite: an observed numeric step reward"], + measured, + ) if any( row.scope == "container" and row.cleanup_complete is not True for row in replays diff --git a/tests/test_validation/test_runtime_repeatability.py b/tests/test_validation/test_runtime_repeatability.py index 184eb1c616..fae37aa0c3 100644 --- a/tests/test_validation/test_runtime_repeatability.py +++ b/tests/test_validation/test_runtime_repeatability.py @@ -321,6 +321,54 @@ def two_steps(evidence): assert result.measured["reward_population_variance"] == [0.0, 0.0] +@pytest.mark.parametrize("scored", [True, False]) +def test_judged_replays_allow_consistent_unscored_steps(tmp_path, scored): + subject = subject_with_replays(tmp_path, judged=True) + + def unscored(evidence): + rows = list(evidence.exchanges) + prefix = mutate_response( + [rows[2]], 0, lambda response: response["data"].update(reward=None) + )[0] + return replace( + evidence, + exchanges=tuple( + rows[:2] + [prefix, rows[3]] + (rows[2:] if scored else []) + ), + ) + + evidence = unscored(subject.runtime_evidence) + evidence = replace( + evidence, + replays=tuple( + replace(row, evidence=unscored(row.evidence)) for row in evidence.replays + ), + ) + result = EpisodeDeterminismGrader().run(replace(subject, runtime_evidence=evidence)) + assert result.status is (CheckStatus.PASS if scored else CheckStatus.SKIP) + assert result.measured["reward_population_variance"] == ( + [None, 0.0] if scored else [None] + ) + + +@pytest.mark.parametrize("terminal", [False, True]) +def test_judged_null_reward_cannot_hide_changed_scoring_or_terminal_failure( + tmp_path, terminal +): + subject = subject_with_replays(tmp_path, judged=True) + evidence = subject.runtime_evidence + replay = evidence.replays[0] + rows = mutate_response( + list(replay.evidence.exchanges), + 2, + lambda response: response["data"].update(reward=None, done=terminal), + ) + replay = replace(replay, evidence=replace(replay.evidence, exchanges=tuple(rows))) + evidence = replace(evidence, replays=(replay,) + evidence.replays[1:]) + result = EpisodeDeterminismGrader().run(replace(subject, runtime_evidence=evidence)) + assert result.status is CheckStatus.FAIL + + @pytest.mark.parametrize("grader", [SeedControlGrader, TrajectoryRecordGrader]) def test_malformed_telemetry_is_a_finding(tmp_path, grader): subject = subject_with_replays(tmp_path) From e8dec9de7ece6b0c22c84dfd7792cb7e80c66e32 Mon Sep 17 00:00:00 2001 From: burtenshaw Date: Mon, 5 Oct 2026 13:48:57 +0200 Subject: [PATCH 11/11] fix: distinguish incomplete seed probes --- .../graders/runtime/repeatability.py | 44 ++++++++-- .../test_runtime_repeatability.py | 87 ++++++++++++++++++- 2 files changed, 121 insertions(+), 10 deletions(-) diff --git a/src/openenv/validation/graders/runtime/repeatability.py b/src/openenv/validation/graders/runtime/repeatability.py index afe0f2fbfc..cd54219787 100644 --- a/src/openenv/validation/graders/runtime/repeatability.py +++ b/src/openenv/validation/graders/runtime/repeatability.py @@ -12,7 +12,7 @@ JUDGED_REPLAYS = 20 -def _trace(evidence): +def _trace(evidence, *, operation=None): trace = [ { "operation": row.operation, @@ -20,6 +20,7 @@ def _trace(evidence): "response": json.loads(row.response_json), } for row in evidence.exchanges + if operation is None or row.operation == operation ] for row in trace: json.dumps(row, allow_nan=False) @@ -75,8 +76,11 @@ def _telemetry(evidence): def _missing_telemetry(grader, evidence): + if evidence is not None and evidence.failure_reason: + # The shared runtime dependency reports incomplete primary collection. + return None if evidence is not None and evidence.telemetry_json is None: - failure = evidence.failure_reason or evidence.telemetry_error + failure = evidence.telemetry_error return CheckResult( check_id=grader.check_id, status=CheckStatus.FAIL if failure else CheckStatus.SKIP, @@ -95,7 +99,10 @@ def run(self, subject): evidence = subject.runtime_evidence result = _missing_telemetry(self, evidence) or super().run(subject) if result.status is CheckStatus.PASS: - if not any(replay.scope == "seed" for replay in evidence.replays): + seed_replays = [ + replay.evidence for replay in evidence.replays if replay.scope == "seed" + ] + if not seed_replays: return result.model_copy( update={ "status": CheckStatus.SKIP, @@ -105,6 +112,18 @@ def run(self, subject): ], } ) + if any( + sample.failure_reason and sample.telemetry_json is None + for sample in seed_replays + ): + return result.model_copy( + update={ + "status": CheckStatus.SKIP, + "evidence": [ + "different-seed reset experiment did not produce seed telemetry" + ], + } + ) return result def check(self, subject, evidence): @@ -114,13 +133,17 @@ def check(self, subject, evidence): [evidence] + [replay.evidence for replay in evidence.replays if replay.scope == "seed"] ): - if sample.failure_reason or sample.telemetry_error: - problems.append(f"replay {index}: reset or telemetry collection failed") - continue - if sample.telemetry_json is None: - problems.append(f"replay {index}: seed telemetry unavailable") + if sample.telemetry_error: + problems.append(f"replay {index}: telemetry collection failed") continue + incomplete = sample.failure_reason and sample.telemetry_json is None + if incomplete: + # Failed later actions leave seed forwarding unproven; an + # observed reset rejection remains a seed experiment failure. + _trace(sample, operation="reset") resets = [row for row in sample.exchanges if row.operation == "reset"] + if incomplete and not resets: + continue if len(resets) != 1: problems.append(f"replay {index}: expected one measured reset") continue @@ -132,6 +155,11 @@ def check(self, subject, evidence): original_seed = seed elif seed == original_seed: problems.append(f"replay {index}: scheduled seed was not changed") + if incomplete: + continue + if sample.telemetry_json is None: + problems.append(f"replay {index}: seed telemetry unavailable") + continue observed = _telemetry(sample).get("seed") if ( not isinstance(observed, dict) diff --git a/tests/test_validation/test_runtime_repeatability.py b/tests/test_validation/test_runtime_repeatability.py index fae37aa0c3..765be87d63 100644 --- a/tests/test_validation/test_runtime_repeatability.py +++ b/tests/test_validation/test_runtime_repeatability.py @@ -380,10 +380,15 @@ def test_malformed_telemetry_is_a_finding(tmp_path, grader): @pytest.mark.parametrize("grader", GRADERS) -def test_truncated_primary_evidence_never_passes(tmp_path, grader): +@pytest.mark.parametrize("telemetry_available", [True, False]) +def test_truncated_primary_evidence_never_passes(tmp_path, grader, telemetry_available): subject = subject_with_replays(tmp_path) evidence = replace( - subject.runtime_evidence, failure_reason="step failed (TimeoutError)" + subject.runtime_evidence, + failure_reason="step failed (TimeoutError)", + telemetry_json=subject.runtime_evidence.telemetry_json + if telemetry_available + else None, ) result = grader().run(replace(subject, runtime_evidence=evidence)) assert result.status is CheckStatus.SKIP @@ -431,6 +436,84 @@ def test_changed_seed_rejection_still_fails_seed_control(tmp_path): ) +@pytest.mark.parametrize( + "original_accepted,telemetry_error", + [(True, None), (False, None), (True, "session telemetry failed (ValueError)")], +) +@pytest.mark.parametrize("step_error", [False, True]) +def test_incomplete_changed_seed_probe_does_not_claim_a_seed_violation( + tmp_path, original_accepted, telemetry_error, step_error +): + subject = subject_with_replays(tmp_path) + evidence = alter_telemetry( + subject.runtime_evidence, + lambda value: value["seed"].update(accepted=original_accepted), + ) + changed = evidence.replays[-1] + rows = changed.evidence.exchanges[:2] + if step_error: + rows += ( + replace( + changed.evidence.exchanges[2], + response_json=json.dumps( + {"type": "error", "data": {"code": "STEP_ERROR"}} + ), + ), + ) + changed = replace( + changed, + evidence=replace( + changed.evidence, + exchanges=rows, + failure_phase="step", + failure_reason="step failed (STEP_ERROR)" + if step_error + else "step failed (TimeoutError)", + telemetry_json=None, + telemetry_error=telemetry_error, + ), + ) + evidence = replace(evidence, replays=evidence.replays[:-1] + (changed,)) + result = SeedControlGrader().run(replace(subject, runtime_evidence=evidence)) + assert result.status is ( + CheckStatus.SKIP + if original_accepted and telemetry_error is None + else CheckStatus.FAIL + ) + if telemetry_error is not None: + assert "telemetry collection failed" in result.evidence[0] + elif original_accepted: + assert result.evidence == [ + "different-seed reset experiment did not produce seed telemetry" + ] + else: + assert "seed was not observed as forwarded" in result.evidence[0] + + +def test_rejected_reset_is_still_a_seed_failure_without_telemetry(tmp_path): + subject = subject_with_replays(tmp_path) + evidence = subject.runtime_evidence + changed = evidence.replays[-1] + rejected = replace( + changed.evidence.exchanges[0], + response_json=json.dumps({"type": "error", "data": {"code": "RESET_ERROR"}}), + ) + changed = replace( + changed, + evidence=replace( + changed.evidence, + exchanges=(rejected,), + failure_phase="reset", + failure_reason="reset failed (RESET_ERROR)", + telemetry_json=None, + ), + ) + evidence = replace(evidence, replays=evidence.replays[:-1] + (changed,)) + result = SeedControlGrader().run(replace(subject, runtime_evidence=evidence)) + assert result.status is CheckStatus.FAIL + assert result.evidence == ["malformed runtime evidence"] + + @pytest.mark.parametrize("judged", [False, True]) @pytest.mark.parametrize("fault", ["timeout", "malformed", "diverged"]) def test_incomplete_replay_preserves_observed_failures(tmp_path, judged, fault):