danrh
07f3dc254f
Introduce model-specific target profiles, fail-closed grounding and validation, content-addressed Director caching, LTX-2.5 contracts, comparison workflows, benchmarks, schemas, documentation, and regression coverage.
327 lines
12 KiB
Python
327 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
from contextlib import redirect_stdout
|
|
import hashlib
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
|
|
from scripts import benchmark_grounding
|
|
|
|
|
|
def _write_json(path: Path, value: dict) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(value), encoding="utf-8")
|
|
|
|
|
|
def _fact(fact_id: str, metric_tag: str = "supported_fact") -> dict:
|
|
return {
|
|
"id": fact_id,
|
|
"claim": fact_id.replace("_", " "),
|
|
"aliases": [],
|
|
"frame_indices": [],
|
|
"timecodes": [],
|
|
"metric_tags": [metric_tag],
|
|
}
|
|
|
|
|
|
def _case(
|
|
case_id: str,
|
|
*,
|
|
category: str = "synthetic_image",
|
|
private: bool = False,
|
|
expected: tuple[str, ...] = ("red_cube",),
|
|
prohibited: tuple[str, ...] = (),
|
|
group: str = "",
|
|
variant: str = "",
|
|
) -> dict:
|
|
return {
|
|
"id": case_id,
|
|
"category": category,
|
|
"fixture": f"fixtures/{case_id}.png",
|
|
"fixture_sha256": hashlib.sha256(b"fixture").hexdigest() if not private else "0" * 64,
|
|
"result": f"results/{case_id}.json",
|
|
"private": private,
|
|
"holdout": False,
|
|
"target_profile": "ltx",
|
|
"seeds": [7],
|
|
"expected_facts": [_fact(item) for item in expected],
|
|
"prohibited_facts": [_fact(item, "unsupported_fact") for item in prohibited],
|
|
"ambiguity_labels": [],
|
|
"counterfactual_group": group,
|
|
"counterfactual_variant": variant,
|
|
"notes": "",
|
|
}
|
|
|
|
|
|
def _manifest(cases: list[dict], seeds: list[int] | None = None) -> dict:
|
|
chosen_seeds = seeds or [7]
|
|
for case in cases:
|
|
case["seeds"] = list(chosen_seeds)
|
|
return {
|
|
"schema_version": benchmark_grounding.MANIFEST_SCHEMA_VERSION,
|
|
"name": "unit benchmark",
|
|
"description": "offline fixture",
|
|
"seeds": list(chosen_seeds),
|
|
"cases": cases,
|
|
}
|
|
|
|
|
|
def _condition(condition_id: str = "unit", **overrides) -> dict:
|
|
condition = {
|
|
"condition_id": condition_id,
|
|
"prompting": "guard_strict",
|
|
"sampling_profile": "checkpoint_defaults",
|
|
"media_variant": "case_default",
|
|
"precision": "nvfp4",
|
|
"telemetry_enabled": True,
|
|
"frame_budget": 0,
|
|
"visual_token_budget": 256,
|
|
"model_revision": "unit-test-revision",
|
|
"release_candidate": False,
|
|
}
|
|
condition.update(overrides)
|
|
return condition
|
|
|
|
|
|
def _result(
|
|
case_id: str,
|
|
reported: list[str],
|
|
unsupported: list[str] | None = None,
|
|
*,
|
|
seeds: list[int] | None = None,
|
|
condition: dict | None = None,
|
|
mode: str = "strict",
|
|
) -> dict:
|
|
chosen_seeds = seeds or [7]
|
|
chosen_condition = condition or _condition()
|
|
return {
|
|
"schema_version": benchmark_grounding.RESULT_SCHEMA_VERSION,
|
|
"case_id": case_id,
|
|
"runs": [
|
|
{
|
|
"seed": seed,
|
|
"mode": mode,
|
|
"condition": dict(chosen_condition),
|
|
"guard_decision": "pass",
|
|
"analysis_status": "grounded",
|
|
"reported_fact_ids": reported,
|
|
"unsupported_fact_ids": unsupported or [],
|
|
"metadata": {},
|
|
}
|
|
for seed in chosen_seeds
|
|
],
|
|
}
|
|
|
|
|
|
class GroundingBenchmarkTests(unittest.TestCase):
|
|
def test_scores_facts_and_counterfactual_pairs_from_recorded_results(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
cases = [
|
|
_case(
|
|
"real",
|
|
expected=("red_cube",),
|
|
prohibited=("dragon",),
|
|
group="image-swap",
|
|
variant="real",
|
|
),
|
|
_case(
|
|
"unrelated",
|
|
expected=("blue_sphere",),
|
|
prohibited=("red_cube",),
|
|
group="image-swap",
|
|
variant="unrelated",
|
|
),
|
|
]
|
|
manifest_path = root / "manifest.json"
|
|
_write_json(manifest_path, _manifest(cases))
|
|
for case in cases:
|
|
fixture = root / case["fixture"]
|
|
fixture.parent.mkdir(parents=True, exist_ok=True)
|
|
fixture.write_bytes(b"fixture")
|
|
_write_json(root / "results/real.json", _result("real", ["red_cube", "dragon"], ["dragon"]))
|
|
_write_json(root / "results/unrelated.json", _result("unrelated", ["blue_sphere"]))
|
|
|
|
report = benchmark_grounding.evaluate_manifest(manifest_path)
|
|
|
|
self.assertEqual(report["metrics"]["supported_fact_recall"], 1.0)
|
|
self.assertAlmostEqual(report["metrics"]["supported_fact_precision"], 2 / 3)
|
|
self.assertAlmostEqual(report["metrics"]["unsupported_fact_rate"], 1 / 3)
|
|
self.assertEqual(report["metrics"]["silent_hallucination_run_count"], 1)
|
|
self.assertEqual(report["counterfactual"]["pair_count"], 1)
|
|
self.assertEqual(report["counterfactual"]["discrimination_rate"], 1.0)
|
|
|
|
def test_missing_private_fixture_is_a_skip(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
manifest_path = root / "manifest.json"
|
|
_write_json(
|
|
manifest_path,
|
|
_manifest(
|
|
[
|
|
_case(
|
|
"private_failure",
|
|
category="private_known_failure",
|
|
private=True,
|
|
)
|
|
]
|
|
),
|
|
)
|
|
|
|
report = benchmark_grounding.evaluate_manifest(manifest_path)
|
|
|
|
self.assertEqual(report["metrics"]["scored_run_count"], 0)
|
|
self.assertEqual(
|
|
report["skipped_cases"],
|
|
[{"case_id": "private_failure", "reason": "private_fixture_missing"}],
|
|
)
|
|
|
|
def test_missing_public_fixture_is_an_error(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
manifest_path = Path(directory) / "manifest.json"
|
|
_write_json(manifest_path, _manifest([_case("public")]))
|
|
|
|
with self.assertRaisesRegex(benchmark_grounding.ManifestError, "Public fixture"):
|
|
benchmark_grounding.evaluate_manifest(manifest_path)
|
|
|
|
def test_fixture_hash_mismatch_is_an_error(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
case = _case("changed")
|
|
manifest_path = root / "manifest.json"
|
|
_write_json(manifest_path, _manifest([case]))
|
|
fixture = root / case["fixture"]
|
|
fixture.parent.mkdir(parents=True, exist_ok=True)
|
|
fixture.write_bytes(b"different bytes")
|
|
|
|
with self.assertRaisesRegex(benchmark_grounding.ManifestError, "hash mismatch"):
|
|
benchmark_grounding.evaluate_manifest(manifest_path)
|
|
|
|
def test_manifest_rejects_fact_ids_that_are_expected_and_prohibited(self):
|
|
case = _case("overlap", expected=("same",), prohibited=("same",))
|
|
|
|
with self.assertRaisesRegex(benchmark_grounding.ManifestError, "both expected and prohibited"):
|
|
benchmark_grounding.validate_manifest(_manifest([case]))
|
|
|
|
def test_conditions_can_coexist_per_seed_but_each_must_cover_every_seed(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
case = _case("matrix")
|
|
manifest_path = root / "manifest.json"
|
|
_write_json(manifest_path, _manifest([case], seeds=[7, 11]))
|
|
fixture = root / case["fixture"]
|
|
fixture.parent.mkdir(parents=True, exist_ok=True)
|
|
fixture.write_bytes(b"fixture")
|
|
first = _result(
|
|
"matrix",
|
|
["red_cube"],
|
|
seeds=[7, 11],
|
|
condition=_condition("defaults"),
|
|
)
|
|
second = _result(
|
|
"matrix",
|
|
["red_cube"],
|
|
seeds=[7, 11],
|
|
condition=_condition(
|
|
"diagnostic",
|
|
sampling_profile="full_48_diagnostic",
|
|
telemetry_enabled=False,
|
|
),
|
|
)
|
|
first["runs"].extend(second["runs"])
|
|
_write_json(root / case["result"], first)
|
|
|
|
report = benchmark_grounding.evaluate_manifest(manifest_path)
|
|
self.assertEqual(report["metrics"]["scored_run_count"], 4)
|
|
|
|
first["runs"].pop()
|
|
_write_json(root / case["result"], first)
|
|
with self.assertRaisesRegex(benchmark_grounding.ResultError, "incomplete"):
|
|
benchmark_grounding.evaluate_manifest(manifest_path)
|
|
|
|
def test_result_rejects_duplicate_seed_condition_and_invalid_typed_condition(self):
|
|
case = benchmark_grounding.validate_manifest(_manifest([_case("duplicate")]))["cases"][0]
|
|
result = _result("duplicate", ["red_cube"])
|
|
result["runs"].append(dict(result["runs"][0]))
|
|
with self.assertRaisesRegex(benchmark_grounding.ResultError, "duplicates seed"):
|
|
benchmark_grounding._validate_result(result, case, Path("result.json"))
|
|
|
|
malformed = _result("duplicate", ["red_cube"])
|
|
malformed["runs"][0]["condition"]["precision"] = "mystery"
|
|
with self.assertRaisesRegex(benchmark_grounding.ResultError, "precision"):
|
|
benchmark_grounding._validate_result(malformed, case, Path("result.json"))
|
|
|
|
def test_release_candidate_requires_strict_mode(self):
|
|
case = benchmark_grounding.validate_manifest(_manifest([_case("release")]))["cases"][0]
|
|
result = _result(
|
|
"release",
|
|
["red_cube"],
|
|
condition=_condition("release", release_candidate=True),
|
|
mode="audit",
|
|
)
|
|
result["runs"][0]["guard_decision"] = "pass"
|
|
with self.assertRaisesRegex(benchmark_grounding.ResultError, "strict mode"):
|
|
benchmark_grounding._validate_result(result, case, Path("result.json"))
|
|
|
|
def test_manifest_rejects_private_category_spoofing_and_path_traversal(self):
|
|
spoofed = _case("spoofed")
|
|
spoofed["private"] = True
|
|
with self.assertRaisesRegex(benchmark_grounding.ManifestError, "private must be false"):
|
|
benchmark_grounding.validate_manifest(_manifest([spoofed]))
|
|
|
|
escaped = _case("escaped")
|
|
escaped["fixture"] = "../outside.png"
|
|
with self.assertRaisesRegex(benchmark_grounding.ManifestError, "confined relative path"):
|
|
benchmark_grounding.validate_manifest(_manifest([escaped]))
|
|
|
|
def test_small_or_missing_release_dataset_never_gets_a_passing_verdict(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
case = _case("not_release_scale")
|
|
manifest_path = root / "manifest.json"
|
|
_write_json(manifest_path, _manifest([case]))
|
|
fixture = root / case["fixture"]
|
|
fixture.parent.mkdir(parents=True, exist_ok=True)
|
|
fixture.write_bytes(b"fixture")
|
|
_write_json(
|
|
root / case["result"],
|
|
_result(
|
|
case["id"],
|
|
["red_cube"],
|
|
condition=_condition("release", release_candidate=True),
|
|
),
|
|
)
|
|
report = benchmark_grounding.evaluate_manifest(manifest_path)
|
|
|
|
self.assertEqual(report["release_candidate"]["status"], "incomplete")
|
|
self.assertFalse(report["release_candidate"]["eligible_for_verdict"])
|
|
self.assertFalse(report["release_candidate"]["passed"])
|
|
|
|
def test_cli_writes_no_file_without_output_option(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
case = _case("ready")
|
|
manifest_path = root / "manifest.json"
|
|
_write_json(manifest_path, _manifest([case]))
|
|
fixture = root / case["fixture"]
|
|
fixture.parent.mkdir(parents=True, exist_ok=True)
|
|
fixture.write_bytes(b"fixture")
|
|
_write_json(root / case["result"], _result("ready", ["red_cube"]))
|
|
before = sorted(path.relative_to(root) for path in root.rglob("*") if path.is_file())
|
|
|
|
output = io.StringIO()
|
|
with redirect_stdout(output):
|
|
exit_code = benchmark_grounding.main([str(manifest_path)])
|
|
after = sorted(path.relative_to(root) for path in root.rglob("*") if path.is_file())
|
|
|
|
self.assertEqual(exit_code, 0)
|
|
self.assertEqual(before, after)
|
|
self.assertEqual(json.loads(output.getvalue())["metrics"]["scored_run_count"], 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|