fix: fingerprint material inputs for resumable archive runs

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

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