[guard.research.ρ.D8+] close crash windows and preview authority gaps

The durable loop had stable request-key machinery, but the exact side-effect-before-state-commit windows were not executable evidence, and persisted previews were only rechecked for freshness rather than against their immutable Artifact, frozen template mapping, current command contract, and remaining budget. This could turn an accidental persisted mutation into authority the model never had.

Add restart fault injection around reason submission, provider completion, preview persistence, command Job persistence, and physical RunSet persistence; prove repeated attempts reuse one provider/dispatch identity. Preserve negative/refused history, prohibit stale or non-typed facts from claiming evidence eligibility in the schema, and freeze the updated prompt digest. Dispatch now rejects preview digest, template/command, consequence, acknowledgement, command schema/version, budget, and fingerprint drift. Repair latent DiracInternal construction on schema failure so fail-closed paths remain typed instead of throwing TypeError.

Author: ivanicu

backend/research/action_compiler.py

@@ -162,9 +162,9 @@ def compile(
             self.preview_validator.iter_errors(preview), key=lambda error: list(error.path))
         if errors:
             error = errors[0]
-            raise failures.DiracInternal(
-                "compiled ActionPreview violates its frozen contract",
-                details={"path": list(error.path), "message": error.message})
+            raise failures.DiracInternal(ValueError(
+                "compiled ActionPreview violates its frozen contract at "
+                f"{list(error.path)}: {error.message}"))
         encoded = canonical_bytes(preview)
         return CompiledAction(
             preview=preview, preview_bytes=encoded,
@@ -184,6 +184,16 @@ def revalidate(
         clock = now or datetime.now(timezone.utc)
         expected_acks = set(preview.get("required_acknowledgements") or [])
         failures_seen = []
+        preview_errors = sorted(
+            self.preview_validator.iter_errors(preview),
+            key=lambda error: list(error.path),
+        )
+        if preview_errors:
+            failures_seen.append("preview_contract")
+        expected_preview_digest = compiled.get("preview_artifact_sha256")
+        if (not isinstance(expected_preview_digest, str)
+                or canonical_digest(dict(preview)) != expected_preview_digest):
+            failures_seen.append("preview_artifact_digest")
         preview_version = int(preview.get("loop_version", -1))
         loop_version = int(loop["version"])
         approved_per_action = (
@@ -205,9 +215,56 @@ def revalidate(
             if actor.get("kind") != "human" or not str(actor.get("id") or "").strip():
                 failures_seen.append("human_actor")
         resolved = preview.get("resolved_command")
-        if resolved and canonical_digest(dict(compiled.get("command_input") or {})) != resolved.get(
-                "input_digest"):
-            failures_seen.append("command_input_digest")
+        template = self.actions.get(str(preview.get("template_id") or ""))
+        if template is None:
+            failures_seen.append("action_template")
+        else:
+            expected_command_id = template["execution"]["command_id"]
+            actual_command_id = (
+                resolved.get("command_id") if isinstance(resolved, Mapping) else None)
+            if actual_command_id != expected_command_id:
+                failures_seen.append("resolved_command")
+            consequence = preview.get("consequence") or {}
+            expected_consequence = template["consequence"]
+            if any(consequence.get(key) != expected_consequence[key]
+                   for key in ("risk_class", "approval", "reversible")):
+                failures_seen.append("consequence_policy")
+            expected_required_acks = (
+                {"physical_fep_compute", "completed_unvalidated_claim_boundary"}
+                if expected_consequence["risk_class"] == "R3" else set())
+            if expected_acks != expected_required_acks:
+                failures_seen.append("required_acknowledgements")
+            try:
+                self._check_budget(
+                    loop, preview.get("estimate") or {}, str(preview["template_id"]))
+            except failures.DiracBudgetExceeded:
+                failures_seen.append("budget")
+        command_input = dict(compiled.get("command_input") or {})
+        if resolved:
+            if canonical_digest(command_input) != resolved.get("input_digest"):
+                failures_seen.append("command_input_digest")
+            try:
+                definition = self.commands.get(str(resolved.get("command_id") or ""))
+                self.commands.validate_input(definition, command_input)
+                if definition.version != resolved.get("command_version"):
+                    failures_seen.append("command_version")
+            except failures.DiracFailure:
+                failures_seen.append("command_contract")
+        elif command_input:
+            failures_seen.append("non_executing_command_input")
+        fingerprint_payload = {
+            "template_id": preview.get("template_id"),
+            "subject_ref": preview.get("subject_ref"),
+            "scientific_question": preview.get("scientific_question"),
+            "command_version": (
+                resolved.get("command_version") if isinstance(resolved, Mapping) else None),
+            "input_digest": (
+                resolved.get("input_digest") if isinstance(resolved, Mapping)
+                else canonical_digest({})),
+            "source_versions": dict(preview.get("source_versions") or {}),
+        }
+        if canonical_digest(fingerprint_payload) != preview.get("action_fingerprint"):
+            failures_seen.append("action_fingerprint")
         if failures_seen:
             raise failures.DiracStalePreview(
                 "scientific context changed after the action preview was created",

backend/research/context_builder.py

@@ -61,9 +61,9 @@ def build(self, loop: Mapping[str, Any],
         }
         missing = sorted(required_domain - set(domain))
         if missing:
-            raise failures.DiracInternal(
-                "FEP adapter returned an incomplete research context",
-                details={"missing": missing})
+            raise failures.DiracInternal(ValueError(
+                "FEP adapter returned an incomplete research context; missing="
+                + ",".join(missing)))
         facts = self._ordered_facts(domain["facts"])
         base = {
             "schema_version": "1.0",
@@ -128,9 +128,9 @@ def _validate(self, document: Mapping[str, Any]) -> None:
         errors = sorted(self.validator.iter_errors(document), key=lambda error: list(error.path))
         if errors:
             first = errors[0]
-            raise failures.DiracInternal(
-                "deterministic FEP context violates its frozen schema",
-                details={"path": list(first.path), "message": first.message})
+            raise failures.DiracInternal(ValueError(
+                "deterministic FEP context violates its frozen schema at "
+                f"{list(first.path)}: {first.message}"))
 
     @staticmethod
     def _ordered_facts(value: Any) -> list[dict[str, Any]]:

backend/research/loop_summary.py

@@ -56,7 +56,7 @@ def build(
                         key=lambda error: list(error.path))
         if errors:
             first = errors[0]
-            raise failures.DiracInternal(
-                "research loop summary violates its frozen schema",
-                details={"path": list(first.path), "message": first.message})
+            raise failures.DiracInternal(ValueError(
+                "research loop summary violates its frozen schema at "
+                f"{list(first.path)}: {first.message}"))
         return canonical_bytes(document)

backend/tests/test_research_action_catalog.py

@@ -25,6 +25,18 @@ def test_catalog_is_bounded_complete_and_model_projection_hides_commands(self):
         projection = json.dumps(catalog.to_model_catalog())
         self.assertNotIn("command_id", projection)
         self.assertNotIn("physics.rbfe", projection)
+        resolved_commands = {
+            template["execution"]["command_id"]
+            for template in catalog.values()
+            if template["execution"]["command_id"] is not None
+        }
+        self.assertEqual(resolved_commands, {
+            "physics.rbfe-run.start",
+            "physics.rbfe-system.prepare",
+            "physics.rbfe-network",
+        })
+        self.assertFalse(any(command.startswith("program.")
+                             for command in resolved_commands))
 
     def test_registry_refuses_unlisted_or_id_mismatched_templates(self):
         with tempfile.TemporaryDirectory() as temporary:

backend/tests/test_research_action_compiler.py

@@ -96,6 +96,7 @@ def test_stale_source_version_expiry_and_input_tamper_all_fail_closed(self):
             loop=current_loop, context={"digest": "sha256:" + "e" * 64},
             proposal=proposal(), now=datetime(2026, 8, 18, 1, tzinfo=timezone.utc))
         pending = {"preview": dict(compiled.preview),
+                   "preview_artifact_sha256": compiled.preview_digest,
                    "command_input": dict(compiled.command_input)}
         awaiting = {**current_loop, "version": 8}
         compiler.revalidate(
@@ -122,6 +123,60 @@ def test_stale_source_version_expiry_and_input_tamper_all_fail_closed(self):
                     acknowledgements=list(compiled.preview["required_acknowledgements"]),
                     actor={"kind": "human", "id": "chemist"}, now=clock)
 
+    def test_persisted_preview_cannot_be_rewritten_into_a_program_mutation(self):
+        resolver = Resolver()
+        compiler = ActionCompiler(resolver)
+        current_loop = loop()
+        compiled = compiler.compile(
+            loop=current_loop, context={"digest": "sha256:" + "e" * 64},
+            proposal=proposal(), now=datetime(2026, 8, 18, 1, tzinfo=timezone.utc))
+        awaiting = {**current_loop, "version": 8}
+        pending = {
+            "preview": deepcopy(dict(compiled.preview)),
+            "preview_artifact_sha256": compiled.preview_digest,
+            "command_input": dict(compiled.command_input),
+        }
+        pending["preview"]["resolved_command"]["command_id"] = (
+            "program.hypothesis.create")
+
+        with self.assertRaises(failures.DiracStalePreview) as caught:
+            compiler.revalidate(
+                pending, loop=awaiting,
+                current_context_digest="sha256:" + "e" * 64,
+                current_source_versions=resolver.source_versions,
+                acknowledgements=list(compiled.preview["required_acknowledgements"]),
+                actor={"kind": "human", "id": "chemist"},
+                now=datetime(2026, 8, 18, 1, 1, tzinfo=timezone.utc))
+        witnesses = caught.exception.details["stale_witnesses"]
+        self.assertIn("preview_artifact_digest", witnesses)
+        self.assertIn("resolved_command", witnesses)
+        self.assertIn("command_contract", witnesses)
+
+    def test_budget_reduction_invalidates_a_previously_affordable_preview(self):
+        resolver = Resolver()
+        compiler = ActionCompiler(resolver)
+        current_loop = loop()
+        compiled = compiler.compile(
+            loop=current_loop, context={"digest": "sha256:" + "e" * 64},
+            proposal=proposal(), now=datetime(2026, 8, 18, 1, tzinfo=timezone.utc))
+        pending = {
+            "preview": dict(compiled.preview),
+            "preview_artifact_sha256": compiled.preview_digest,
+            "command_input": dict(compiled.command_input),
+        }
+        awaiting = deepcopy({**current_loop, "version": 8})
+        awaiting["budget_remaining"]["fep_runsets"] = 0
+
+        with self.assertRaises(failures.DiracStalePreview) as caught:
+            compiler.revalidate(
+                pending, loop=awaiting,
+                current_context_digest="sha256:" + "e" * 64,
+                current_source_versions=resolver.source_versions,
+                acknowledgements=list(compiled.preview["required_acknowledgements"]),
+                actor={"kind": "human", "id": "chemist"},
+                now=datetime(2026, 8, 18, 1, 1, tzinfo=timezone.utc))
+        self.assertIn("budget", caught.exception.details["stale_witnesses"])
+
     def test_template_outside_frozen_grant_cannot_compile(self):
         current = loop()
         current["policy"]["session_grant"]["allowed_template_ids"] = []

backend/tests/test_research_context_builder.py

@@ -5,6 +5,8 @@
 from copy import deepcopy
 from uuid import uuid4
 
+import failures
+
 from research.context_builder import ContextBuilder
 
 
@@ -115,3 +117,42 @@ def test_untrusted_content_cannot_add_actions_or_change_claim_boundary(self):
         row = next(item for item in built.document["facts"] if item["fact_id"] == "high")
         self.assertFalse(row["claim_boundary"]["eligible_as_scientific_evidence"])
         self.assertNotIn("command_id", json.dumps(built.document["available_actions"]))
+
+    def test_negative_and_human_refused_actions_survive_context_construction(self):
+        source = domain()
+        source["action_history"] = [
+            {
+                "action_fingerprint": "sha256:" + "4" * 64,
+                "template_id": "fep.run_selected_edge.v1",
+                "subject_ref": ref("free_energy_transformation", "edge-negative"),
+                "result": "failed_negative_result",
+                "human_rejected": False,
+            },
+            {
+                "action_fingerprint": "sha256:" + "5" * 64,
+                "template_id": "fep.defer_for_experiment.v1",
+                "subject_ref": ref("free_energy_transformation", "edge-refused"),
+                "result": "rejected",
+                "human_rejected": True,
+            },
+        ]
+
+        built = ContextBuilder().build(loop(), source)
+
+        self.assertEqual(built.document["action_history"], source["action_history"])
+
+    def test_unverified_or_stale_fact_cannot_be_marked_eligible_evidence(self):
+        for source_class, stale in (
+            ("unverified_external", False),
+            ("method_result", False),
+            ("typed_evidence", True),
+        ):
+            with self.subTest(source_class=source_class, stale=stale):
+                source = domain()
+                unsafe = fact("unsafe")
+                unsafe["source_class"] = source_class
+                unsafe["freshness"]["stale"] = stale
+                unsafe["claim_boundary"]["eligible_as_scientific_evidence"] = True
+                source["facts"] = [unsafe]
+                with self.assertRaises(failures.DiracInternal):
+                    ContextBuilder().build(loop(), source)