diff --git a/docs/operations.md b/docs/operations.md index 1fb9bb14..9a3e2420 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -236,6 +236,19 @@ Operation success, recorded execution state and the experiment verdict are distinct conclusions; a completed run does not upgrade imported knowledge or unvalidated candidates. +### Inspect a fixed attempt — `status --record-ref` + +Pass a JSON `RecordRef` file and its trusted `--storage-config` JSON. An +optional `--assessment-ref` JSON adds the archived claim outcomes and scopes +for the same record revision. The output shows committed execution facts, +including phase, process outcome, cleanup and recording time. A successful +read exits 0. Invalid references, archive integrity failures and mismatched +revisions exit 2. + +The same read-only view is available through +`inferrsi.apps.api.read_attempt_view`. The legacy `status --run-id` form +reads the legacy experiment directory. + ## Campaign summary figure ### Summarize a campaign figure — `tools/campaign_summary_chart.py` diff --git a/src/inferrsi/apps/api.py b/src/inferrsi/apps/api.py index 2260a08f..26e91dd2 100644 --- a/src/inferrsi/apps/api.py +++ b/src/inferrsi/apps/api.py @@ -30,6 +30,7 @@ preparation_result, validate_result, ) +from .views.attempt import read_attempt_view as read_attempt_view _PREPARE_FIELDS = {"schema_version", "request"} _BENCHMARK_REQUEST_FIELDS = { diff --git a/src/inferrsi/apps/views/attempt.py b/src/inferrsi/apps/views/attempt.py new file mode 100644 index 00000000..4c7a2467 --- /dev/null +++ b/src/inferrsi/apps/views/attempt.py @@ -0,0 +1,88 @@ +"""Read a fixed Harness revision and its optional archived assessment.""" + +from copy import deepcopy + +from inferrsi._json_v1 import decode +from inferrsi.harness.errors import PublicError + +from ..archive import read_assessment +from ..artifact_reader import ArtifactReader +from ..bridge import check_assessment_identity +from ..result import _check_ref + +_EXECUTION_FIELDS = ( + "run_id", "attempt_id", "record_version", "phase", "started", "side_effect_unknown", + "process_outcome", "exit_code", "recorded_at", "prepared_ref", +) +_CLAIM_FIELDS = ("claim_id", "validation_scope", "verdict", "reasons", "missing_evidence") + + +def read_attempt_view( + record_ref: dict, *, storage_config: dict, assessment_ref: dict | None = None, +) -> dict: + """Return fixed execution facts and claim outcomes for one record revision.""" + try: + _check_ref(record_ref, "attempt_id", "record_version") + if record_ref["artifact"]["media_type"] != "application/json": + raise ValueError("expected a JSON record") + except ValueError: + raise PublicError( + "InvalidRequest", stage="attempt_view", message="Invalid RecordRef." + ) from None + + with ArtifactReader(storage_config).open(record_ref["artifact"]) as stream: + data = stream.read() + try: + record = decode(data) + if ( + not isinstance(record, dict) + or type(record.get("schema_version")) is not int or record["schema_version"] != 1 + or type(record.get("record_version")) is not int + ): + raise ValueError("expected a v1 committed record") + execution = {field: record[field] for field in _EXECUTION_FIELDS} + execution["cleanup_status"] = record["cleanup"]["status"] + candidate = record["candidate"] + execution["candidate_id"] = candidate["candidate_id"] if candidate is not None else None + except (KeyError, TypeError, ValueError): + raise PublicError( + "ArtifactIntegrityError", stage="attempt_view", + message="Invalid committed record bytes.", + ) from None + if any(record[field] != record_ref[field] for field in ("attempt_id", "record_version")): + raise PublicError( + "IdentityMismatch", stage="attempt_view", + message="RecordRef identifies a different record revision.", + ) + + evaluation = None + if assessment_ref is not None: + assessment = read_assessment(storage_config, assessment_ref) + expected = { + "record_ref": record_ref, + **{key: record.get(key) for key in ( + "attempt_id", "prepared_ref", "protocol_digest", "candidate", + )}, + } + if any(assessment.get(key) != value for key, value in expected.items()): + raise PublicError( + "IdentityMismatch", stage="attempt_view", + message="AssessmentRef belongs to a different fixed attempt record.", + ) + try: + check_assessment_identity(assessment) + evaluation = { + "assessed_at": assessment["assessed_at"], + "claims": [{key: claim[key] for key in _CLAIM_FIELDS} + for claim in assessment["claims"]], + } + except (KeyError, TypeError, ValueError): + raise PublicError( + "ArtifactIntegrityError", stage="attempt_view", + message="Invalid archived assessment for this record.", + ) from None + + return { + "record_ref": deepcopy(record_ref), "execution": execution, + "assessment_ref": deepcopy(assessment_ref), "assessment": evaluation, + } diff --git a/src/inferrsi/arguments.py b/src/inferrsi/arguments.py index 2a116a69..22a9be81 100644 --- a/src/inferrsi/arguments.py +++ b/src/inferrsi/arguments.py @@ -211,7 +211,14 @@ def parser() -> argparse.ArgumentParser: name, help="Read a recorded experiment" if name == "status" else "Regenerate its HTML report", ) - child.add_argument("--run-id", required=True) + if name == "status": + source = child.add_mutually_exclusive_group(required=True) + source.add_argument("--run-id") + source.add_argument("--record-ref", type=Path, help="JSON file containing a RecordRef") + child.add_argument("--storage-config", type=Path, help="Trusted storage config JSON") + child.add_argument("--assessment-ref", type=Path, help="Optional AssessmentRef JSON") + else: + child.add_argument("--run-id", required=True) child.add_argument("--output-dir", default="runs") dashboard = commands.add_parser( diff --git a/src/inferrsi/cli.py b/src/inferrsi/cli.py index c075bce6..00288814 100644 --- a/src/inferrsi/cli.py +++ b/src/inferrsi/cli.py @@ -329,6 +329,29 @@ def main(argv: list[str] | None = None) -> int: print(text, end="") return 0 if args.command in ("status", "report"): + if args.command == "status" and args.record_ref is not None: + if args.storage_config is None: + raise ValueError("--record-ref requires --storage-config") + from .apps.api import read_attempt_view + from .harness.errors import PublicError + + record_ref = json.loads(args.record_ref.read_text(encoding="utf-8")) + config = json.loads(args.storage_config.read_text(encoding="utf-8")) + assessment_ref = ( + json.loads(args.assessment_ref.read_text(encoding="utf-8")) + if args.assessment_ref is not None else None + ) + if args.assessment_ref is not None and not isinstance(assessment_ref, dict): + raise ValueError("--assessment-ref requires an AssessmentRef JSON object") + try: + _print(read_attempt_view( + record_ref, storage_config=config, assessment_ref=assessment_ref, + )) + except PublicError as exc: + raise ValueError(str(exc)) from exc + return 0 + if args.command == "status" and (args.storage_config or args.assessment_ref): + raise ValueError("--storage-config and --assessment-ref require --record-ref") directory = _run_directory(args.output_dir, args.run_id) if args.command == "status": _print(json.loads((directory / "status.json").read_text(encoding="utf-8"))) diff --git a/tests/test_apps_assess.py b/tests/test_apps_assess.py index d52a14a1..1d529ed8 100644 --- a/tests/test_apps_assess.py +++ b/tests/test_apps_assess.py @@ -1,5 +1,6 @@ -"""Apps assembles offline assessment from a fixed record and publishes its result.""" +"""Apps assembles and reads offline assessments from fixed records.""" +import json import subprocess import sys from copy import deepcopy @@ -11,7 +12,9 @@ from inferrsi._json_v1 import decode, encode from inferrsi.apps import archive from inferrsi.apps.api import assess as run_assess +from inferrsi.apps.api import read_attempt_view from inferrsi.apps.artifact_reader import ArtifactReader +from inferrsi.cli import main from inferrsi.harness.errors import PublicError from inferrsi.harness.run_store import RunStore @@ -217,6 +220,101 @@ def test_read_assessment_rejects_the_wrong_identity_even_if_bytes_match(offline) archive.read_assessment(config, ref) +def test_fixed_attempt_view_keeps_process_facts_and_claim_outcomes_separate(offline): + request, config, directory = offline + record_ref = request["record_ref"] + record = _fixture("record-payload") + ref = _fixture("fixed-refs")["assessment_ref"] + ref["artifact"] = _write_export(directory, "assessment.json", encode(_fixture("assessment"))) + + assert read_attempt_view(record_ref, storage_config=config)["assessment"] is None + view = read_attempt_view(record_ref, storage_config=config, assessment_ref=ref) + assert view["record_ref"] == record_ref + assert view["assessment_ref"] == ref + assert view["execution"] == { + **{key: record[key] for key in ( + "run_id", "attempt_id", "record_version", "phase", "started", "side_effect_unknown", + "process_outcome", "exit_code", "recorded_at", "prepared_ref", + )}, + "candidate_id": record["candidate"]["candidate_id"], + "cleanup_status": record["cleanup"]["status"], + } + assert [(c["claim_id"], c["validation_scope"], c["verdict"]) + for c in view["assessment"]["claims"]] == [ + (c["claim_id"], c["validation_scope"], c["verdict"]) + for c in _fixture("assessment")["claims"] + ] + assert view["assessment"]["claims"][0]["missing_evidence"] == ["format"] + view["record_ref"]["record_version"] = 42 + assert record_ref["record_version"] == 1 + + +def test_fixed_attempt_view_rejects_a_swapped_record_identity(offline): + request, config, _ = offline + ref = {**request["record_ref"], "record_version": 2} + with pytest.raises(PublicError, match="IdentityMismatch"): + read_attempt_view(ref, storage_config=config) + + +def test_fixed_attempt_view_rejects_an_assessment_of_another_revision(offline): + request, config, directory = offline + decision = _fixture("assessment") + decision["record_ref"]["record_version"] = 2 + _sign_assessment(decision) + ref = { + "assessment_id": decision["assessment_id"], "schema_version": 1, + "artifact": _write_export(directory, "assessment.json", encode(decision)), + } + with pytest.raises(PublicError, match="IdentityMismatch"): + read_attempt_view(request["record_ref"], storage_config=config, assessment_ref=ref) + + +def test_fixed_attempt_view_rechecks_assessment_semantic_identity(offline): + request, config, directory = offline + decision = _fixture("assessment") + decision["claims"][0]["verdict"] = "PASS" + ref = { + "assessment_id": decision["assessment_id"], "schema_version": 1, + "artifact": _write_export(directory, "assessment.json", encode(decision)), + } + with pytest.raises(PublicError, match="ArtifactIntegrityError"): + read_attempt_view(request["record_ref"], storage_config=config, assessment_ref=ref) + + +def test_fixed_status_command_matches_the_python_view(tmp_path, offline, capsys): + request, config, directory = offline + record_ref = request["record_ref"] + assessment_ref = _fixture("fixed-refs")["assessment_ref"] + assessment_ref["artifact"] = _write_export( + directory, "assessment.json", encode(_fixture("assessment")), + ) + flags = ["status"] + for name, value in ( + ("record-ref", record_ref), ("storage-config", config), ("assessment-ref", assessment_ref), + ): + path = tmp_path / f"{name}.json" + path.write_bytes(encode(value)) + flags.extend((f"--{name}", str(path))) + assert main(flags) == 0 + assert json.loads(capsys.readouterr().out) == read_attempt_view( + record_ref, storage_config=config, assessment_ref=assessment_ref, + ) + + +@pytest.mark.parametrize(("args", "message"), [ + (["--record-ref", "record.json"], "--record-ref requires --storage-config"), + (["--run-id", "legacy", "--storage-config", "storage.json"], "require --record-ref"), + (["--run-id", "legacy", "--record-ref", "record.json"], "not allowed with argument"), +]) +def test_fixed_status_requires_its_own_flag_combination(args, message, capsys): + with pytest.raises(SystemExit) as error: + main(["status", *args]) + assert error.value.code == 2 + captured = capsys.readouterr() + assert captured.out == "" + assert message in captured.err + + def test_new_process_reassesses_the_requested_sqlite_revision(tmp_path, offline): request, config, _ = offline store = RunStore(config["mounts"][1]["root"], config["harness_namespace"]) @@ -240,7 +338,7 @@ def test_new_process_reassesses_the_requested_sqlite_revision(tmp_path, offline) [sys.executable, "-c", """ from pathlib import Path from inferrsi._json_v1 import decode, encode -from inferrsi.apps.api import assess as run_assess +from inferrsi.apps.api import assess as run_assess, read_attempt_view workspace = Path.cwd() request, config, decision = decode((workspace / 'offline.json').read_bytes()) @@ -255,6 +353,11 @@ def assessor(prepared, evidence, *, artifacts): encode_assessment=encode, output_id='reassessed') assert output.result['exit_code'] == 9 +view = read_attempt_view(request['record_ref'], storage_config=config, + assessment_ref=output.result['assessment_ref']) +assert view['execution']['record_version'] == 1 +assert view['execution']['phase'] == 'running' +assert view['assessment']['claims'][0]['verdict'] == 'INCONCLUSIVE' """], cwd=tmp_path, capture_output=True, text=True, check=True, )