diff --git a/run_archive.py b/run_archive.py index 56bde51..aedbd58 100644 --- a/run_archive.py +++ b/run_archive.py @@ -202,6 +202,39 @@ def _sha256_file(path: Path) -> str: return digest.hexdigest() +def _material_fingerprints(materials: Sequence[Path]) -> List[Dict[str, str]]: + """Return the immutable path/content identity for local source materials.""" + return [ + { + "path": str(path), + "sha256": _sha256_file(path), + } + for path in materials + ] + + +def _validate_material_fingerprints(object_job: ObjectJob, row: Dict[str, object]) -> None: + """Reject resume when effective material inputs differ from the persisted run.""" + expected = _material_fingerprints(object_job.materials) + stored = row.get("material_fingerprints") + if stored is None: + if expected: + raise RunnerError( + "state lacks material fingerprints for %s; cannot safely resume this material-backed run; " + "use a new input filename" % object_job.name + ) + return + if not isinstance(stored, list) or len(stored) != len(expected): + raise RunnerError("state material association is inconsistent for %s" % object_job.name) + for stored_item, expected_item in zip(stored, expected): + if not isinstance(stored_item, dict): + raise RunnerError("state material fingerprints are invalid for %s" % object_job.name) + if str(stored_item.get("path") or "") != expected_item["path"]: + raise RunnerError("state material association is inconsistent for %s" % object_job.name) + if str(stored_item.get("sha256") or "") != expected_item["sha256"]: + raise RunnerError("material content changed after this run started: %s" % expected_item["path"]) + + def _dedupe(items: Iterable[object]) -> List[str]: seen = set() result: List[str] = [] @@ -341,6 +374,7 @@ def _new_state(job: JobFile) -> Dict[str, object]: "session_id": "", "archive": str(item.archive_path), "archive_sha256": "", + "material_fingerprints": _material_fingerprints(item.materials), "verification_submissions": 0, "last_error": "", } @@ -383,6 +417,7 @@ def load_or_create_state(job: JobFile) -> Dict[str, object]: raise RunnerError("state project association is inconsistent for %s" % item.name) if str(row.get("archive") or "") != str(item.archive_path): raise RunnerError("state archive association is inconsistent for %s" % item.name) + _validate_material_fingerprints(item, row) return state @@ -795,6 +830,7 @@ def _session_metadata( "language": job.language, "output_path": str(object_job.archive_path), "source_materials": [str(path) for path in object_job.materials], + "source_material_fingerprints": list(item_state.get("material_fingerprints") or []), "runtime_materials": [str(item.get("runtime_path") or "") for item in staged_materials], "status": status, "archive_sha256": str(item_state.get("archive_sha256") or ""), @@ -847,6 +883,7 @@ def process_object( verbose: bool, ) -> None: """Run or resume exactly one queue item until accepted or exhausted.""" + _validate_material_fingerprints(object_job, item_state) shell_state = _open_or_create_session(app, object_job, item_state) item_state["status"] = "running" item_state["session_id"] = shell_state.session_id diff --git a/tests/test_run_archive_material_provenance.py b/tests/test_run_archive_material_provenance.py new file mode 100644 index 0000000..b119048 --- /dev/null +++ b/tests/test_run_archive_material_provenance.py @@ -0,0 +1,116 @@ +"""Regression coverage for immutable local-material provenance. + +This module reuses the offline import harness from ``test_run_archive_offline`` +so the contract remains deterministic and requires no Moonshine runtime, API +credentials, network access, or model calls. +""" + +from __future__ import annotations + +import json +import sys +import tempfile +import unittest +from pathlib import Path + +import test_run_archive_offline as harness + + +class MaterialProvenanceRegressionTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.runner, cls._module_patcher = harness._load_runner_module() + + @classmethod + def tearDownClass(cls): + sys.modules.pop(harness.RUNNER_MODULE_NAME, None) + cls._module_patcher.stop() + + def setUp(self): + self._temporary_directory = tempfile.TemporaryDirectory() + self.addCleanup(self._temporary_directory.cleanup) + self.temp_root = Path(self._temporary_directory.name) + self.task_dir = self.temp_root / "Creative-Intelligence" + self.task_dir.mkdir() + self._original_task_dir = self.runner.TASK_DIR + self.runner.TASK_DIR = self.task_dir + self.addCleanup(setattr, self.runner, "TASK_DIR", self._original_task_dir) + + def _material_job(self): + inputs = self.temp_root / "inputs" + inputs.mkdir() + material = inputs / "notes.md" + material.write_text("original source material\n", encoding="utf-8") + queue_path = inputs / "queue.json" + queue_path.write_text( + json.dumps( + { + "format": self.runner.FORMAT_ID, + "language": "en", + "objects": [ + { + "name": "Bochner formula", + "materials": ["notes.md"], + } + ], + }, + ensure_ascii=False, + indent=2, + ) + + "\n", + encoding="utf-8", + ) + return material, queue_path + + def test_new_state_records_resolved_material_path_and_sha256(self): + material, queue_path = self._material_job() + job = self.runner.load_job(queue_path) + + state = self.runner.load_or_create_state(job) + + self.assertEqual( + state["objects"][0]["material_fingerprints"], + [ + { + "path": str(material.resolve()), + "sha256": self.runner._sha256_file(material), + } + ], + ) + self.assertEqual( + self.runner.load_or_create_state(self.runner.load_job(queue_path)), + state, + ) + + def test_state_rejects_material_content_mutation_after_run_started(self): + material, queue_path = self._material_job() + first_job = self.runner.load_job(queue_path) + state = self.runner.load_or_create_state(first_job) + self.assertEqual(state["status"], "pending") + self.assertTrue(first_job.state_path.exists()) + + material.write_text("mutated source material\n", encoding="utf-8") + resumed_job = self.runner.load_job(queue_path) + + with self.assertRaisesRegex( + self.runner.RunnerError, + "material content changed after this run started", + ): + self.runner.load_or_create_state(resumed_job) + + def test_material_backed_legacy_state_without_fingerprint_fails_closed(self): + _, queue_path = self._material_job() + job = self.runner.load_job(queue_path) + state = self.runner.load_or_create_state(job) + state["objects"][0].pop("material_fingerprints") + harness._stub_write_json(job.state_path, state) + + with self.assertRaisesRegex( + self.runner.RunnerError, + "state lacks material fingerprints", + ): + self.runner.load_or_create_state(self.runner.load_job(queue_path)) + + +if __name__ == "__main__": + unittest.main()