fix: harden resumable discovery lifecycle integrity

Author: Charlie-Wang-03

run_archive.py

@@ -43,6 +43,7 @@
 AGENT_SLUG = "moonshine-core"
 STATE_SCHEMA_VERSION = 1
 SOURCE_CONTEXT_TOKEN_BUDGET = 60_000
+DISCOVERY_ROW_STATUSES = {"selecting", "proposed", "running", "verified", "failed", "stopped"}
 
 BASE_EXPOSED_TOOLS = [
     "load_skill_definition",
@@ -280,6 +281,20 @@ def _validate_material_fingerprints(object_job: ObjectJob, row: Dict[str, object
             raise RunnerError("material content changed after this run started: %s" % expected_item["path"])
 
 
+def _validate_discovery_verified_archive(name: str, archive: str, expected_hash: str) -> None:
+    """Reject a verified discovery row whose published artifact is not intact."""
+    path = Path(str(archive or ""))
+    digest = str(expected_hash or "").strip()
+    if not archive or not digest or not path.exists() or not path.is_file():
+        raise RunnerError("verified discovery state has no intact archive for %s" % name)
+    try:
+        current_hash = _sha256_text(path.read_text(encoding="utf-8").strip())
+    except (OSError, UnicodeDecodeError) as exc:
+        raise RunnerError("verified discovery archive cannot be read: %s (%s)" % (path, exc)) from exc
+    if current_hash != digest:
+        raise RunnerError("verified discovery archive changed on disk: %s" % path)
+
+
 def _dedupe(items: Iterable[object]) -> List[str]:
     seen = set()
     result: List[str] = []
@@ -594,16 +609,24 @@ def load_or_create_state(job: JobFile) -> Dict[str, object]:
         ]
         if list(state.get("concept_references") or []) != expected_references:
             raise RunnerError("state concept references do not match the discovery input")
+        seen_discovery_names = set()
         for expected_index, row in enumerate(rows, start=1):
             if not isinstance(row, dict) or int(row.get("index") or 0) != expected_index:
                 raise RunnerError("discovery state contains an invalid object record")
+            status = str(row.get("status") or "")
+            if status not in DISCOVERY_ROW_STATUSES:
+                raise RunnerError("unknown discovery state status '%s'" % status)
             if str(row.get("project_slug") or "") != _discovery_project_slug(job, expected_index):
                 raise RunnerError("discovery state project association is inconsistent")
             if not str(row.get("session_id") or ""):
                 raise RunnerError("discovery state contains an unbound session record")
             name = str(row.get("name") or "").strip()
             archive = str(row.get("archive") or "").strip()
             if name:
+                normalized_name = name.casefold()
+                if normalized_name in seen_discovery_names:
+                    raise RunnerError("duplicate discovery object name: %s" % name)
+                seen_discovery_names.add(normalized_name)
                 branch = str(row.get("branch") or "").strip()
                 if branch not in job.branches:
                     raise RunnerError("discovery state contains an invalid branch for %s" % name)
@@ -621,8 +644,16 @@ def load_or_create_state(job: JobFile) -> Dict[str, object]:
                 )
                 if archive != expected_archive:
                     raise RunnerError("state archive association is inconsistent for %s" % name)
+                if status == "verified":
+                    _validate_discovery_verified_archive(
+                        name,
+                        archive,
+                        str(row.get("archive_sha256") or ""),
+                    )
             elif archive:
                 raise RunnerError("discovery state has an archive path without an object name")
+            elif status == "verified":
+                raise RunnerError("verified discovery state has no intact archive")
     return state
 
 
@@ -874,6 +905,7 @@ def verify_archive(
                 raise ValueError("object_name is required for branch discovery")
             if not selected_branch:
                 raise ValueError("branch must be one of the supplied mathematical branches")
+            selected_name = _ensure_discovery_name_unattempted(selected_name)
         else:
             selected_name = object_job.name
             selected_branch = object_job.branch
@@ -1253,6 +1285,17 @@ def _attempted_object_names() -> List[str]:
     return sorted(names.values(), key=str.casefold)
 
 
+def _ensure_discovery_name_unattempted(name: str) -> str:
+    """Fail closed when a new discovery reuses an exact prior attempted name."""
+    candidate = str(name or "").strip()
+    if not candidate:
+        raise RunnerError("discovery object name is empty")
+    attempted = {item.casefold() for item in _attempted_object_names()}
+    if candidate.casefold() in attempted:
+        raise RunnerError("discovery object was already attempted: %s" % candidate)
+    return candidate
+
+
 def _discovery_project_slug(job: JobFile, index: int) -> str:
     return "math-object-archive-%s-%03d" % (slugify(job.key, prefix="batch"), index)
 
@@ -1743,6 +1786,7 @@ def publish_accepted(
             name = candidate_map.get(name.casefold(), "")
             if not name:
                 raise RunnerError("verified object is outside the supplied candidate pool")
+        name = _ensure_discovery_name_unattempted(name)
         row["name"] = name
         row["branch"] = branch
         row["source_urls"] = []
@@ -1981,6 +2025,8 @@ def publish_accepted(
                         for candidate_name, _source in job.concept_references
                     }
                     name = candidate_map.get(name.casefold(), "")
+                if name:
+                    name = _ensure_discovery_name_unattempted(name)
                 branch_map = {branch.casefold(): branch for branch in job.branches}
                 branch = branch_map.get(str(control.get("branch") or "").strip().casefold(), "")
                 if name and branch:

tests/test_discovery_state_integrity.py

@@ -0,0 +1,245 @@
+"""Regression coverage for branch-discovery lifecycle integrity.
+
+These tests pin fail-closed contracts for resumable discovery state, verified
+artifact integrity, and duplicate-safe exact-name selection without requiring a
+real Moonshine runtime or model provider.
+"""
+
+from __future__ import annotations
+
+import tempfile
+import types
+import unittest
+from pathlib import Path
+
+from tests.test_run_archive_offline import _load_runner_module, _stub_write_json
+
+
+PASS_REVIEW = {
+    "mathematical": {"verdict": "pass", "issues": [], "rationale": "fixture"},
+    "historical": {"verdict": "pass", "issues": [], "rationale": "fixture"},
+    "format": {"verdict": "pass", "issues": [], "rationale": "fixture"},
+    "repair_targets": [],
+    "summary": "fixture passed",
+}
+
+
+class _VerificationProvider:
+    def __init__(self):
+        self.calls = 0
+
+    def generate_structured(self, **_kwargs):
+        self.calls += 1
+        return PASS_REVIEW
+
+
+class _Registry:
+    def __init__(self):
+        self.definition = None
+
+    def register(self, definition):
+        self.definition = definition
+
+
+class DiscoveryStateIntegrityTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.runner, cls._module_patcher = _load_runner_module()
+
+    @classmethod
+    def tearDownClass(cls):
+        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 _job(self, target_archives=2):
+        return self.runner.build_discovery_job(
+            ["Category Theory"],
+            target_archives=target_archives,
+            run_name="discovery-integrity",
+        )
+
+    def _row(self, job, index, name, status="verified"):
+        return {
+            "index": index,
+            "name": name,
+            "branch": "Category Theory",
+            "source_urls": [],
+            "status": status,
+            "project_slug": self.runner._discovery_project_slug(job, index),
+            "session_id": "session-%s" % index,
+            "archive": str(
+                self.task_dir
+                / "archives"
+                / job.key
+                / ("%03d-%s.md" % (index, self.runner._safe_filename(name, "object")))
+            ),
+            "archive_sha256": self.runner._sha256_text("fixture-%s" % index),
+            "verification_submissions": 1,
+            "failure_stage": "",
+            "last_verification": {},
+            "last_error": "",
+        }
+
+    def _persist_state(self, job, rows):
+        state = self.runner.load_or_create_state(job)
+        state["objects"] = rows
+        _stub_write_json(job.state_path, state)
+        return state
+
+    def _register_verifier(self, *, discovery=True):
+        registry = _Registry()
+        app = types.SimpleNamespace(tool_registry=registry)
+        shell_state = self.runner.ShellState(
+            mode="chat",
+            project_slug="discovery-project",
+            session_id="discovery-session",
+            agent_slug=self.runner.AGENT_SLUG,
+        )
+        object_job = self.runner.ObjectJob(
+            index=1,
+            name="" if discovery else "Yoneda lemma",
+            materials=(),
+            project_slug=shell_state.project_slug,
+            archive_path=self.task_dir / "pending.md",
+            branch="" if discovery else "Category Theory",
+        )
+        self.runner.register_verification_tool(
+            app,
+            object_job=object_job,
+            shell_state=shell_state,
+            format_specification="No fenced placeholders.",
+            material_context="(none)",
+            discovery_branches=["Category Theory"] if discovery else (),
+        )
+        self.assertIsNotNone(registry.definition)
+        return registry.definition.handler, shell_state
+
+    def _write_attempt_history(self, name, status):
+        path = self.task_dir / "runs" / ("history-%s.state.json" % status)
+        _stub_write_json(
+            path,
+            {
+                "schema_version": self.runner.STATE_SCHEMA_VERSION,
+                "objects": [{"name": name, "status": status}],
+            },
+        )
+
+    def test_discovery_state_rejects_unknown_row_status(self):
+        job = self._job(target_archives=1)
+        self._persist_state(job, [self._row(job, 1, "Yoneda lemma", status="corrupted")])
+
+        with self.assertRaisesRegex(self.runner.RunnerError, "unknown.*status|invalid.*status"):
+            self.runner.load_or_create_state(job)
+
+    def test_discovery_state_rejects_case_insensitive_duplicate_object_names(self):
+        job = self._job(target_archives=2)
+        self._persist_state(
+            job,
+            [
+                self._row(job, 1, "Yoneda lemma", status="failed"),
+                self._row(job, 2, "yoneda LEMMA", status="verified"),
+            ],
+        )
+
+        with self.assertRaisesRegex(self.runner.RunnerError, "duplicate.*object|duplicate.*name"):
+            self.runner.load_or_create_state(job)
+
+    def test_discovery_state_rejects_missing_verified_archive(self):
+        job = self._job(target_archives=1)
+        self._persist_state(job, [self._row(job, 1, "Yoneda lemma", status="verified")])
+
+        with self.assertRaisesRegex(self.runner.RunnerError, "verified.*archive|archive.*missing|intact"):
+            self.runner.load_or_create_state(job)
+
+    def test_discovery_state_rejects_tampered_verified_archive(self):
+        job = self._job(target_archives=1)
+        row = self._row(job, 1, "Yoneda lemma", status="verified")
+        archive = Path(row["archive"])
+        archive.parent.mkdir(parents=True, exist_ok=True)
+        archive.write_text("tampered\n", encoding="utf-8")
+        self._persist_state(job, [row])
+
+        with self.assertRaisesRegex(self.runner.RunnerError, "verified.*changed|hash|integrity"):
+            self.runner.load_or_create_state(job)
+
+    def test_discovery_state_accepts_intact_verified_archive(self):
+        job = self._job(target_archives=1)
+        row = self._row(job, 1, "Yoneda lemma", status="verified")
+        archive = Path(row["archive"])
+        archive.parent.mkdir(parents=True, exist_ok=True)
+        archive.write_text("fixture-1\n", encoding="utf-8")
+        self._persist_state(job, [row])
+
+        loaded = self.runner.load_or_create_state(job)
+        self.assertEqual(loaded["objects"][0]["status"], "verified")
+
+    def test_discovery_verifier_rejects_previously_attempted_exact_name(self):
+        for prior_status in ("verified", "failed"):
+            with self.subTest(prior_status=prior_status):
+                self._write_attempt_history("Yoneda lemma", prior_status)
+                handler, shell_state = self._register_verifier(discovery=True)
+                provider = _VerificationProvider()
+                runtime = {
+                    "project_slug": shell_state.project_slug,
+                    "session_id": shell_state.session_id,
+                    "verification_provider": provider,
+                }
+                with self.assertRaisesRegex(
+                    self.runner.RunnerError,
+                    "already attempted|duplicate|previously attempted",
+                ):
+                    handler(
+                        runtime,
+                        archive="# Archive | yoneda LEMMA\n\nfixture",
+                        object_name="yoneda LEMMA",
+                        branch="Category Theory",
+                    )
+                self.assertEqual(provider.calls, 0)
+
+    def test_discovery_verifier_allows_unattempted_name(self):
+        self._write_attempt_history("Yoneda lemma", "verified")
+        handler, shell_state = self._register_verifier(discovery=True)
+        provider = _VerificationProvider()
+        runtime = {
+            "project_slug": shell_state.project_slug,
+            "session_id": shell_state.session_id,
+            "verification_provider": provider,
+        }
+
+        result = handler(
+            runtime,
+            archive="# Archive | Kan extension\n\nfixture",
+            object_name="Kan extension",
+            branch="Category Theory",
+        )
+
+        self.assertTrue(result["passed"])
+        self.assertEqual(provider.calls, 1)
+
+    def test_non_discovery_verifier_keeps_existing_failed_repair_path_available(self):
+        self._write_attempt_history("Yoneda lemma", "failed")
+        handler, shell_state = self._register_verifier(discovery=False)
+        provider = _VerificationProvider()
+        runtime = {
+            "project_slug": shell_state.project_slug,
+            "session_id": shell_state.session_id,
+            "verification_provider": provider,
+        }
+
+        result = handler(runtime, archive="# Archive | Yoneda lemma\n\nfixture")
+
+        self.assertTrue(result["passed"])
+        self.assertEqual(provider.calls, 1)
+
+
+if __name__ == "__main__":
+    unittest.main()