diff --git a/CONTEXT.md b/CONTEXT.md index cfd99f8..f088ef1 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -12,6 +12,26 @@ _Avoid_: Model Judge, external VLM judge, standalone VLM evaluator A compact set of visual evidence, metrics, baseline comparisons, and decision metadata prepared for agent visual review or human escalation. _Avoid_: Screenshot folder, dashboard dump, artifact pile +**Evidence Producer**: +A downstream-project-owned runner that executes robot, simulator, or hardware-specific behavior and emits roboharness evidence artifacts. Roboharness may define artifact schemas and assembly rules, but it does not own the downstream robot runtime, safety startup, controller transport, or task planner. +_Avoid_: Roboharness runtime, hidden simulator backend, adapter that owns project control + +**Semantic Snapshot**: +A named semantic phase sample containing the project state and metrics needed for deterministic replay, checking, or review. It is not required to come from a `SimulatorBackend.step()` loop. +_Avoid_: Raw screenshot, simulator checkpoint only, full episode recording + +**Semantic Snapshot Bundle**: +An ordered, replayable set of **Semantic Snapshot** records plus run metadata. It is the evidence-producer handoff into renderers, metric checks, log diagnostics, proof-pack assembly, or visual review preparation. +_Avoid_: Screenshot folder, ad hoc artifact directory, renderer-specific dump + +**Renderer Report**: +The structured output from rendering a **Semantic Snapshot Bundle** through one renderer. It records capture/motion trustworthiness, per-snapshot image evidence, renderer metadata, and flags; it does not decide final approval by itself. +_Avoid_: Visual verdict, screenshot list, renderer log only + +**Autonomous Evidence Report**: +The machine-readable report that aggregates run identity, plan/runtime metadata, summary metrics, snapshot metrics, renderer reports, verdict reasons, and failure taxonomy before proof-pack approval or human escalation. +_Avoid_: HTML report, visual review record, baseline blessing + **Visual Reviewer Invocation**: An isolated agent visual review call that reads a proof pack and returns a structured verdict. It may run through a subagent, CLI trigger, CI job, or MCP tool, but its context is bounded to the evidence and review contract. _Avoid_: Separate VLM, same-session self-review, screenshot glance diff --git a/docs/adr/0002-post-hoc-semantic-snapshot-evidence.md b/docs/adr/0002-post-hoc-semantic-snapshot-evidence.md new file mode 100644 index 0000000..d4e349c --- /dev/null +++ b/docs/adr/0002-post-hoc-semantic-snapshot-evidence.md @@ -0,0 +1,56 @@ +# Post-hoc semantic snapshot evidence is first-class + +Status: accepted + +Roboharness will support post-hoc semantic snapshot evidence as a first-class +artifact path alongside the existing `SimulatorBackend.step()` harness loop. A +downstream project may run its own robot runtime, simulator launch topology, +hardware-live surface, safety checks, and controller transport, then hand +Roboharness an ordered semantic snapshot bundle. Roboharness owns the reusable +artifact language around that handoff: semantic snapshot bundles, renderer +reports, autonomous evidence reports, proof-pack assembly inputs, baseline +comparison surfaces, and bounded visual review preparation. + +The `SimulatorBackend` protocol remains the simple package-first path for +step-oriented simulators. It is not the only way to use Roboharness. This keeps +hardware-style and deploy-live flows from pretending to be action-sequence +loops just to fit the original core API. + +Project-specific execution stays outside Roboharness. The downstream project +owns task planning, runtime startup, robot model loading, safety evidence, +control backend selection, renderer implementation, and any domain-specific +semantic checks. Roboharness may validate and aggregate the artifacts those +systems emit, but it must not absorb the project runtime. + +The first dogfood target is GR00T whole-body control. That repo already +demonstrates a useful flow: run a representative manipulation surface, record +semantic phase snapshots, replay those snapshots through Meshcat or MuJoCo, +then aggregate snapshot metrics, renderer trustworthiness, verdict reasons, and +failure taxonomy into an autonomous report. If Roboharness cannot express that +flow naturally, the Roboharness abstraction is incomplete. + +Consequences: + +- Add a public `roboharness.evidence` package for artifact models and JSON + helpers. +- Keep `roboharness.core.Harness` and `SimulatorBackend` intact as one ingestion + path rather than broadening them into downstream runtime orchestration. +- Keep `roboharness.approval` focused on paired evidence, visual review records, + and approval summaries. +- Let project `HarnessContract` workflows refer to evidence expectations over + time, but do not turn the contract into a renderer or runtime configuration + system. +- Validate GR00T-style snapshot bundles and renderer reports through fixtures + before migrating the GR00T runner itself. + +Rejected alternatives: + +- A GR00T-only adapter layer in the downstream repo. This would prove only that + another wrapper can be written, not that Roboharness has the right reusable + evidence abstraction. +- Forcing deploy-live and hardware-live evidence through `SimulatorBackend`. + That would couple Roboharness to action-loop assumptions that are false for + many real robot validation surfaces. +- Moving robot runtime/session ownership into Roboharness. That would expand + scope beyond the evidence and approval layer and make each downstream robot + integration harder to maintain. diff --git a/docs/plans/gr00t-dogfood-semantic-snapshot-evidence.md b/docs/plans/gr00t-dogfood-semantic-snapshot-evidence.md new file mode 100644 index 0000000..1c3b09f --- /dev/null +++ b/docs/plans/gr00t-dogfood-semantic-snapshot-evidence.md @@ -0,0 +1,106 @@ +# GR00T Dogfood: Semantic Snapshot Evidence + +Status: active + +## Context + +GR00T whole-body control built a mature visual harness around semantic phase +snapshots, deterministic replay rendering, renderer trustworthiness reports, +and autonomous visual reports. That success should move the reusable evidence +language back into Roboharness instead of leaving each downstream repository to +rebuild the same layer. + +This is dogfooding in the strict sense: GR00T should be able to use +Roboharness for the visual evidence lifecycle without Roboharness taking over +GR00T's robot runtime, task planner, control backends, safety startup, or +renderer implementation. + +## Accepted Direction + +- Roboharness owns public artifact concepts for semantic snapshot bundles, + renderer reports, and autonomous evidence reports. +- GR00T visual harness dogfooding treats `roboharness.evidence` as a hard + artifact-layer dependency, not an optional fallback. During local development, + GR00T installs the sibling Roboharness checkout into its active `.venv`; + after the API supports the repo's visual harness, GR00T can switch to a `uv` + git-main dependency. +- `SimulatorBackend.step()` remains one useful ingestion path, not the only + Roboharness model. +- Downstream projects own evidence producers: runtime sessions, robot-specific + execution, semantic snapshot recording, and renderer implementations. +- Roboharness owns reusable artifact validation, JSON round-tripping, + proof-pack assembly inputs, baseline/approval surfaces, and bounded visual + review preparation. +- GR00T compatibility is a first acceptance gate, but the first Roboharness + slice uses small fixtures rather than importing or depending on the GR00T + repository. + +## Non-Goals + +- Do not migrate the GR00T runner in this slice. +- Do not add a new simulator backend. +- Do not change MuJoCo wedge approval semantics. +- Do not make `HarnessContract` a runtime or renderer configuration system. +- Do not bless new baselines or alter human escalation policy. + +## First Slice + +1. Add a public `roboharness.evidence` package with typed artifact models and + JSON helpers. +2. Support GR00T-style `snapshot_bundle.json` and renderer report shapes through + round-trip tests. +3. Export the new artifact API from package boundaries without importing + optional simulator dependencies. +4. Record the architecture direction in `CONTEXT.md` and ADR 0002. + +## Acceptance Gates + +- A core-package import still works without optional simulator extras. +- Unit tests prove semantic snapshot bundles, renderer reports, and autonomous + evidence reports can be parsed and re-emitted without losing unknown + downstream fields. +- Focused lint passes for touched Python files. + +## Intuitive Flow Precheck + +- Scope: keep this slice to public artifact models, JSON helpers, exports, and + fixture tests. Do not migrate downstream runners. +- Risks: preserve unknown downstream fields so GR00T can adopt the API without + losing project-specific metrics, runtime, plan, or renderer metadata. +- Tests: add focused unit tests for GR00T-style snapshot bundles, renderer + reports, autonomous evidence reports, and core package import without + optional simulator extras. +- DX: expose the API from `roboharness.evidence` and top-level package exports + so downstream contracts can import it without depending on examples. +- Execution: implement in one bounded slice, then run focused tests and lint on + touched Python files. + +## Later Slices + +- [x] Add a GR00T dogfood gate that installs Roboharness explicitly and fails + when `roboharness.evidence` is missing or stale. +- [x] Generate visual review manifests from evidence bundles and contract + dimensions. +- [x] Add proof-pack assembly helpers that consume autonomous evidence reports. +- [x] Teach a GR00T project harness skill to use the new Roboharness evidence + API. +- Migrate GR00T visual harness code only after fixture compatibility and public + boundaries are stable. + +## Follow-Up Slice: Proof Pack And Static Review + +Implemented after the first artifact-model slice: + +- `roboharness.evidence` now exposes case-level proof-pack assembly for + downstream visual harness case directories. +- The first proof-pack assembler consumes `autonomous_report.json`, + `snapshot_bundle.json`, and renderer `report.json` files without importing + GR00T or owning downstream runtime execution. +- Static visual-review manifest generation is current-only and selects + case-local keyframes from the proof pack. It sets + `allow_automatic_visual_pass=false`, so current-only review can veto or + escalate but cannot bless an automatic visual pass. +- The `roboharness proof-pack` CLI writes `proof_pack.json` and optionally a + validated `visual_review_manifest.json`. +- GR00T owns the project-specific dogfood gate and generated project harness + contract under its `skills/visual-harness/roboharness/` directory. diff --git a/src/roboharness/__init__.py b/src/roboharness/__init__.py index 59bc33d..b743b24 100644 --- a/src/roboharness/__init__.py +++ b/src/roboharness/__init__.py @@ -43,6 +43,26 @@ evaluate_policy, ) from roboharness.evaluate.result import EvaluationResult, Operator, Severity, Verdict +from roboharness.evidence import ( + AutonomousEvidenceReport, + CaseProofPack, + ProofPackArtifact, + ProofPackImageRef, + RenderedImage, + RendererReport, + RendererSnapshot, + SemanticSnapshot, + SemanticSnapshotBundle, + SuiteProofPack, + SuiteProofPackCase, + VisualReviewQueue, + VisualReviewQueueItem, + build_case_proof_pack, + build_paired_visual_review_manifest, + build_static_visual_review_manifest, + build_suite_proof_pack, + build_visual_review_queue, +) from roboharness.runner import BatchResult, ParallelTrialRunner, TrialSpec from roboharness.storage.history import EvaluationHistory, EvaluationRecord, TrendResult @@ -55,8 +75,10 @@ "REACH_PROTOCOL", "ApprovalPolicy", "AssertionEngine", + "AutonomousEvidenceReport", "BatchResult", "CaptureResult", + "CaseProofPack", "Checkpoint", "CheckpointStore", "ComponentAssumption", @@ -79,8 +101,17 @@ "MetricGate", "Operator", "ParallelTrialRunner", + "ProofPackArtifact", + "ProofPackImageRef", + "RenderedImage", + "RendererReport", + "RendererSnapshot", "SemanticPhase", + "SemanticSnapshot", + "SemanticSnapshotBundle", "Severity", + "SuiteProofPack", + "SuiteProofPackCase", "TaskPhase", "TaskProtocol", "TrendResult", @@ -88,6 +119,13 @@ "ValidationCommand", "Verdict", "VisualReviewDimension", + "VisualReviewQueue", + "VisualReviewQueueItem", + "build_case_proof_pack", + "build_paired_visual_review_manifest", + "build_static_visual_review_manifest", + "build_suite_proof_pack", + "build_visual_review_queue", "check_eval_threshold", "default_registry", "evaluate_policy", diff --git a/src/roboharness/approval/__init__.py b/src/roboharness/approval/__init__.py index 177b928..e648f31 100644 --- a/src/roboharness/approval/__init__.py +++ b/src/roboharness/approval/__init__.py @@ -13,20 +13,24 @@ from roboharness.approval.visual_review import ( MANIFEST_SCHEMA_VERSION, RECORD_SCHEMA_VERSION, + VISUAL_REVIEW_SUMMARY_SCHEMA_VERSION, VisualReviewPackage, VisualReviewResult, VisualReviewValidationError, build_visual_review_prompt, build_visual_review_schema, + build_visual_review_summary, ingest_visual_review_record, validate_visual_review_manifest, validate_visual_review_record, write_visual_review_package, + write_visual_review_summary, ) __all__ = [ "MANIFEST_SCHEMA_VERSION", "RECORD_SCHEMA_VERSION", + "VISUAL_REVIEW_SUMMARY_SCHEMA_VERSION", "EvidencePair", "EvidenceStatus", "EvidenceTarget", @@ -36,6 +40,7 @@ "VisualReviewValidationError", "build_visual_review_prompt", "build_visual_review_schema", + "build_visual_review_summary", "ingest_visual_review_record", "render_lightbox_shell", "render_zoomable_image", @@ -44,4 +49,5 @@ "validate_visual_review_manifest", "validate_visual_review_record", "write_visual_review_package", + "write_visual_review_summary", ] diff --git a/src/roboharness/approval/visual_review.py b/src/roboharness/approval/visual_review.py index cc61375..9915b63 100644 --- a/src/roboharness/approval/visual_review.py +++ b/src/roboharness/approval/visual_review.py @@ -12,6 +12,7 @@ MANIFEST_SCHEMA_VERSION = "roboharness_visual_review_manifest/v1" RECORD_SCHEMA_VERSION = "roboharness_visual_review/v1" +VISUAL_REVIEW_SUMMARY_SCHEMA_VERSION = "roboharness_visual_review_summary/v1" SUPPORTED_DIMENSIONS = frozenset( { @@ -98,6 +99,56 @@ def __init__(self, errors: Sequence[str]): super().__init__("; ".join(self.errors)) +def build_visual_review_summary( + manifest: Mapping[str, Any], + record: Mapping[str, Any], + *, + manifest_path: str = "visual_review_manifest.json", + record_path: str = "visual_review.json", +) -> dict[str, Any]: + """Build a persisted summary from a bounded visual review record.""" + + result = ingest_visual_review_record( + manifest, + record, + manifest_path=manifest_path, + record_path=record_path, + ) + case_id = manifest.get("case_id") if isinstance(manifest.get("case_id"), str) else "" + if not case_id and isinstance(record.get("case_id"), str): + case_id = str(record["case_id"]) + return { + "schema_version": VISUAL_REVIEW_SUMMARY_SCHEMA_VERSION, + "case_id": case_id, + "is_valid": result.is_valid, + "effective_visual_verdict": result.effective_visual_verdict, + "summary": dict(result.summary), + } + + +def write_visual_review_summary( + manifest: Mapping[str, Any], + record: Mapping[str, Any], + path: str | Path, + *, + manifest_path: str = "visual_review_manifest.json", + record_path: str = "visual_review.json", +) -> Path: + """Write a persisted summary from a bounded visual review record.""" + + output_path = Path(path) + save_json( + build_visual_review_summary( + manifest, + record, + manifest_path=manifest_path, + record_path=record_path, + ), + output_path, + ) + return output_path + + def build_visual_review_schema() -> dict[str, Any]: """Return the JSON schema given to the visual reviewer.""" return { diff --git a/src/roboharness/cli.py b/src/roboharness/cli.py index 821b814..c0bf771 100644 --- a/src/roboharness/cli.py +++ b/src/roboharness/cli.py @@ -9,6 +9,10 @@ from typing import Any from roboharness._utils import load_json as _load_json +from roboharness.approval.visual_review import ( + validate_visual_review_manifest, + write_visual_review_summary, +) from roboharness.contract import ( check_project_harness_skill, generate_project_harness_skill, @@ -24,6 +28,11 @@ ) from roboharness.evaluate.constraints import load_constraints from roboharness.evaluate.defaults import GRASP_DEFAULTS +from roboharness.evidence import ( + build_case_proof_pack, + build_static_visual_review_manifest, + write_case_proof_pack, +) from roboharness.storage.history import EvaluationHistory @@ -374,6 +383,71 @@ def main(argv: list[str] | None = None) -> int: help="Minimum success rate (0.0-1.0) for CI pass/fail. Exit 1 if below.", ) + # proof-pack + proof_pack_parser = subparsers.add_parser( + "proof-pack", + help="Assemble a case proof pack from downstream visual harness artifacts.", + ) + proof_pack_parser.add_argument( + "case_dir", + type=Path, + help="Case directory containing autonomous_report.json.", + ) + proof_pack_parser.add_argument( + "--output", + type=Path, + default=None, + help="Where to write proof_pack.json. Defaults to /proof_pack.json.", + ) + proof_pack_parser.add_argument( + "--visual-review-manifest", + type=Path, + default=None, + help="Optionally write a current-only static visual review manifest.", + ) + proof_pack_parser.add_argument( + "--task-intent", + type=str, + default="Review the visual harness case for obvious static pose and task success issues.", + help="Task intent used when writing a visual review manifest.", + ) + proof_pack_parser.add_argument( + "--format", + dest="output_format", + choices=["human", "json"], + default="human", + help="Output format (default: human).", + ) + + # visual-review-summary + visual_review_summary_parser = subparsers.add_parser( + "visual-review-summary", + help="Validate a visual review record and write its approval summary.", + ) + visual_review_summary_parser.add_argument( + "manifest", + type=Path, + help="Path to visual_review_manifest.json.", + ) + visual_review_summary_parser.add_argument( + "record", + type=Path, + help="Path to visual_review.json.", + ) + visual_review_summary_parser.add_argument( + "--output", + type=Path, + default=None, + help="Where to write visual_review_summary.json. Defaults next to the record.", + ) + visual_review_summary_parser.add_argument( + "--format", + dest="output_format", + choices=["human", "json"], + default="human", + help="Output format (default: human).", + ) + # contract contract_parser = subparsers.add_parser( "contract", @@ -523,6 +597,72 @@ def main(argv: list[str] | None = None) -> int: exit_code = 1 return exit_code + if args.command == "proof-pack": + try: + proof_pack = build_case_proof_pack(args.case_dir) + output_path = args.output or (args.case_dir / "proof_pack.json") + output_path.parent.mkdir(parents=True, exist_ok=True) + write_case_proof_pack(proof_pack, output_path) + manifest: dict[str, Any] | None = None + if args.visual_review_manifest is not None: + manifest = build_static_visual_review_manifest( + proof_pack, + task_intent=args.task_intent, + ) + validate_visual_review_manifest(manifest, current_root=args.case_dir) + args.visual_review_manifest.parent.mkdir(parents=True, exist_ok=True) + with args.visual_review_manifest.open("w") as f: + json.dump(manifest, f, indent=2) + except (FileNotFoundError, ValueError) as exc: + print(f"Error: {exc}", file=sys.stderr) + return 1 + + if args.output_format == "json": + payload: dict[str, Any] = { + "proof_pack_path": str(output_path), + "proof_pack": proof_pack.to_dict(), + } + if args.visual_review_manifest is not None: + payload["visual_review_manifest_path"] = str(args.visual_review_manifest) + payload["visual_review_manifest"] = manifest + print(json.dumps(payload, indent=2)) + else: + print(f"Proof pack written to {output_path}") + print(f" case_id={proof_pack.case_id}") + print(f" verdict={proof_pack.verdict}") + print(f" selected_phase={proof_pack.selected_phase}") + print(f" renderer_evidence={len(proof_pack.renderer_evidence)}") + if args.visual_review_manifest is not None: + print(f"Visual review manifest written to {args.visual_review_manifest}") + return 0 + + if args.command == "visual-review-summary": + try: + manifest = _load_json(args.manifest) + record = _load_json(args.record) + output_path = args.output or (args.record.parent / "visual_review_summary.json") + output_path.parent.mkdir(parents=True, exist_ok=True) + summary_path = write_visual_review_summary( + manifest, + record, + output_path, + manifest_path=str(args.manifest), + record_path=str(args.record), + ) + summary = _load_json(summary_path) + except FileNotFoundError as exc: + print(f"Error: {exc}", file=sys.stderr) + return 1 + + if args.output_format == "json": + print(json.dumps(summary, indent=2)) + else: + print(f"Visual review summary written to {summary_path}") + print(f" case_id={summary['case_id']}") + print(f" effective_visual_verdict={summary['effective_visual_verdict']}") + print(f" is_valid={summary['is_valid']}") + return 0 + if args.command == "contract": if args.contract_command is None: contract_parser.print_help() diff --git a/src/roboharness/evidence/__init__.py b/src/roboharness/evidence/__init__.py new file mode 100644 index 0000000..8ee04b7 --- /dev/null +++ b/src/roboharness/evidence/__init__.py @@ -0,0 +1,85 @@ +"""Evidence artifact models for post-hoc robot validation runs.""" + +from roboharness.evidence.artifacts import ( + AUTONOMOUS_EVIDENCE_REPORT_SCHEMA_VERSION, + SEMANTIC_SNAPSHOT_BUNDLE_SCHEMA_VERSION, + AutonomousEvidenceReport, + RenderedImage, + RendererReport, + RendererSnapshot, + SemanticSnapshot, + SemanticSnapshotBundle, + load_autonomous_evidence_report, + load_renderer_report, + load_semantic_snapshot_bundle, + write_autonomous_evidence_report, + write_renderer_report, + write_semantic_snapshot_bundle, +) +from roboharness.evidence.proof_pack import ( + CASE_PROOF_PACK_SCHEMA_VERSION, + DEFAULT_STATIC_REVIEW_VIEWS, + STATIC_VISUAL_DIMENSIONS, + SUITE_PROOF_PACK_SCHEMA_VERSION, + VISUAL_REVIEW_QUEUE_SCHEMA_VERSION, + CaseProofPack, + ProofPackArtifact, + ProofPackImageRef, + SuiteProofPack, + SuiteProofPackCase, + VisualReviewQueue, + VisualReviewQueueItem, + build_case_proof_pack, + build_paired_visual_review_manifest, + build_static_visual_review_manifest, + build_suite_proof_pack, + build_visual_review_queue, + load_case_proof_pack, + load_suite_proof_pack, + write_case_proof_pack, + write_paired_visual_review_manifest, + write_static_visual_review_manifest, + write_suite_proof_pack, + write_visual_review_queue, +) + +__all__ = [ + "AUTONOMOUS_EVIDENCE_REPORT_SCHEMA_VERSION", + "CASE_PROOF_PACK_SCHEMA_VERSION", + "DEFAULT_STATIC_REVIEW_VIEWS", + "SEMANTIC_SNAPSHOT_BUNDLE_SCHEMA_VERSION", + "STATIC_VISUAL_DIMENSIONS", + "SUITE_PROOF_PACK_SCHEMA_VERSION", + "VISUAL_REVIEW_QUEUE_SCHEMA_VERSION", + "AutonomousEvidenceReport", + "CaseProofPack", + "ProofPackArtifact", + "ProofPackImageRef", + "RenderedImage", + "RendererReport", + "RendererSnapshot", + "SemanticSnapshot", + "SemanticSnapshotBundle", + "SuiteProofPack", + "SuiteProofPackCase", + "VisualReviewQueue", + "VisualReviewQueueItem", + "build_case_proof_pack", + "build_paired_visual_review_manifest", + "build_static_visual_review_manifest", + "build_suite_proof_pack", + "build_visual_review_queue", + "load_autonomous_evidence_report", + "load_case_proof_pack", + "load_renderer_report", + "load_semantic_snapshot_bundle", + "load_suite_proof_pack", + "write_autonomous_evidence_report", + "write_case_proof_pack", + "write_paired_visual_review_manifest", + "write_renderer_report", + "write_semantic_snapshot_bundle", + "write_static_visual_review_manifest", + "write_suite_proof_pack", + "write_visual_review_queue", +] diff --git a/src/roboharness/evidence/artifacts.py b/src/roboharness/evidence/artifacts.py new file mode 100644 index 0000000..eb0d6e4 --- /dev/null +++ b/src/roboharness/evidence/artifacts.py @@ -0,0 +1,342 @@ +"""Typed evidence artifacts shared by downstream robot harnesses.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from roboharness._utils import load_json, save_json + +SEMANTIC_SNAPSHOT_BUNDLE_SCHEMA_VERSION = "roboharness_semantic_snapshot_bundle/v1" +AUTONOMOUS_EVIDENCE_REPORT_SCHEMA_VERSION = "roboharness_autonomous_evidence_report/v1" + + +def _payload_without(data: dict[str, Any], keys: set[str]) -> dict[str, Any]: + return {key: value for key, value in data.items() if key not in keys} + + +def _put_if_not_none(payload: dict[str, Any], key: str, value: Any) -> None: + if value is not None: + payload[key] = value + + +@dataclass(frozen=True) +class SemanticSnapshot: + """One named semantic phase sample emitted by an evidence producer.""" + + name: str + state: dict[str, Any] = field(default_factory=dict) + metrics: dict[str, Any] = field(default_factory=dict) + metadata: dict[str, Any] = field(default_factory=dict) + extra: dict[str, Any] = field(default_factory=dict) + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> SemanticSnapshot: + reserved = {"name", "state", "metrics", "metadata"} + return cls( + name=str(data["name"]), + state=dict(data.get("state") or {}), + metrics=dict(data.get("metrics") or {}), + metadata=dict(data.get("metadata") or {}), + extra=_payload_without(data, reserved), + ) + + def to_dict(self) -> dict[str, Any]: + payload = dict(self.extra) + payload["name"] = self.name + if self.state: + payload["state"] = dict(self.state) + if self.metadata: + payload["metadata"] = dict(self.metadata) + payload["metrics"] = dict(self.metrics) + return payload + + +@dataclass(frozen=True) +class SemanticSnapshotBundle: + """Ordered semantic snapshots plus run metadata for replay or review.""" + + snapshots: tuple[SemanticSnapshot, ...] + snapshot_order: tuple[str, ...] = () + metadata: dict[str, Any] = field(default_factory=dict) + schema_version: str | int = SEMANTIC_SNAPSHOT_BUNDLE_SCHEMA_VERSION + extra: dict[str, Any] = field(default_factory=dict) + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> SemanticSnapshotBundle: + snapshots = tuple(SemanticSnapshot.from_dict(item) for item in data.get("snapshots", ())) + order = tuple(str(name) for name in data.get("snapshot_order") or ()) + if not order: + order = tuple(snapshot.name for snapshot in snapshots) + reserved = {"schema_version", "snapshot_order", "snapshots", "metadata"} + return cls( + snapshots=snapshots, + snapshot_order=order, + metadata=dict(data.get("metadata") or {}), + schema_version=data.get("schema_version", SEMANTIC_SNAPSHOT_BUNDLE_SCHEMA_VERSION), + extra=_payload_without(data, reserved), + ) + + def to_dict(self) -> dict[str, Any]: + payload = dict(self.extra) + payload["schema_version"] = self.schema_version + payload["snapshot_order"] = list( + self.snapshot_order or (snapshot.name for snapshot in self.snapshots) + ) + payload["snapshots"] = [snapshot.to_dict() for snapshot in self.snapshots] + if self.metadata: + payload["metadata"] = dict(self.metadata) + return payload + + def write_json(self, path: str | Path) -> Path: + return write_semantic_snapshot_bundle(self, path) + + +@dataclass(frozen=True) +class RenderedImage: + """Single rendered image reference inside a renderer report.""" + + path: str + camera: str | None = None + view: str | None = None + metadata: dict[str, Any] = field(default_factory=dict) + extra: dict[str, Any] = field(default_factory=dict) + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> RenderedImage: + reserved = {"path", "camera", "view", "metadata"} + return cls( + path=str(data["path"]), + camera=None if data.get("camera") is None else str(data["camera"]), + view=None if data.get("view") is None else str(data["view"]), + metadata=dict(data.get("metadata") or {}), + extra=_payload_without(data, reserved), + ) + + def to_dict(self) -> dict[str, Any]: + payload = dict(self.extra) + _put_if_not_none(payload, "camera", self.camera) + _put_if_not_none(payload, "view", self.view) + payload["path"] = self.path + if self.metadata: + payload["metadata"] = dict(self.metadata) + return payload + + +@dataclass(frozen=True) +class RendererSnapshot: + """Renderer evidence for one semantic snapshot.""" + + name: str + images: tuple[RenderedImage, ...] = () + metrics: dict[str, Any] = field(default_factory=dict) + capture_ok: bool | None = None + motion_ok: bool | None = None + metadata: dict[str, Any] = field(default_factory=dict) + extra: dict[str, Any] = field(default_factory=dict) + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> RendererSnapshot: + reserved = {"name", "images", "metrics", "capture_ok", "motion_ok", "metadata"} + return cls( + name=str(data["name"]), + images=tuple(RenderedImage.from_dict(item) for item in data.get("images", ())), + metrics=dict(data.get("metrics") or {}), + capture_ok=data.get("capture_ok"), + motion_ok=data.get("motion_ok"), + metadata=dict(data.get("metadata") or {}), + extra=_payload_without(data, reserved), + ) + + def to_dict(self) -> dict[str, Any]: + payload = dict(self.extra) + payload["name"] = self.name + _put_if_not_none(payload, "capture_ok", self.capture_ok) + _put_if_not_none(payload, "motion_ok", self.motion_ok) + payload["metrics"] = dict(self.metrics) + payload["images"] = [image.to_dict() for image in self.images] + if self.metadata: + payload["metadata"] = dict(self.metadata) + return payload + + +@dataclass(frozen=True) +class RendererReport: + """Structured output from rendering a semantic snapshot bundle.""" + + output_dir: str + renderer: str + capture_ok: bool | None = None + motion_ok: bool | None = None + snapshots: tuple[RendererSnapshot, ...] = () + flags: tuple[str, ...] = () + trustworthiness_flags: tuple[dict[str, Any], ...] = () + metadata: dict[str, Any] = field(default_factory=dict) + schema_version: str | None = None + extra: dict[str, Any] = field(default_factory=dict) + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> RendererReport: + reserved = { + "schema_version", + "output_dir", + "renderer", + "capture_ok", + "motion_ok", + "snapshots", + "flags", + "trustworthiness_flags", + "metadata", + } + return cls( + output_dir=str(data["output_dir"]), + renderer=str(data.get("renderer") or "unknown"), + capture_ok=data.get("capture_ok"), + motion_ok=data.get("motion_ok"), + snapshots=tuple(RendererSnapshot.from_dict(item) for item in data.get("snapshots", ())), + flags=tuple(str(flag) for flag in data.get("flags", ())), + trustworthiness_flags=tuple( + dict(flag) for flag in data.get("trustworthiness_flags", ()) + ), + metadata=dict(data.get("metadata") or {}), + schema_version=data.get("schema_version"), + extra=_payload_without(data, reserved), + ) + + def to_dict(self) -> dict[str, Any]: + payload = dict(self.extra) + _put_if_not_none(payload, "schema_version", self.schema_version) + payload["output_dir"] = self.output_dir + payload["renderer"] = self.renderer + _put_if_not_none(payload, "capture_ok", self.capture_ok) + _put_if_not_none(payload, "motion_ok", self.motion_ok) + payload["flags"] = list(self.flags) + payload["trustworthiness_flags"] = [dict(flag) for flag in self.trustworthiness_flags] + payload["metadata"] = dict(self.metadata) + payload["snapshots"] = [snapshot.to_dict() for snapshot in self.snapshots] + return payload + + def write_json(self, path: str | Path) -> Path: + return write_renderer_report(self, path) + + +@dataclass(frozen=True) +class AutonomousEvidenceReport: + """Machine-readable run report before approval or human escalation.""" + + case_id: str + output_dir: str + verdict: str + verdict_reasons: tuple[str, ...] = () + summary_metrics: dict[str, Any] = field(default_factory=dict) + snapshot_metrics: dict[str, dict[str, Any]] = field(default_factory=dict) + renderer_reports: dict[str, RendererReport] = field(default_factory=dict) + failure_taxonomy: tuple[dict[str, Any], ...] = () + runtime: dict[str, Any] = field(default_factory=dict) + plan: dict[str, Any] = field(default_factory=dict) + metadata: dict[str, Any] = field(default_factory=dict) + schema_version: str | None = None + extra: dict[str, Any] = field(default_factory=dict) + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> AutonomousEvidenceReport: + reserved = { + "schema_version", + "case_id", + "output_dir", + "verdict", + "verdict_reasons", + "summary_metrics", + "snapshot_metrics", + "renderer_reports", + "failure_taxonomy", + "runtime", + "plan", + "metadata", + } + renderer_reports = { + str(name): RendererReport.from_dict(report) + for name, report in dict(data.get("renderer_reports") or {}).items() + } + return cls( + case_id=str(data["case_id"]), + output_dir=str(data.get("output_dir") or ""), + verdict=str(data.get("verdict") or ""), + verdict_reasons=tuple(str(reason) for reason in data.get("verdict_reasons", ())), + summary_metrics=dict(data.get("summary_metrics") or {}), + snapshot_metrics={ + str(name): dict(metrics) + for name, metrics in dict(data.get("snapshot_metrics") or {}).items() + }, + renderer_reports=renderer_reports, + failure_taxonomy=tuple(dict(item) for item in data.get("failure_taxonomy", ())), + runtime=dict(data.get("runtime") or {}), + plan=dict(data.get("plan") or {}), + metadata=dict(data.get("metadata") or {}), + schema_version=data.get("schema_version"), + extra=_payload_without(data, reserved), + ) + + def to_dict(self) -> dict[str, Any]: + payload = dict(self.extra) + _put_if_not_none(payload, "schema_version", self.schema_version) + payload["case_id"] = self.case_id + payload["output_dir"] = self.output_dir + payload["verdict"] = self.verdict + payload["verdict_reasons"] = list(self.verdict_reasons) + payload["failure_taxonomy"] = [dict(item) for item in self.failure_taxonomy] + payload["runtime"] = dict(self.runtime) + payload["plan"] = dict(self.plan) + if self.metadata: + payload["metadata"] = dict(self.metadata) + payload["summary_metrics"] = dict(self.summary_metrics) + payload["snapshot_metrics"] = { + name: dict(metrics) for name, metrics in self.snapshot_metrics.items() + } + payload["renderer_reports"] = { + name: report.to_dict() for name, report in self.renderer_reports.items() + } + return payload + + def write_json(self, path: str | Path) -> Path: + return write_autonomous_evidence_report(self, path) + + +def load_semantic_snapshot_bundle(path: str | Path) -> SemanticSnapshotBundle: + """Load a semantic snapshot bundle from JSON.""" + return SemanticSnapshotBundle.from_dict(load_json(Path(path))) + + +def write_semantic_snapshot_bundle(bundle: SemanticSnapshotBundle, path: str | Path) -> Path: + """Write a semantic snapshot bundle to JSON.""" + output_path = Path(path) + save_json(bundle.to_dict(), output_path) + return output_path + + +def load_renderer_report(path: str | Path) -> RendererReport: + """Load a renderer report from JSON.""" + return RendererReport.from_dict(load_json(Path(path))) + + +def write_renderer_report(report: RendererReport, path: str | Path) -> Path: + """Write a renderer report to JSON.""" + output_path = Path(path) + save_json(report.to_dict(), output_path) + return output_path + + +def load_autonomous_evidence_report(path: str | Path) -> AutonomousEvidenceReport: + """Load an autonomous evidence report from JSON.""" + return AutonomousEvidenceReport.from_dict(load_json(Path(path))) + + +def write_autonomous_evidence_report( + report: AutonomousEvidenceReport, + path: str | Path, +) -> Path: + """Write an autonomous evidence report to JSON.""" + output_path = Path(path) + save_json(report.to_dict(), output_path) + return output_path diff --git a/src/roboharness/evidence/proof_pack.py b/src/roboharness/evidence/proof_pack.py new file mode 100644 index 0000000..55e1a54 --- /dev/null +++ b/src/roboharness/evidence/proof_pack.py @@ -0,0 +1,981 @@ +"""Proof-pack assembly for downstream visual harness evidence.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from roboharness._utils import load_json, save_json +from roboharness.approval.visual_review import MANIFEST_SCHEMA_VERSION +from roboharness.evidence.artifacts import ( + AutonomousEvidenceReport, + RenderedImage, + RendererReport, + SemanticSnapshotBundle, + load_autonomous_evidence_report, + load_renderer_report, + load_semantic_snapshot_bundle, +) + +CASE_PROOF_PACK_SCHEMA_VERSION = "roboharness_case_proof_pack/v1" +SUITE_PROOF_PACK_SCHEMA_VERSION = "roboharness_suite_proof_pack/v1" +VISUAL_REVIEW_QUEUE_SCHEMA_VERSION = "roboharness_visual_review_queue/v1" +STATIC_VISUAL_DIMENSIONS = ( + "robot_posture", + "hand_pose", + "object_relative_position", + "obvious_collision_or_penetration", + "task_success_visual_check", +) +DEFAULT_STATIC_REVIEW_VIEWS = ("front2back", "left2right", "top2down") +DEFAULT_METRIC_SUMMARY_KEYS = ( + "semantic_visual_ok", + "workspace_framing_ok", + "snapshot_state_progress_ok", + "final_snapshot_name", + "grasp_accuracy_snapshot_name", + "holding_snapshot_name", + "grip_center_error_mm", + "holding_grip_center_error_mm", + "pinch_gap_error_mm", + "pinch_elevation_deg", + "index_middle_vertical_deg", + "pelvis_roll_deg", + "pelvis_pitch_deg", + "pelvis_height_m", + "non_primary_arm_drift_deg", + "cmd_vs_actual_max_delta_deg", + "waypoint_wrist_gap_mm", + "waypoint_first_segment_mm", + "planning_success", + "pregrasp_converged", + "approach_converged", + "lift_converged", + "holding_reached", + "control_backend", + "runtime_surface", + "render_mujoco_enabled", + "render_meshcat_capture_s", + "render_mujoco_capture_s", + "render_live_total_s", + "render_total_s", + "whole_case_wall_time_s", +) + + +@dataclass(frozen=True) +class ProofPackArtifact: + """A case-local artifact consumed by a proof pack.""" + + id: str + path: str + kind: str + + def to_dict(self) -> dict[str, Any]: + return {"id": self.id, "path": self.path, "kind": self.kind} + + +@dataclass(frozen=True) +class ProofPackImageRef: + """One case-local rendered image reference selected for review.""" + + renderer: str + phase: str + view: str + path: str + metadata: dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> dict[str, Any]: + payload: dict[str, Any] = { + "renderer": self.renderer, + "phase": self.phase, + "view": self.view, + "path": self.path, + } + if self.metadata: + payload["metadata"] = dict(self.metadata) + return payload + + +@dataclass(frozen=True) +class CaseProofPack: + """Compact evidence bundle prepared from one downstream visual harness case.""" + + case_id: str + output_dir: str + verdict: str + verdict_reasons: tuple[str, ...] + failure_taxonomy: tuple[dict[str, Any], ...] + snapshot_order: tuple[str, ...] + selected_phase: str + metric_summary: dict[str, Any] + renderer_evidence: tuple[ProofPackImageRef, ...] + artifacts: tuple[ProofPackArtifact, ...] + review_mode: str = "current_only" + schema_version: str = CASE_PROOF_PACK_SCHEMA_VERSION + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> CaseProofPack: + return cls( + case_id=str(data["case_id"]), + output_dir=str(data.get("output_dir") or ""), + verdict=str(data.get("verdict") or ""), + verdict_reasons=tuple(str(reason) for reason in data.get("verdict_reasons", ())), + failure_taxonomy=tuple(dict(item) for item in data.get("failure_taxonomy", ())), + snapshot_order=tuple(str(name) for name in data.get("snapshot_order", ())), + selected_phase=str(data.get("selected_phase") or ""), + metric_summary=dict(data.get("metric_summary") or {}), + renderer_evidence=tuple( + ProofPackImageRef( + renderer=str(item["renderer"]), + phase=str(item["phase"]), + view=str(item["view"]), + path=str(item["path"]), + metadata=dict(item.get("metadata") or {}), + ) + for item in data.get("renderer_evidence", ()) + ), + artifacts=tuple( + ProofPackArtifact( + id=str(item["id"]), + path=str(item["path"]), + kind=str(item["kind"]), + ) + for item in data.get("artifacts", ()) + ), + review_mode=str(data.get("review_mode") or "current_only"), + schema_version=str(data.get("schema_version") or CASE_PROOF_PACK_SCHEMA_VERSION), + ) + + def to_dict(self) -> dict[str, Any]: + return { + "schema_version": self.schema_version, + "case_id": self.case_id, + "output_dir": self.output_dir, + "verdict": self.verdict, + "verdict_reasons": list(self.verdict_reasons), + "failure_taxonomy": [dict(item) for item in self.failure_taxonomy], + "snapshot_order": list(self.snapshot_order), + "selected_phase": self.selected_phase, + "metric_summary": dict(self.metric_summary), + "renderer_evidence": [item.to_dict() for item in self.renderer_evidence], + "artifacts": [item.to_dict() for item in self.artifacts], + "review_mode": self.review_mode, + } + + def write_json(self, path: str | Path) -> Path: + return write_case_proof_pack(self, path) + + +@dataclass(frozen=True) +class SuiteProofPackCase: + """One case entry inside a suite proof pack.""" + + case_id: str + case_dir: str + status: str + proof_pack_path: str | None + visual_review_manifest_path: str | None + selected_phase: str | None = None + verdict: str | None = None + renderer_evidence_count: int = 0 + error: str | None = None + + def to_dict(self) -> dict[str, Any]: + payload: dict[str, Any] = { + "case_id": self.case_id, + "case_dir": self.case_dir, + "status": self.status, + "proof_pack_path": self.proof_pack_path, + "visual_review_manifest_path": self.visual_review_manifest_path, + "renderer_evidence_count": int(self.renderer_evidence_count), + } + if self.selected_phase is not None: + payload["selected_phase"] = self.selected_phase + if self.verdict is not None: + payload["verdict"] = self.verdict + if self.error is not None: + payload["error"] = self.error + return payload + + +@dataclass(frozen=True) +class SuiteProofPack: + """Suite-level index over case proof packs prepared for visual review.""" + + suite_name: str + suite_dir: str + suite_report_path: str + cases: tuple[SuiteProofPackCase, ...] + schema_version: str = SUITE_PROOF_PACK_SCHEMA_VERSION + + @property + def reviewable_count(self) -> int: + return sum(1 for case in self.cases if case.status == "reviewable") + + @property + def skipped_count(self) -> int: + return sum(1 for case in self.cases if case.status != "reviewable") + + def to_dict(self) -> dict[str, Any]: + return { + "schema_version": self.schema_version, + "suite_name": self.suite_name, + "suite_dir": self.suite_dir, + "suite_report_path": self.suite_report_path, + "total_cases": len(self.cases), + "reviewable_count": self.reviewable_count, + "skipped_count": self.skipped_count, + "cases": [case.to_dict() for case in self.cases], + } + + def write_json(self, path: str | Path) -> Path: + return write_suite_proof_pack(self, path) + + +@dataclass(frozen=True) +class VisualReviewQueueItem: + """One bounded visual review item selected from a suite proof pack.""" + + case_id: str + case_dir: str + visual_review_manifest_path: str + proof_pack_path: str + selected_phase: str + verdict: str + + def to_dict(self) -> dict[str, Any]: + return { + "case_id": self.case_id, + "case_dir": self.case_dir, + "visual_review_manifest_path": self.visual_review_manifest_path, + "proof_pack_path": self.proof_pack_path, + "selected_phase": self.selected_phase, + "verdict": self.verdict, + } + + +@dataclass(frozen=True) +class VisualReviewQueue: + """Review queue derived from a suite proof pack.""" + + suite_name: str + suite_dir: str + items: tuple[VisualReviewQueueItem, ...] + schema_version: str = VISUAL_REVIEW_QUEUE_SCHEMA_VERSION + + def to_dict(self) -> dict[str, Any]: + return { + "schema_version": self.schema_version, + "suite_name": self.suite_name, + "suite_dir": self.suite_dir, + "total_items": len(self.items), + "items": [item.to_dict() for item in self.items], + } + + def write_json(self, path: str | Path) -> Path: + return write_visual_review_queue(self, path) + + +def build_case_proof_pack( + case_dir: str | Path, + *, + preferred_renderers: tuple[str, ...] = ("mujoco", "meshcat"), + preferred_views: tuple[str, ...] = DEFAULT_STATIC_REVIEW_VIEWS, + metric_keys: tuple[str, ...] = DEFAULT_METRIC_SUMMARY_KEYS, +) -> CaseProofPack: + """Build a case-level proof pack from a downstream visual harness case directory.""" + + root = Path(case_dir) + report_path = root / "autonomous_report.json" + if not report_path.exists(): + raise FileNotFoundError(f"autonomous report not found: {report_path}") + report = load_autonomous_evidence_report(report_path) + + bundle_path = root / "snapshot_bundle.json" + bundle = load_semantic_snapshot_bundle(bundle_path) if bundle_path.exists() else None + renderer_reports = _load_case_renderer_reports(root, report) + snapshot_order = _snapshot_order(report, bundle) + selected_phase = _select_review_phase(report, snapshot_order) + metric_summary = _metric_summary(report, metric_keys) + renderer_evidence = _select_renderer_evidence( + root, + renderer_reports, + selected_phase=selected_phase, + preferred_renderers=preferred_renderers, + preferred_views=preferred_views, + ) + artifacts = _artifact_refs( + root, + report, + bundle_path=bundle_path, + renderer_reports=renderer_reports, + ) + + return CaseProofPack( + case_id=report.case_id, + output_dir=_case_relative_path(root, Path(report.output_dir)) or root.as_posix(), + verdict=report.verdict, + verdict_reasons=report.verdict_reasons, + failure_taxonomy=report.failure_taxonomy, + snapshot_order=snapshot_order, + selected_phase=selected_phase, + metric_summary=metric_summary, + renderer_evidence=renderer_evidence, + artifacts=artifacts, + ) + + +def build_suite_proof_pack( + suite_report_path: str | Path, + *, + write_missing_case_artifacts: bool = True, + task_intent: str | None = None, +) -> SuiteProofPack: + """Build a suite-level index over case proof packs and visual manifests.""" + + report_path = Path(suite_report_path) + suite_dir = report_path.parent + suite_report = load_json(report_path) + suite_name = str( + suite_report.get("suite_name") or report_path.stem.removeprefix("suite_report_") + ) + cases: list[SuiteProofPackCase] = [] + for result in _suite_results(suite_report): + case_id = str(result.get("case_id") or "") + case_dir = _suite_case_dir(suite_dir, result) + proof_pack_path = case_dir / "proof_pack.json" + manifest_path = case_dir / "visual_review_manifest.json" + if not case_id: + cases.append( + SuiteProofPackCase( + case_id="", + case_dir=_suite_relative_path(suite_dir, case_dir) or case_dir.as_posix(), + status="skipped", + proof_pack_path=None, + visual_review_manifest_path=None, + error="suite result is missing case_id", + ) + ) + continue + if not case_dir.exists(): + cases.append( + SuiteProofPackCase( + case_id=case_id, + case_dir=_suite_relative_path(suite_dir, case_dir) or case_dir.as_posix(), + status="skipped", + proof_pack_path=None, + visual_review_manifest_path=None, + error="case directory does not exist", + ) + ) + continue + if write_missing_case_artifacts and not proof_pack_path.exists(): + try: + proof_pack = build_case_proof_pack(case_dir) + write_case_proof_pack(proof_pack, proof_pack_path) + write_static_visual_review_manifest( + proof_pack, + manifest_path, + task_intent=task_intent or _default_task_intent(proof_pack), + ) + except Exception as exc: # pragma: no cover - exact failures are data dependent + cases.append( + SuiteProofPackCase( + case_id=case_id, + case_dir=_suite_relative_path(suite_dir, case_dir) or case_dir.as_posix(), + status="skipped", + proof_pack_path=None, + visual_review_manifest_path=None, + error=f"{type(exc).__name__}: {exc}", + ) + ) + continue + if not proof_pack_path.exists() or not manifest_path.exists(): + cases.append( + SuiteProofPackCase( + case_id=case_id, + case_dir=_suite_relative_path(suite_dir, case_dir) or case_dir.as_posix(), + status="skipped", + proof_pack_path=( + _suite_relative_path(suite_dir, proof_pack_path) + if proof_pack_path.exists() + else None + ), + visual_review_manifest_path=( + _suite_relative_path(suite_dir, manifest_path) + if manifest_path.exists() + else None + ), + error="case proof pack or visual review manifest is missing", + ) + ) + continue + + proof_pack = load_case_proof_pack(proof_pack_path) + cases.append( + SuiteProofPackCase( + case_id=case_id, + case_dir=_suite_relative_path(suite_dir, case_dir) or case_dir.as_posix(), + status="reviewable", + proof_pack_path=_suite_relative_path(suite_dir, proof_pack_path), + visual_review_manifest_path=_suite_relative_path(suite_dir, manifest_path), + selected_phase=proof_pack.selected_phase, + verdict=proof_pack.verdict, + renderer_evidence_count=len(proof_pack.renderer_evidence), + ) + ) + return SuiteProofPack( + suite_name=suite_name, + suite_dir=suite_dir.as_posix(), + suite_report_path=_suite_relative_path(suite_dir, report_path) or report_path.name, + cases=tuple(cases), + ) + + +def write_suite_proof_pack(suite_proof_pack: SuiteProofPack, path: str | Path) -> Path: + """Write a suite proof pack to JSON.""" + + output_path = Path(path) + save_json(suite_proof_pack.to_dict(), output_path) + return output_path + + +def load_suite_proof_pack(path: str | Path) -> SuiteProofPack: + """Load a suite proof pack from JSON.""" + + data = load_json(Path(path)) + return SuiteProofPack( + suite_name=str(data["suite_name"]), + suite_dir=str(data.get("suite_dir") or ""), + suite_report_path=str(data.get("suite_report_path") or ""), + cases=tuple( + SuiteProofPackCase( + case_id=str(item.get("case_id") or ""), + case_dir=str(item.get("case_dir") or ""), + status=str(item.get("status") or ""), + proof_pack_path=( + None + if item.get("proof_pack_path") is None + else str(item.get("proof_pack_path")) + ), + visual_review_manifest_path=( + None + if item.get("visual_review_manifest_path") is None + else str(item.get("visual_review_manifest_path")) + ), + selected_phase=( + None if item.get("selected_phase") is None else str(item.get("selected_phase")) + ), + verdict=None if item.get("verdict") is None else str(item.get("verdict")), + renderer_evidence_count=int(item.get("renderer_evidence_count") or 0), + error=None if item.get("error") is None else str(item.get("error")), + ) + for item in data.get("cases", ()) + ), + schema_version=str(data.get("schema_version") or SUITE_PROOF_PACK_SCHEMA_VERSION), + ) + + +def build_visual_review_queue(suite_proof_pack: SuiteProofPack) -> VisualReviewQueue: + """Build a review queue from reviewable case entries in a suite proof pack.""" + + items: list[VisualReviewQueueItem] = [] + for case in suite_proof_pack.cases: + if ( + case.status != "reviewable" + or case.proof_pack_path is None + or case.visual_review_manifest_path is None + or case.selected_phase is None + or case.verdict is None + ): + continue + items.append( + VisualReviewQueueItem( + case_id=case.case_id, + case_dir=case.case_dir, + visual_review_manifest_path=case.visual_review_manifest_path, + proof_pack_path=case.proof_pack_path, + selected_phase=case.selected_phase, + verdict=case.verdict, + ) + ) + return VisualReviewQueue( + suite_name=suite_proof_pack.suite_name, + suite_dir=suite_proof_pack.suite_dir, + items=tuple(items), + ) + + +def write_visual_review_queue(queue: VisualReviewQueue, path: str | Path) -> Path: + """Write a suite visual review queue to JSON.""" + + output_path = Path(path) + save_json(queue.to_dict(), output_path) + return output_path + + +def write_case_proof_pack(proof_pack: CaseProofPack, path: str | Path) -> Path: + """Write a case proof pack to JSON.""" + + output_path = Path(path) + save_json(proof_pack.to_dict(), output_path) + return output_path + + +def load_case_proof_pack(path: str | Path) -> CaseProofPack: + """Load a case proof pack from JSON.""" + + from roboharness._utils import load_json + + return CaseProofPack.from_dict(load_json(Path(path))) + + +def build_static_visual_review_manifest( + proof_pack: CaseProofPack, + *, + task_intent: str, + dimensions: tuple[str, ...] = STATIC_VISUAL_DIMENSIONS, + preferred_renderer: str | None = None, + required: bool = True, +) -> dict[str, Any]: + """Build a current-only static keyframe visual review manifest from a proof pack.""" + + evidence = _manifest_evidence(proof_pack, preferred_renderer=preferred_renderer) + if not evidence: + raise ValueError(f"proof pack {proof_pack.case_id!r} has no renderer evidence") + views = _dedupe(ref.view for ref in evidence) + current_paths = _dedupe(ref.path for ref in evidence) + metric_fallback = tuple(proof_pack.metric_summary) + return { + "schema_version": MANIFEST_SCHEMA_VERSION, + "case_id": proof_pack.case_id, + "mode": "current_only", + "task_intent": task_intent, + "dimensions": [ + { + "id": dimension, + "required": required, + "phase": proof_pack.selected_phase, + "evidence_type": "current_static_keyframe", + "views": list(views), + "current": list(current_paths), + "metric_fallback": list(metric_fallback), + "why_not_metricized": ( + "Static rendered keyframes catch posture, contact, and task-agreement " + "issues that scalar metrics do not fully encode." + ), + } + for dimension in dimensions + ], + "metric_summary": dict(proof_pack.metric_summary), + "review_policy": { + "requires_paired_evidence": False, + "allow_automatic_visual_pass": False, + "human_escalation_reasons": ["current_only_review_cannot_auto_pass"], + }, + "proof_pack": { + "schema_version": proof_pack.schema_version, + "selected_phase": proof_pack.selected_phase, + "renderer_evidence": [ref.to_dict() for ref in evidence], + "artifacts": [artifact.to_dict() for artifact in proof_pack.artifacts], + }, + } + + +def build_paired_visual_review_manifest( + current_proof_pack: CaseProofPack, + baseline_proof_pack: CaseProofPack, + *, + task_intent: str, + dimensions: tuple[str, ...] = STATIC_VISUAL_DIMENSIONS, + preferred_renderer: str | None = None, + required: bool = True, + mode: str = "regression", +) -> dict[str, Any]: + """Build an explicit current-vs-baseline static visual review manifest.""" + + if mode not in {"regression", "migration"}: + raise ValueError("paired visual review mode must be 'regression' or 'migration'") + if current_proof_pack.case_id != baseline_proof_pack.case_id: + raise ValueError( + "paired visual review requires matching case_id values, got " + f"{current_proof_pack.case_id!r} and {baseline_proof_pack.case_id!r}" + ) + current_evidence = _manifest_evidence( + current_proof_pack, + preferred_renderer=preferred_renderer, + ) + baseline_evidence = _manifest_evidence( + baseline_proof_pack, + preferred_renderer=preferred_renderer, + ) + if not current_evidence: + raise ValueError( + f"current proof pack {current_proof_pack.case_id!r} has no renderer evidence" + ) + if not baseline_evidence: + raise ValueError( + f"baseline proof pack {baseline_proof_pack.case_id!r} has no renderer evidence" + ) + pairs = _paired_manifest_evidence(current_evidence, baseline_evidence) + if not pairs: + raise ValueError(f"no paired renderer evidence for case {current_proof_pack.case_id!r}") + views = _dedupe(pair[0].view for pair in pairs) + current_paths = _dedupe(pair[0].path for pair in pairs) + baseline_paths = _dedupe(pair[1].path for pair in pairs) + metric_fallback = tuple(current_proof_pack.metric_summary) + human_reasons: list[str] = [] + allow_automatic_visual_pass = mode == "regression" + if mode == "migration": + human_reasons.append("baseline_blessing_required") + return { + "schema_version": MANIFEST_SCHEMA_VERSION, + "case_id": current_proof_pack.case_id, + "mode": mode, + "task_intent": task_intent, + "dimensions": [ + { + "id": dimension, + "required": required, + "phase": current_proof_pack.selected_phase, + "evidence_type": "paired_keyframe", + "views": list(views), + "current": list(current_paths), + "baseline": list(baseline_paths), + "metric_fallback": list(metric_fallback), + "why_not_metricized": ( + "Paired static keyframes compare posture, contact, and " + "task-agreement evidence against an explicit baseline pack." + ), + } + for dimension in dimensions + ], + "metric_summary": dict(current_proof_pack.metric_summary), + "review_policy": { + "requires_paired_evidence": True, + "allow_automatic_visual_pass": allow_automatic_visual_pass, + "human_escalation_reasons": human_reasons, + }, + "proof_pack": { + "schema_version": current_proof_pack.schema_version, + "selected_phase": current_proof_pack.selected_phase, + "baseline_selected_phase": baseline_proof_pack.selected_phase, + "renderer_evidence": [current.to_dict() for current, _ in pairs], + "baseline_renderer_evidence": [baseline.to_dict() for _, baseline in pairs], + "artifacts": [artifact.to_dict() for artifact in current_proof_pack.artifacts], + "baseline_artifacts": [ + artifact.to_dict() for artifact in baseline_proof_pack.artifacts + ], + }, + } + + +def write_paired_visual_review_manifest( + current_proof_pack: CaseProofPack, + baseline_proof_pack: CaseProofPack, + path: str | Path, + *, + task_intent: str, + dimensions: tuple[str, ...] = STATIC_VISUAL_DIMENSIONS, + preferred_renderer: str | None = None, + required: bool = True, + mode: str = "regression", +) -> Path: + """Write an explicit current-vs-baseline visual review manifest to JSON.""" + + manifest = build_paired_visual_review_manifest( + current_proof_pack, + baseline_proof_pack, + task_intent=task_intent, + dimensions=dimensions, + preferred_renderer=preferred_renderer, + required=required, + mode=mode, + ) + output_path = Path(path) + save_json(manifest, output_path) + return output_path + + +def write_static_visual_review_manifest( + proof_pack: CaseProofPack, + path: str | Path, + *, + task_intent: str, + dimensions: tuple[str, ...] = STATIC_VISUAL_DIMENSIONS, + preferred_renderer: str | None = None, + required: bool = True, +) -> Path: + """Write a current-only static visual review manifest to JSON.""" + + manifest = build_static_visual_review_manifest( + proof_pack, + task_intent=task_intent, + dimensions=dimensions, + preferred_renderer=preferred_renderer, + required=required, + ) + output_path = Path(path) + save_json(manifest, output_path) + return output_path + + +def _load_case_renderer_reports( + root: Path, + report: AutonomousEvidenceReport, +) -> dict[str, RendererReport]: + renderer_reports: dict[str, RendererReport] = {} + for name in report.renderer_reports: + renderer_report_path = root / name / "report.json" + renderer_reports[name] = ( + load_renderer_report(renderer_report_path) + if renderer_report_path.exists() + else report.renderer_reports[name] + ) + for renderer_report_path in sorted(root.glob("*/report.json")): + name = renderer_report_path.parent.name + if name not in renderer_reports: + renderer_reports[name] = load_renderer_report(renderer_report_path) + return renderer_reports + + +def _snapshot_order( + report: AutonomousEvidenceReport, + bundle: SemanticSnapshotBundle | None, +) -> tuple[str, ...]: + raw_order = report.extra.get("snapshot_order") + if isinstance(raw_order, list) and all(isinstance(item, str) for item in raw_order): + return tuple(raw_order) + if bundle is not None: + return bundle.snapshot_order + if report.snapshot_metrics: + return tuple(report.snapshot_metrics) + return () + + +def _select_review_phase( + report: AutonomousEvidenceReport, + snapshot_order: tuple[str, ...], +) -> str: + final_snapshot = report.summary_metrics.get("final_snapshot_name") + if isinstance(final_snapshot, str) and final_snapshot: + return final_snapshot + failure_phase = _failure_phase(report.failure_taxonomy) + if failure_phase is not None: + return failure_phase + return snapshot_order[-1] if snapshot_order else "unknown" + + +def _failure_phase(failure_taxonomy: tuple[dict[str, Any], ...]) -> str | None: + for item in failure_taxonomy: + for key in ("phase", "snapshot", "snapshot_name"): + value = item.get(key) + if isinstance(value, str) and value: + return value + return None + + +def _metric_summary( + report: AutonomousEvidenceReport, + metric_keys: tuple[str, ...], +) -> dict[str, Any]: + metrics: dict[str, Any] = {} + for key in metric_keys: + if key in report.summary_metrics: + metrics[key] = report.summary_metrics[key] + elif key in report.extra: + metrics[key] = report.extra[key] + metrics["verdict"] = report.verdict + metrics["verdict_reasons"] = list(report.verdict_reasons) + return metrics + + +def _select_renderer_evidence( + root: Path, + renderer_reports: dict[str, RendererReport], + *, + selected_phase: str, + preferred_renderers: tuple[str, ...], + preferred_views: tuple[str, ...], +) -> tuple[ProofPackImageRef, ...]: + ordered_renderer_names = [name for name in preferred_renderers if name in renderer_reports] + [ + name for name in renderer_reports if name not in preferred_renderers + ] + refs: list[ProofPackImageRef] = [] + for renderer_name in ordered_renderer_names: + renderer_report = renderer_reports[renderer_name] + snapshot = _find_snapshot(renderer_report, selected_phase) + if snapshot is None: + continue + images_by_view = _images_by_view(snapshot.images) + for view in preferred_views: + image = images_by_view.get(view) + if image is None: + continue + refs.append( + ProofPackImageRef( + renderer=renderer_report.renderer or renderer_name, + phase=snapshot.name, + view=view, + path=_case_relative_path(root, Path(image.path)) or image.path, + metadata=_image_metadata(image), + ) + ) + return tuple(refs) + + +def _find_snapshot(renderer_report: RendererReport, name: str) -> Any | None: + for snapshot in renderer_report.snapshots: + if snapshot.name == name: + return snapshot + return renderer_report.snapshots[-1] if renderer_report.snapshots else None + + +def _images_by_view(images: tuple[RenderedImage, ...]) -> dict[str, RenderedImage]: + result: dict[str, RenderedImage] = {} + for image in images: + view = image.camera or image.view + if view is not None: + result[view] = image + return result + + +def _image_metadata(image: RenderedImage) -> dict[str, Any]: + metadata = dict(image.metadata) + for key in ( + "unique_colors", + "foreground_fraction", + "workspace_visible", + "workspace_center_xy", + "diff_from_previous", + ): + if key in image.extra: + metadata[key] = image.extra[key] + return metadata + + +def _artifact_refs( + root: Path, + report: AutonomousEvidenceReport, + *, + bundle_path: Path, + renderer_reports: dict[str, RendererReport], +) -> tuple[ProofPackArtifact, ...]: + artifacts = [ + ProofPackArtifact( + id="autonomous_report", + path="autonomous_report.json", + kind="autonomous_evidence_report", + ) + ] + if bundle_path.exists(): + artifacts.append( + ProofPackArtifact( + id="snapshot_bundle", + path="snapshot_bundle.json", + kind="semantic_snapshot_bundle", + ) + ) + for name, renderer_report in renderer_reports.items(): + renderer_path = root / name / "report.json" + artifacts.append( + ProofPackArtifact( + id=f"{name}_renderer_report", + path=( + _case_relative_path(root, renderer_path) + or _case_relative_path(root, Path(renderer_report.output_dir) / "report.json") + or f"{name}/report.json" + ), + kind="renderer_report", + ) + ) + return tuple(artifacts) + + +def _suite_results(suite_report: dict[str, Any]) -> tuple[dict[str, Any], ...]: + results = suite_report.get("results") + if not isinstance(results, list): + return () + return tuple(item for item in results if isinstance(item, dict)) + + +def _suite_case_dir(suite_dir: Path, result: dict[str, Any]) -> Path: + output_dir = result.get("output_dir") + if isinstance(output_dir, str) and output_dir: + path = Path(output_dir) + return path if path.is_absolute() else suite_dir / path + artifact_dir_name = result.get("artifact_dir_name") + if isinstance(artifact_dir_name, str) and artifact_dir_name: + return suite_dir / artifact_dir_name + return suite_dir / str(result.get("case_id") or "") + + +def _suite_relative_path(root: Path, path: Path) -> str | None: + if not path.as_posix(): + return None + root_resolved = root.resolve() + candidate = path if path.is_absolute() else Path.cwd() / path + try: + return candidate.resolve().relative_to(root_resolved).as_posix() + except ValueError: + return None + + +def _default_task_intent(proof_pack: CaseProofPack) -> str: + return ( + f"Review {proof_pack.case_id} static visual harness keyframes against " + "the bounded robot evidence and metric summary." + ) + + +def _case_relative_path(root: Path, path: Path) -> str | None: + if not path.as_posix(): + return None + root_resolved = root.resolve() + candidate = path + if not candidate.is_absolute(): + candidate = Path.cwd() / candidate + try: + return candidate.resolve().relative_to(root_resolved).as_posix() + except ValueError: + return None + + +def _manifest_evidence( + proof_pack: CaseProofPack, + *, + preferred_renderer: str | None, +) -> tuple[ProofPackImageRef, ...]: + if preferred_renderer is None: + if any(ref.renderer == "mujoco" for ref in proof_pack.renderer_evidence): + preferred_renderer = "mujoco" + elif proof_pack.renderer_evidence: + preferred_renderer = proof_pack.renderer_evidence[0].renderer + selected = tuple( + ref for ref in proof_pack.renderer_evidence if ref.renderer == preferred_renderer + ) + return selected or proof_pack.renderer_evidence + + +def _paired_manifest_evidence( + current_evidence: tuple[ProofPackImageRef, ...], + baseline_evidence: tuple[ProofPackImageRef, ...], +) -> tuple[tuple[ProofPackImageRef, ProofPackImageRef], ...]: + baseline_by_key = {(ref.renderer, ref.view): ref for ref in baseline_evidence} + pairs: list[tuple[ProofPackImageRef, ProofPackImageRef]] = [] + for current in current_evidence: + baseline = baseline_by_key.get((current.renderer, current.view)) + if baseline is not None: + pairs.append((current, baseline)) + return tuple(pairs) + + +def _dedupe(values: Any) -> tuple[str, ...]: + deduped: list[str] = [] + for value in values: + text = str(value) + if text not in deduped: + deduped.append(text) + return tuple(deduped) diff --git a/tests/unit/approval/test_visual_review.py b/tests/unit/approval/test_visual_review.py index 8f7846c..32b7dec 100644 --- a/tests/unit/approval/test_visual_review.py +++ b/tests/unit/approval/test_visual_review.py @@ -9,10 +9,13 @@ from roboharness.approval.visual_review import ( MANIFEST_SCHEMA_VERSION, RECORD_SCHEMA_VERSION, + VISUAL_REVIEW_SUMMARY_SCHEMA_VERSION, VisualReviewValidationError, + build_visual_review_summary, ingest_visual_review_record, validate_visual_review_manifest, write_visual_review_package, + write_visual_review_summary, ) @@ -119,6 +122,32 @@ def test_ingest_visual_review_record_accepts_paired_pass() -> None: assert result.summary["metric_findings"][0]["id"] == "visual.hand_pose" +def test_build_visual_review_summary_persists_effective_verdict(tmp_path: Path) -> None: + summary = build_visual_review_summary( + _manifest(), + _record(), + manifest_path="case/visual_review_manifest.json", + record_path="case/visual_review.json", + ) + + assert summary["schema_version"] == VISUAL_REVIEW_SUMMARY_SCHEMA_VERSION + assert summary["case_id"] == "case-1" + assert summary["is_valid"] is True + assert summary["effective_visual_verdict"] == "PASS" + assert summary["summary"]["manifest_path"] == "case/visual_review_manifest.json" + assert summary["summary"]["record_path"] == "case/visual_review.json" + + path = write_visual_review_summary( + _manifest(mode="current_only"), + _record(evidence=["current/lift/front_rgb.png"]), + tmp_path / "visual_review_summary.json", + ) + written = json.loads(path.read_text(encoding="utf-8")) + assert written["schema_version"] == VISUAL_REVIEW_SUMMARY_SCHEMA_VERSION + assert written["effective_visual_verdict"] == "NEEDS_HUMAN" + assert written["summary"]["needs_human_reasons"] == ["current_only_review_cannot_auto_pass"] + + def test_ingest_visual_review_record_fails_on_declared_dimension_failure() -> None: result = ingest_visual_review_record(_manifest(), _record(verdict="FAIL")) diff --git a/tests/unit/cli/test_cli.py b/tests/unit/cli/test_cli.py index 2731a55..d868c00 100644 --- a/tests/unit/cli/test_cli.py +++ b/tests/unit/cli/test_cli.py @@ -264,6 +264,140 @@ def test_contract_generate_and_check( captured = capsys.readouterr() assert "is current" in captured.out + def test_proof_pack_writes_case_artifacts( + self, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], + ) -> None: + case_dir = tmp_path / "case-1" + meshcat_dir = case_dir / "meshcat" + meshcat_dir.mkdir(parents=True) + (meshcat_dir / "final_front2back.png").write_bytes(b"png") + renderer_report = { + "output_dir": meshcat_dir.as_posix(), + "renderer": "meshcat", + "capture_ok": True, + "motion_ok": True, + "flags": [], + "trustworthiness_flags": [], + "snapshots": [ + { + "name": "final", + "capture_ok": True, + "motion_ok": True, + "metrics": {}, + "images": [ + { + "camera": "front2back", + "path": (meshcat_dir / "final_front2back.png").as_posix(), + } + ], + } + ], + } + (meshcat_dir / "report.json").write_text(json.dumps(renderer_report)) + (case_dir / "snapshot_bundle.json").write_text( + json.dumps( + { + "schema_version": 2, + "snapshot_order": ["final"], + "snapshots": [{"name": "final", "metrics": {}}], + } + ) + ) + (case_dir / "autonomous_report.json").write_text( + json.dumps( + { + "case_id": "case-1", + "output_dir": case_dir.as_posix(), + "verdict": "pass", + "verdict_reasons": [], + "failure_taxonomy": [], + "summary_metrics": {"final_snapshot_name": "final"}, + "snapshot_metrics": {}, + "renderer_reports": {"meshcat": renderer_report}, + } + ) + ) + + manifest_path = case_dir / "visual_review_manifest.json" + ret = main( + [ + "proof-pack", + str(case_dir), + "--visual-review-manifest", + str(manifest_path), + ] + ) + + assert ret == 0 + captured = capsys.readouterr() + assert "Proof pack written" in captured.out + proof_pack = json.loads((case_dir / "proof_pack.json").read_text()) + assert proof_pack["case_id"] == "case-1" + assert proof_pack["renderer_evidence"][0]["path"] == "meshcat/final_front2back.png" + manifest = json.loads(manifest_path.read_text()) + assert manifest["mode"] == "current_only" + + def test_visual_review_summary_cli_writes_ingested_record_summary( + self, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], + ) -> None: + manifest = { + "schema_version": "roboharness_visual_review_manifest/v1", + "case_id": "case-1", + "mode": "regression", + "task_intent": "Compare case evidence against the explicit baseline.", + "dimensions": [ + { + "id": "hand_pose", + "required": True, + "phase": "final", + "evidence_type": "paired_keyframe", + "views": ["front"], + "current": ["current.png"], + "baseline": ["baseline.png"], + "metric_fallback": ["grip_center_error_mm"], + } + ], + "metric_summary": {"grip_center_error_mm": 3.0}, + "review_policy": { + "requires_paired_evidence": True, + "allow_automatic_visual_pass": True, + }, + } + record = { + "schema_version": "roboharness_visual_review/v1", + "case_id": "case-1", + "reviewer_context": "unit_test", + "overall_visual_verdict": "PASS", + "dimensions": [ + { + "id": "hand_pose", + "verdict": "PASS", + "confidence": "medium", + "evidence": ["current.png", "baseline.png"], + "rationale": "Current and baseline evidence match.", + } + ], + "needs_human_reasons": [], + } + manifest_path = tmp_path / "visual_review_manifest.json" + record_path = tmp_path / "visual_review.json" + manifest_path.write_text(json.dumps(manifest), encoding="utf-8") + record_path.write_text(json.dumps(record), encoding="utf-8") + + ret = main(["visual-review-summary", str(manifest_path), str(record_path)]) + + assert ret == 0 + captured = capsys.readouterr() + assert "Visual review summary written" in captured.out + summary = json.loads((tmp_path / "visual_review_summary.json").read_text()) + assert summary["schema_version"] == "roboharness_visual_review_summary/v1" + assert summary["effective_visual_verdict"] == "PASS" + assert summary["summary"]["blocking_dimensions"] == [] + class TestVariantLayout: """Test with TaskStore-style layout: task/variant/trial/checkpoint.""" diff --git a/tests/unit/evidence/test_artifacts.py b/tests/unit/evidence/test_artifacts.py new file mode 100644 index 0000000..0147b08 --- /dev/null +++ b/tests/unit/evidence/test_artifacts.py @@ -0,0 +1,152 @@ +"""Evidence artifact compatibility tests.""" + +from __future__ import annotations + +from roboharness.evidence import ( + AutonomousEvidenceReport, + RendererReport, + SemanticSnapshotBundle, + load_autonomous_evidence_report, + load_renderer_report, + load_semantic_snapshot_bundle, +) + + +def _groot_style_renderer_report() -> dict[str, object]: + return { + "output_dir": "case_001/meshcat", + "renderer": "meshcat", + "capture_ok": True, + "motion_ok": True, + "flags": ["workspace_framing_warning"], + "trustworthiness_flags": [ + { + "code": "workspace_framing_failure", + "phase": "approach", + "severity": "warning", + } + ], + "metadata": { + "capture_wall_s": 0.42, + "replay_enabled": True, + "render_entrypoint": "snapshot_bundle_renderer", + }, + "snapshots": [ + { + "name": "pregrasp", + "capture_ok": True, + "motion_ok": True, + "metrics": { + "semantic_milestone": "pregrasp", + "control_backend": "decoupled_wbc", + "runtime_surface": "in_process", + }, + "images": [ + { + "camera": "front", + "path": "pregrasp/front.png", + "unique_colors": 128, + "green_pixels": 42, + "workspace_visible": True, + } + ], + } + ], + } + + +def test_semantic_snapshot_bundle_round_trips_groot_style_payload(tmp_path) -> None: + payload = { + "schema_version": 2, + "snapshot_order": ["pregrasp", "lift"], + "metadata": { + "robot_type": "g1", + "case_id": "X36_Y28_Z13", + "control_backend": "decoupled_wbc", + }, + "snapshots": [ + { + "name": "pregrasp", + "q": [0.0, 0.1, 0.2], + "viz_q": [0.0, 0.1, 0.2], + "bottle_xyz": [0.36, 0.28, 0.13], + "camera_focus_xyz": [0.4, 0.2, 0.5], + "grasp_markers": {"visible_grasp_trajectory_keys": ["approach_waypoints"]}, + "metrics": { + "semantic_milestone": "pregrasp", + "state_source": "env_state_act", + "snapshot_provenance": {"runtime_surface": "in_process"}, + }, + }, + { + "name": "lift", + "q": [0.3, 0.4, 0.5], + "metrics": { + "semantic_milestone": "lift", + "grip_center_error_mm": 12.5, + }, + }, + ], + } + + bundle = SemanticSnapshotBundle.from_dict(payload) + assert bundle.snapshot_order == ("pregrasp", "lift") + assert bundle.snapshots[0].extra["q"] == [0.0, 0.1, 0.2] + assert bundle.to_dict() == payload + + path = bundle.write_json(tmp_path / "snapshot_bundle.json") + assert load_semantic_snapshot_bundle(path).to_dict() == payload + + +def test_renderer_report_round_trips_groot_style_report(tmp_path) -> None: + payload = _groot_style_renderer_report() + + report = RendererReport.from_dict(payload) + assert report.renderer == "meshcat" + assert report.snapshots[0].images[0].extra["unique_colors"] == 128 + assert report.to_dict() == payload + + path = report.write_json(tmp_path / "meshcat_report.json") + assert load_renderer_report(path).to_dict() == payload + + +def test_autonomous_evidence_report_preserves_downstream_fields(tmp_path) -> None: + payload = { + "robot_type": "g1", + "case_id": "X36_Y28_Z13", + "output_dir": "case_001", + "runner": { + "runner_type": "g1_visual_harness", + "runtime_surface": "in_process", + "replay_source": "saved_visual_packet", + }, + "verdict": "pass", + "verdict_reasons": [], + "semantic_visual_ok": True, + "workspace_framing_ok": True, + "snapshot_state_progress_ok": True, + "failure_taxonomy": [], + "runtime": {"control_backend": "decoupled_wbc"}, + "plan": {"intent_level": "task_intent", "control_level": "planned_control"}, + "summary_metrics": { + "render_replay_enabled": True, + "render_total_s": 1.25, + }, + "thresholds": {"max_grip_center_error_mm": 50.0}, + "snapshot_order": ["pregrasp"], + "snapshot_metrics": { + "pregrasp": { + "semantic_milestone": "pregrasp", + "control_backend": "decoupled_wbc", + } + }, + "renderer_reports": {"meshcat": _groot_style_renderer_report()}, + } + + report = AutonomousEvidenceReport.from_dict(payload) + assert report.extra["robot_type"] == "g1" + assert report.renderer_reports["meshcat"].renderer == "meshcat" + assert report.to_dict() == payload + + path = report.write_json(tmp_path / "autonomous_report.json") + assert load_autonomous_evidence_report(path).to_dict() == payload diff --git a/tests/unit/evidence/test_proof_pack.py b/tests/unit/evidence/test_proof_pack.py new file mode 100644 index 0000000..1a49f36 --- /dev/null +++ b/tests/unit/evidence/test_proof_pack.py @@ -0,0 +1,457 @@ +from __future__ import annotations + +import json +from dataclasses import replace +from pathlib import Path + +import pytest + +from roboharness.approval.visual_review import ( + VisualReviewValidationError, + ingest_visual_review_record, + validate_visual_review_manifest, +) +from roboharness.evidence import ( + CASE_PROOF_PACK_SCHEMA_VERSION, + STATIC_VISUAL_DIMENSIONS, + SUITE_PROOF_PACK_SCHEMA_VERSION, + VISUAL_REVIEW_QUEUE_SCHEMA_VERSION, + build_case_proof_pack, + build_paired_visual_review_manifest, + build_static_visual_review_manifest, + build_suite_proof_pack, + build_visual_review_queue, + load_case_proof_pack, + load_suite_proof_pack, + write_visual_review_queue, +) + + +def _write_json(path: Path, payload: dict[str, object]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(payload, indent=2), encoding="utf-8") + + +def _write_image(path: Path) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(b"png") + + +def _renderer_report(case_dir: Path, renderer: str) -> dict[str, object]: + snapshots = [] + for snapshot_name in ("00_planned", "03_close_done", "09_home_final_done"): + images = [] + for camera in ("front2back", "left2right", "top2down"): + relative_path = f"{renderer}/{snapshot_name}_{camera}.png" + _write_image(case_dir / relative_path) + images.append( + { + "camera": camera, + "path": (case_dir / relative_path).as_posix(), + "unique_colors": 128, + "workspace_visible": True, + } + ) + snapshots.append( + { + "name": snapshot_name, + "capture_ok": True, + "motion_ok": True, + "metrics": {"semantic_milestone": snapshot_name}, + "images": images, + } + ) + return { + "output_dir": (case_dir / renderer).as_posix(), + "renderer": renderer, + "capture_ok": True, + "motion_ok": True, + "flags": [], + "trustworthiness_flags": [], + "metadata": {}, + "snapshots": snapshots, + } + + +def _write_groot_case(case_dir: Path) -> None: + meshcat_report = _renderer_report(case_dir, "meshcat") + mujoco_report = _renderer_report(case_dir, "mujoco") + _write_json(case_dir / "meshcat" / "report.json", meshcat_report) + _write_json(case_dir / "mujoco" / "report.json", mujoco_report) + _write_json( + case_dir / "snapshot_bundle.json", + { + "schema_version": 2, + "snapshot_order": ["00_planned", "03_close_done", "09_home_final_done"], + "metadata": { + "robot_type": "g1", + "case_id": "X36_Y28_Z13", + "control_backend": "decoupled_wbc", + "runtime_surface": "in_process", + }, + "snapshots": [ + { + "name": "00_planned", + "q": [0.0], + "metrics": {"semantic_milestone": "planned"}, + }, + { + "name": "03_close_done", + "q": [1.0], + "metrics": {"semantic_milestone": "close_done"}, + }, + { + "name": "09_home_final_done", + "q": [2.0], + "metrics": {"semantic_milestone": "home_final_done"}, + }, + ], + }, + ) + _write_json( + case_dir / "autonomous_report.json", + { + "case_id": "X36_Y28_Z13", + "output_dir": case_dir.as_posix(), + "robot_type": "g1", + "verdict": "pass", + "verdict_reasons": [], + "failure_taxonomy": [], + "runtime": {"control_backend": "decoupled_wbc"}, + "plan": {"intent_level": "task_intent"}, + "snapshot_order": ["00_planned", "03_close_done", "09_home_final_done"], + "snapshot_metrics": { + "03_close_done": { + "grip_center_error_mm": 12.0, + "semantic_milestone": "close_done", + } + }, + "summary_metrics": { + "final_snapshot_name": "09_home_final_done", + "grasp_accuracy_snapshot_name": "03_close_done", + "semantic_visual_ok": True, + "workspace_framing_ok": True, + "grip_center_error_mm": 12.0, + "pinch_gap_error_mm": 4.0, + "render_mujoco_enabled": True, + "render_total_s": 3.25, + }, + "renderer_reports": { + "meshcat": meshcat_report, + "mujoco": mujoco_report, + }, + }, + ) + + +def _write_suite_report(path: Path, case_dirs: list[Path]) -> None: + _write_json( + path, + { + "suite_name": "representative", + "output_root": path.parent.as_posix(), + "suite_verdict": "pass", + "total_cases": len(case_dirs), + "pass_count": len(case_dirs), + "fail_count": 0, + "execution_error_count": 0, + "results": [ + { + "case_id": case_dir.name, + "output_dir": case_dir.as_posix(), + "status": "pass", + "report_json": (case_dir / "autonomous_report.json").as_posix(), + } + for case_dir in case_dirs + ], + }, + ) + + +def test_build_case_proof_pack_from_groot_style_case(tmp_path: Path) -> None: + case_dir = tmp_path / "X36_Y28_Z13" + _write_groot_case(case_dir) + + proof_pack = build_case_proof_pack(case_dir) + + assert proof_pack.schema_version == CASE_PROOF_PACK_SCHEMA_VERSION + assert proof_pack.case_id == "X36_Y28_Z13" + assert proof_pack.verdict == "pass" + assert proof_pack.selected_phase == "09_home_final_done" + assert proof_pack.snapshot_order == ("00_planned", "03_close_done", "09_home_final_done") + assert proof_pack.metric_summary["grip_center_error_mm"] == 12.0 + assert {ref.renderer for ref in proof_pack.renderer_evidence} == {"meshcat", "mujoco"} + assert {ref.path for ref in proof_pack.renderer_evidence if ref.renderer == "mujoco"} == { + "mujoco/09_home_final_done_front2back.png", + "mujoco/09_home_final_done_left2right.png", + "mujoco/09_home_final_done_top2down.png", + } + assert [artifact.path for artifact in proof_pack.artifacts] == [ + "autonomous_report.json", + "snapshot_bundle.json", + "meshcat/report.json", + "mujoco/report.json", + ] + + path = proof_pack.write_json(case_dir / "proof_pack.json") + assert load_case_proof_pack(path).to_dict() == proof_pack.to_dict() + + +def test_build_suite_proof_pack_and_visual_review_queue_from_case_artifacts( + tmp_path: Path, +) -> None: + case_dir = tmp_path / "X36_Y28_Z13" + _write_groot_case(case_dir) + suite_report_path = tmp_path / "suite_report_representative.json" + _write_suite_report(suite_report_path, [case_dir]) + + suite_proof_pack = build_suite_proof_pack(suite_report_path) + + assert suite_proof_pack.schema_version == SUITE_PROOF_PACK_SCHEMA_VERSION + assert suite_proof_pack.suite_name == "representative" + assert suite_proof_pack.reviewable_count == 1 + assert suite_proof_pack.skipped_count == 0 + assert suite_proof_pack.cases[0].case_id == "X36_Y28_Z13" + assert suite_proof_pack.cases[0].status == "reviewable" + assert suite_proof_pack.cases[0].proof_pack_path == "X36_Y28_Z13/proof_pack.json" + assert ( + suite_proof_pack.cases[0].visual_review_manifest_path + == "X36_Y28_Z13/visual_review_manifest.json" + ) + assert (case_dir / "proof_pack.json").exists() + assert (case_dir / "visual_review_manifest.json").exists() + + suite_path = suite_proof_pack.write_json(tmp_path / "suite_proof_pack.json") + assert load_suite_proof_pack(suite_path).to_dict() == suite_proof_pack.to_dict() + + queue = build_visual_review_queue(suite_proof_pack) + assert queue.schema_version == VISUAL_REVIEW_QUEUE_SCHEMA_VERSION + assert queue.suite_name == "representative" + assert len(queue.items) == 1 + assert queue.items[0].visual_review_manifest_path == ("X36_Y28_Z13/visual_review_manifest.json") + queue_path = write_visual_review_queue(queue, tmp_path / "visual_review_queue.json") + queue_payload = json.loads(queue_path.read_text(encoding="utf-8")) + assert queue_payload["total_items"] == 1 + + +def test_suite_proof_pack_skips_execution_errors(tmp_path: Path) -> None: + suite_report_path = tmp_path / "suite_report_representative.json" + _write_json( + suite_report_path, + { + "suite_name": "representative", + "output_root": tmp_path.as_posix(), + "results": [ + { + "case_id": "BROKEN", + "output_dir": (tmp_path / "BROKEN").as_posix(), + "status": "execution_error", + "error_type": "RuntimeError", + "error": "boom", + } + ], + }, + ) + + suite_proof_pack = build_suite_proof_pack(suite_report_path) + queue = build_visual_review_queue(suite_proof_pack) + + assert suite_proof_pack.reviewable_count == 0 + assert suite_proof_pack.skipped_count == 1 + assert suite_proof_pack.cases[0].status == "skipped" + assert suite_proof_pack.cases[0].error == "case directory does not exist" + assert queue.items == () + + +def test_static_visual_review_manifest_uses_mujoco_keyframes_and_current_only_policy( + tmp_path: Path, +) -> None: + case_dir = tmp_path / "X36_Y28_Z13" + _write_groot_case(case_dir) + proof_pack = build_case_proof_pack(case_dir) + + manifest = build_static_visual_review_manifest( + proof_pack, + task_intent="Confirm the G1 final home pose and task success evidence.", + ) + + assert manifest["mode"] == "current_only" + assert manifest["review_policy"] == { + "requires_paired_evidence": False, + "allow_automatic_visual_pass": False, + "human_escalation_reasons": ["current_only_review_cannot_auto_pass"], + } + assert [dimension["id"] for dimension in manifest["dimensions"]] == list( + STATIC_VISUAL_DIMENSIONS + ) + first_dimension = manifest["dimensions"][0] + assert first_dimension["phase"] == "09_home_final_done" + assert first_dimension["evidence_type"] == "current_static_keyframe" + assert first_dimension["current"] == [ + "mujoco/09_home_final_done_front2back.png", + "mujoco/09_home_final_done_left2right.png", + "mujoco/09_home_final_done_top2down.png", + ] + validate_visual_review_manifest(manifest, current_root=case_dir) + + record = { + "schema_version": "roboharness_visual_review/v1", + "case_id": "X36_Y28_Z13", + "reviewer_context": "unit_test", + "overall_visual_verdict": "PASS", + "dimensions": [ + { + "id": dimension["id"], + "verdict": "PASS", + "confidence": "medium", + "evidence": list(dimension["current"]), + "rationale": "Static keyframes show no obvious issue.", + } + for dimension in manifest["dimensions"] + ], + "needs_human_reasons": [], + } + result = ingest_visual_review_record(manifest, record) + assert result.effective_visual_verdict == "NEEDS_HUMAN" + assert result.summary["needs_human_reasons"] == ["current_only_review_cannot_auto_pass"] + + +def test_paired_visual_review_manifest_requires_explicit_baseline_evidence( + tmp_path: Path, +) -> None: + current_dir = tmp_path / "current" / "X36_Y28_Z13" + baseline_dir = tmp_path / "baseline" / "X36_Y28_Z13" + _write_groot_case(current_dir) + _write_groot_case(baseline_dir) + current_proof_pack = build_case_proof_pack(current_dir) + baseline_proof_pack = build_case_proof_pack(baseline_dir) + + manifest = build_paired_visual_review_manifest( + current_proof_pack, + baseline_proof_pack, + task_intent="Confirm the current run does not regress against the blessed baseline.", + ) + + assert manifest["mode"] == "regression" + assert manifest["review_policy"] == { + "requires_paired_evidence": True, + "allow_automatic_visual_pass": True, + "human_escalation_reasons": [], + } + first_dimension = manifest["dimensions"][0] + assert first_dimension["evidence_type"] == "paired_keyframe" + assert first_dimension["current"] == [ + "mujoco/09_home_final_done_front2back.png", + "mujoco/09_home_final_done_left2right.png", + "mujoco/09_home_final_done_top2down.png", + ] + assert first_dimension["baseline"] == first_dimension["current"] + validate_visual_review_manifest( + manifest, + current_root=current_dir, + baseline_root=baseline_dir, + ) + + record = { + "schema_version": "roboharness_visual_review/v1", + "case_id": "X36_Y28_Z13", + "reviewer_context": "unit_test", + "overall_visual_verdict": "PASS", + "dimensions": [ + { + "id": dimension["id"], + "verdict": "PASS", + "confidence": "medium", + "evidence": list(dimension["current"]) + list(dimension["baseline"]), + "rationale": "Current and baseline keyframes match within visual tolerance.", + } + for dimension in manifest["dimensions"] + ], + "needs_human_reasons": [], + } + result = ingest_visual_review_record(manifest, record) + assert result.effective_visual_verdict == "PASS" + + +def test_paired_visual_review_manifest_migration_requires_baseline_blessing( + tmp_path: Path, +) -> None: + current_dir = tmp_path / "current" / "X36_Y28_Z13" + baseline_dir = tmp_path / "baseline" / "X36_Y28_Z13" + _write_groot_case(current_dir) + _write_groot_case(baseline_dir) + manifest = build_paired_visual_review_manifest( + build_case_proof_pack(current_dir), + build_case_proof_pack(baseline_dir), + task_intent="Confirm migration evidence before human baseline blessing.", + mode="migration", + ) + record = { + "schema_version": "roboharness_visual_review/v1", + "case_id": "X36_Y28_Z13", + "reviewer_context": "unit_test", + "overall_visual_verdict": "PASS", + "dimensions": [ + { + "id": dimension["id"], + "verdict": "PASS", + "confidence": "medium", + "evidence": list(dimension["current"]) + list(dimension["baseline"]), + "rationale": "Current and baseline keyframes match within visual tolerance.", + } + for dimension in manifest["dimensions"] + ], + "needs_human_reasons": [], + } + + result = ingest_visual_review_record(manifest, record) + + assert manifest["review_policy"]["allow_automatic_visual_pass"] is False + assert result.effective_visual_verdict == "NEEDS_HUMAN" + assert result.summary["needs_human_reasons"] == ["baseline_blessing_required"] + + +def test_paired_visual_review_manifest_rejects_case_mismatch(tmp_path: Path) -> None: + current_dir = tmp_path / "current" / "X36_Y28_Z13" + baseline_dir = tmp_path / "baseline" / "X36_Y28_Z13" + _write_groot_case(current_dir) + _write_groot_case(baseline_dir) + baseline_proof_pack = build_case_proof_pack(baseline_dir) + baseline_proof_pack = replace(baseline_proof_pack, case_id="OTHER_CASE") + + with pytest.raises(ValueError, match="matching case_id"): + build_paired_visual_review_manifest( + build_case_proof_pack(current_dir), + baseline_proof_pack, + task_intent="Confirm explicit paired evidence.", + ) + + +def test_static_visual_review_manifest_path_boundary_rejects_escape(tmp_path: Path) -> None: + case_dir = tmp_path / "X36_Y28_Z13" + _write_groot_case(case_dir) + proof_pack = build_case_proof_pack(case_dir) + manifest = build_static_visual_review_manifest( + proof_pack, + task_intent="Confirm static visual evidence.", + ) + manifest["dimensions"][0]["current"] = ["../escape.png"] + + with pytest.raises(VisualReviewValidationError, match="escapes its evidence root"): + validate_visual_review_manifest(manifest, current_root=case_dir) + + +def test_static_visual_review_manifest_rejects_temporal_dimension_for_v1( + tmp_path: Path, +) -> None: + case_dir = tmp_path / "X36_Y28_Z13" + _write_groot_case(case_dir) + proof_pack = build_case_proof_pack(case_dir) + manifest = build_static_visual_review_manifest( + proof_pack, + task_intent="Confirm static visual evidence.", + dimensions=("trajectory_naturalness",), + ) + + with pytest.raises(VisualReviewValidationError, match="temporal"): + validate_visual_review_manifest(manifest, current_root=case_dir) diff --git a/tests/unit/test_package_imports.py b/tests/unit/test_package_imports.py index 736330b..b07dce6 100644 --- a/tests/unit/test_package_imports.py +++ b/tests/unit/test_package_imports.py @@ -27,11 +27,14 @@ def find_spec(self, fullname, path=None, target=None): sys.meta_path.insert(0, BlockGymImports()) import roboharness - from roboharness import AssertionEngine, Harness + from roboharness import AssertionEngine, Harness, SemanticSnapshotBundle + from roboharness.evidence import RendererReport assert roboharness.__version__ assert AssertionEngine is not None assert Harness is not None + assert SemanticSnapshotBundle is not None + assert RendererReport is not None """ )