Read fixed attempt status and archived claims in Apps

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`

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 = {

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,
+    }

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(

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")))

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,
     )