285 lines
11 KiB
Python
285 lines
11 KiB
Python
"""Deterministic environmental falsification replay report."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import numpy as np
|
|
|
|
from evals.audio_sensorium.synth import synthesize as synthesize_audio
|
|
from evals.sensorimotor_sensorium.synth import synthesize as synthesize_sensorimotor
|
|
from evals.vision_sensorium.synth import synthesize as synthesize_vision
|
|
from sensorium.audio.canonical import canonicalize as canonicalize_audio
|
|
from sensorium.audio.compiler import AudioCompiler
|
|
from sensorium.audio.checksum import sha256_json
|
|
from sensorium.environment import (
|
|
ObservationUnitRef,
|
|
build_experiment_plan,
|
|
build_expected_observation_frame,
|
|
build_hypothesis_claim,
|
|
build_observation_frame,
|
|
compare_expected_to_observation,
|
|
run_falsification_scenario,
|
|
)
|
|
from sensorium.logs import import_witness_jsonl, import_witness_records
|
|
from sensorium.sensorimotor.compiler import SensorimotorCompiler
|
|
from sensorium.vision import VisionCompiler, canonicalize_image
|
|
from sensorium.vision.grid import iter_tile_signals
|
|
|
|
_ROOT = Path(__file__).resolve().parent
|
|
_AUDIO_SR = 24_000
|
|
|
|
|
|
def _load_json(name: str) -> dict[str, Any]:
|
|
return json.loads((_ROOT / name).read_text(encoding="utf-8"))
|
|
|
|
|
|
def _load_jsonl(name: str) -> list[dict[str, Any]]:
|
|
return [
|
|
json.loads(line)
|
|
for line in (_ROOT / name).read_text(encoding="utf-8").splitlines()
|
|
if line.strip()
|
|
]
|
|
|
|
|
|
def _trace_safe(value: object) -> bool:
|
|
if isinstance(value, (np.ndarray, bytes, bytearray)):
|
|
return False
|
|
if isinstance(value, dict):
|
|
return all(_trace_safe(child) for child in value.values())
|
|
if isinstance(value, (list, tuple)):
|
|
return all(_trace_safe(child) for child in value)
|
|
return True
|
|
|
|
|
|
def _compile_unit(spec: dict[str, Any]):
|
|
modality = spec["modality"]
|
|
signal = spec["signal"]
|
|
if modality == "audio":
|
|
return AudioCompiler().compile_signal(
|
|
canonicalize_audio(synthesize_audio(signal), _AUDIO_SR)
|
|
)
|
|
if modality == "vision":
|
|
image = canonicalize_image(synthesize_vision(signal))
|
|
tile = iter_tile_signals(image)[0]
|
|
return VisionCompiler().compile_tile(tile)
|
|
if modality == "sensorimotor":
|
|
return SensorimotorCompiler().compile_signal(synthesize_sensorimotor(signal))
|
|
raise ValueError(f"unsupported falsification fixture modality: {modality!r}")
|
|
|
|
|
|
def _refs(spec: dict[str, Any]) -> tuple[ObservationUnitRef, ...]:
|
|
return tuple(
|
|
ObservationUnitRef(slot_id=slot_id, unit=_compile_unit(unit_spec))
|
|
for slot_id, unit_spec in sorted(spec.items())
|
|
)
|
|
|
|
|
|
def _actual_spec(case: dict[str, Any]) -> dict[str, Any]:
|
|
actual = case["actual"]
|
|
if actual == "same":
|
|
return case["expected"]
|
|
return actual
|
|
|
|
|
|
def _case_report(index: int, case: dict[str, Any]) -> dict[str, object]:
|
|
expected_refs = _refs(case["expected"])
|
|
actual_refs = _refs(_actual_spec(case))
|
|
expected = build_expected_observation_frame(
|
|
monotonic_tick=index,
|
|
source_clock="environment-falsification-fixture",
|
|
unit_refs=expected_refs,
|
|
causal_parent_ids=(),
|
|
)
|
|
actual = build_observation_frame(
|
|
monotonic_tick=index,
|
|
source_clock="environment-falsification-fixture",
|
|
units=tuple(ref.unit for ref in actual_refs),
|
|
causal_parent_ids=(expected.expected_id,),
|
|
)
|
|
run = compare_expected_to_observation(expected, actual, actual_refs=actual_refs)
|
|
expected_verdict = str(case["expected_verdict"])
|
|
row = {
|
|
"id": case["id"],
|
|
"expected_verdict": expected_verdict,
|
|
"actual_verdict": run.verdict,
|
|
"verdict_ok": run.verdict == expected_verdict,
|
|
"trace_hygiene_ok": _trace_safe(run.as_dict()),
|
|
"expected_sha256": expected.expected_sha256,
|
|
"actual_trace_hash": actual.trace_hash,
|
|
"run_trace_hash": run.trace_hash,
|
|
"residual": run.residual.as_dict(),
|
|
}
|
|
return row
|
|
|
|
|
|
def _report_hash(report_without_hash: dict[str, object]) -> str:
|
|
return sha256_json(report_without_hash)
|
|
|
|
|
|
def _frame_report(cases: list[dict[str, object]]) -> dict[str, object]:
|
|
passed = sum(
|
|
1
|
|
for case in cases
|
|
if case["verdict_ok"] is True and case["trace_hygiene_ok"] is True
|
|
)
|
|
return {
|
|
"lane": "environment-falsification",
|
|
"version": "v1",
|
|
"total": len(cases),
|
|
"passed": passed,
|
|
"failed": len(cases) - passed,
|
|
"cases": cases,
|
|
}
|
|
|
|
|
|
def _scenario_case_report(index: int, scenario: dict[str, Any]) -> dict[str, object]:
|
|
hypothesis_spec = scenario["hypothesis"]
|
|
hypothesis = build_hypothesis_claim(
|
|
claim_id=str(hypothesis_spec["claim_id"]),
|
|
claim_text=str(hypothesis_spec["claim_text"]),
|
|
domain=str(hypothesis_spec["domain"]),
|
|
basis_trace_hashes=tuple(hypothesis_spec.get("basis_trace_hashes", ())),
|
|
)
|
|
expected_frames = []
|
|
actual_frames_by_expected_id = {}
|
|
actual_refs_by_expected_id = {}
|
|
for offset, frame_spec in enumerate(scenario["frames"]):
|
|
tick = int(frame_spec.get("tick", index * 100 + offset))
|
|
expected_refs = _refs(frame_spec["expected"])
|
|
expected = build_expected_observation_frame(
|
|
monotonic_tick=tick,
|
|
source_clock="environment-falsification-scenario-fixture",
|
|
unit_refs=expected_refs,
|
|
causal_parent_ids=tuple(frame_spec.get("causal_parent_ids", ())),
|
|
)
|
|
expected_frames.append(expected)
|
|
if frame_spec["actual"] == "missing":
|
|
continue
|
|
actual_spec = frame_spec["expected"] if frame_spec["actual"] == "same" else frame_spec["actual"]
|
|
actual_refs = _refs(actual_spec)
|
|
actual = build_observation_frame(
|
|
monotonic_tick=tick,
|
|
source_clock="environment-falsification-scenario-fixture",
|
|
units=tuple(ref.unit for ref in actual_refs),
|
|
causal_parent_ids=(expected.expected_id,),
|
|
)
|
|
actual_frames_by_expected_id[expected.expected_id] = actual
|
|
actual_refs_by_expected_id[expected.expected_id] = actual_refs
|
|
|
|
plan = build_experiment_plan(hypothesis=hypothesis, expected_frames=expected_frames)
|
|
report = run_falsification_scenario(
|
|
plan,
|
|
actual_frames_by_expected_id=actual_frames_by_expected_id,
|
|
actual_refs_by_expected_id=actual_refs_by_expected_id,
|
|
)
|
|
expected_verdict = str(scenario["expected_verdict"])
|
|
row = {
|
|
"id": scenario["id"],
|
|
"expected_verdict": expected_verdict,
|
|
"actual_verdict": report.verdict,
|
|
"verdict_ok": report.verdict == expected_verdict,
|
|
"trace_hygiene_ok": _trace_safe(report.as_dict()),
|
|
"hypothesis_sha256": hypothesis.hypothesis_sha256,
|
|
"plan_sha256": plan.plan_sha256,
|
|
"scenario_sha256": report.scenario_sha256,
|
|
"scenario_report_sha256": report.report_sha256,
|
|
"total_count": report.total_count,
|
|
"supported_count": report.supported_count,
|
|
"falsified_count": report.falsified_count,
|
|
"run_trace_hashes": [run.trace_hash for run in report.runs],
|
|
}
|
|
return row
|
|
|
|
|
|
def _witness_import_report() -> dict[str, object]:
|
|
payloads = _load_json("witness_payloads.json")["payloads"]
|
|
|
|
def resolve(payload_ref: str):
|
|
return _compile_unit(payloads[payload_ref])
|
|
|
|
path = _ROOT / "witness_log.jsonl"
|
|
imported = import_witness_jsonl(path, resolve_payload_ref=resolve)
|
|
repeated = import_witness_jsonl(path, resolve_payload_ref=resolve)
|
|
rows = _load_jsonl("witness_log.jsonl")
|
|
permuted = import_witness_records(reversed(rows), resolve_payload_ref=resolve)
|
|
trace = imported.as_dict()
|
|
frame_trace_hashes = [frame.trace_hash for frame in imported.frames]
|
|
return {
|
|
"id": "jsonl_witness_import",
|
|
"record_count": imported.manifest.record_count,
|
|
"frame_count": len(imported.frames),
|
|
"trace_hash": imported.trace_hash,
|
|
"manifest_sha256": imported.manifest.manifest_sha256,
|
|
"frame_trace_hashes": frame_trace_hashes,
|
|
"deterministic_reimport_ok": imported.trace_hash == repeated.trace_hash,
|
|
"order_stability_ok": imported.trace_hash == permuted.trace_hash,
|
|
"trace_hygiene_ok": _trace_safe(trace) and "pixels" not in str(trace) and "action_trace" not in str(trace),
|
|
"no_actuation_ok": all(
|
|
not unit.pack_id.startswith("motor")
|
|
for frame in imported.frames
|
|
for unit in frame.units
|
|
),
|
|
}
|
|
|
|
|
|
def build_environment_falsification_report() -> dict[str, object]:
|
|
fixtures = _load_json("fixtures.json")["fixtures"]
|
|
cases = [_case_report(idx, case) for idx, case in enumerate(fixtures)]
|
|
frame_report = _frame_report(cases)
|
|
frame_report_sha256 = _report_hash(frame_report)
|
|
scenario_fixtures = _load_json("scenario_fixtures.json")["scenarios"]
|
|
scenario_cases = [
|
|
_scenario_case_report(idx, scenario)
|
|
for idx, scenario in enumerate(scenario_fixtures)
|
|
]
|
|
scenario_passed = sum(
|
|
1
|
|
for case in scenario_cases
|
|
if case["verdict_ok"] is True and case["trace_hygiene_ok"] is True
|
|
)
|
|
witness_import = _witness_import_report()
|
|
witness_ok = all(
|
|
witness_import[key] is True
|
|
for key in (
|
|
"deterministic_reimport_ok",
|
|
"order_stability_ok",
|
|
"trace_hygiene_ok",
|
|
"no_actuation_ok",
|
|
)
|
|
)
|
|
frame_passed = int(frame_report["passed"])
|
|
frame_failed = int(frame_report["failed"])
|
|
total = len(cases) + len(scenario_cases) + 1
|
|
passed = frame_passed + scenario_passed + (1 if witness_ok else 0)
|
|
report = {
|
|
"lane": "environment-falsification",
|
|
"version": "v1",
|
|
"total": total,
|
|
"passed": passed,
|
|
"failed": frame_failed + (len(scenario_cases) - scenario_passed) + (0 if witness_ok else 1),
|
|
"cases": cases,
|
|
"frame_report_sha256": frame_report_sha256,
|
|
"scenario_cases": scenario_cases,
|
|
"witness_import": witness_import,
|
|
}
|
|
report["report_sha256"] = _report_hash(report)
|
|
expected_hashes = _load_json("expected_hashes.json")
|
|
expected_frame_report_sha256 = expected_hashes.get(
|
|
"frame_report_sha256",
|
|
expected_hashes["report_sha256"],
|
|
)
|
|
report["expected_frame_report_sha256"] = expected_frame_report_sha256
|
|
report["expected_frame_report_hash_ok"] = frame_report_sha256 == expected_frame_report_sha256
|
|
report["expected_report_sha256"] = expected_hashes["report_sha256"]
|
|
report["expected_report_hash_ok"] = report["report_sha256"] == expected_hashes["report_sha256"]
|
|
if not report["expected_frame_report_hash_ok"]:
|
|
report["failed"] = int(report["failed"]) + 1
|
|
if not report["expected_report_hash_ok"]:
|
|
report["failed"] = int(report["failed"]) + 1
|
|
return report
|
|
|
|
|
|
__all__ = ["build_environment_falsification_report"]
|