[guard.research.ρ.D8+] prove the durable loop across the real dispatch fence

The first real PostgreSQL acceptance stopped before provider I/O because migration 047 admitted only campaign preparation Jobs; memory-backed controller tests could not reveal that database-enforced boundary. Admit only research.loop.create as the second fenced dispatcher, require the new migration at startup, and keep shutdown from resurrecting timer work. The dynamic fake provider, repeatable PostgreSQL/security tests, and acceptance harness belong to the same action because they are the external witness that this exact fence change yields one deduplicated create-to-approval-to-RunSet-to-refresh-to-stop loop while secrets and cross-tenant artifacts remain inaccessible. Real Qwen endpoints and production migration remain deliberately outside this commit.

Author: ivanicu

backend/db/migrations/050_research_loop_dispatch.sql

@@ -0,0 +1,83 @@
+-- 050 · Admit the research-loop reasoner into the durable local dispatch fence.
+--
+-- 047 deliberately admitted only physics.rbfe-campaign.prepare.  The research
+-- loop submits ai.research.propose under research.loop.create with the same
+-- client-key/outbox/fencing protocol; without this explicit second command the
+-- loop reaches its first context snapshot and then blocks before provider I/O.
+BEGIN;
+
+CREATE OR REPLACE FUNCTION app.enforce_job_dispatch_fence()
+RETURNS trigger
+LANGUAGE plpgsql
+AS $$
+BEGIN
+    IF TG_OP = 'DELETE' THEN
+        IF EXISTS (
+            SELECT 1 FROM app.job j
+            WHERE j.id = OLD.job_id AND j.state IN ('queued','running')
+        ) THEN
+            RAISE EXCEPTION USING
+                ERRCODE = '23514',
+                CONSTRAINT = 'job_dispatch_active_delete_forbidden',
+                MESSAGE = 'an active Job dispatch may only leave through terminal cleanup';
+        END IF;
+        RETURN OLD;
+    END IF;
+    IF TG_OP = 'INSERT' AND NOT EXISTS (
+        SELECT 1
+        FROM app.job j
+        WHERE j.id = NEW.job_id
+          AND j.request_key IS NOT NULL
+          AND j.command_id IN (
+              'physics.rbfe-campaign.prepare',
+              'research.loop.create')
+          AND j.input_sha256 = NEW.payload_sha256
+          AND j.request_digest = NEW.execution_digest
+          AND j.state IN ('queued','running')
+    ) THEN
+        RAISE EXCEPTION USING
+            ERRCODE = '23514',
+            CONSTRAINT = 'job_dispatch_requires_keyed_active_job',
+            MESSAGE = 'job_dispatch requires an allow-listed active client-key Job';
+    END IF;
+    IF TG_OP = 'UPDATE' AND NEW.fencing_token < OLD.fencing_token THEN
+        RAISE EXCEPTION USING
+            ERRCODE = '23514',
+            CONSTRAINT = 'job_dispatch_token_monotone',
+            MESSAGE = 'job_dispatch fencing_token may not decrease';
+    END IF;
+    IF TG_OP = 'UPDATE' AND (
+        NEW.job_id IS DISTINCT FROM OLD.job_id
+        OR NEW.payload IS DISTINCT FROM OLD.payload
+        OR NEW.payload_sha256 IS DISTINCT FROM OLD.payload_sha256
+        OR NEW.execution_digest IS DISTINCT FROM OLD.execution_digest
+        OR NEW.execution_adapter IS DISTINCT FROM OLD.execution_adapter
+    ) THEN
+        RAISE EXCEPTION USING
+            ERRCODE = '23514',
+            CONSTRAINT = 'job_dispatch_witness_immutable',
+            MESSAGE = 'job_dispatch admitted witness columns are immutable';
+    END IF;
+    IF TG_OP = 'UPDATE' AND (
+        NEW.lease_owner IS DISTINCT FROM OLD.lease_owner
+        OR NEW.claimed_at IS DISTINCT FROM OLD.claimed_at
+        OR (OLD.state = 'running' AND NEW.state = 'claimed')
+    ) AND NEW.fencing_token <= OLD.fencing_token THEN
+        RAISE EXCEPTION USING
+            ERRCODE = '23514',
+            CONSTRAINT = 'job_dispatch_reclaim_requires_new_token',
+            MESSAGE = 'dispatch ownership/reclaim changes require a newer token';
+    END IF;
+    RETURN NEW;
+END;
+$$;
+
+COMMENT ON FUNCTION app.enforce_job_dispatch_fence() IS
+    'Rejects non-allow-listed/non-keyed witnesses, mutation, active deletion, and token rollback.';
+
+INSERT INTO meta.migration (filename,sha256,content_sha256,hash_source)
+VALUES ('050_research_loop_dispatch.sql','\x5a923ab42cbd8a64f9117319e2823bdf38bd7ab2d305c544ea8ab90cb637c0b2'::bytea,
+        '\x5a923ab42cbd8a64f9117319e2823bdf38bd7ab2d305c544ea8ab90cb637c0b2'::bytea,'content')
+ON CONFLICT (filename) DO NOTHING;
+
+COMMIT;

backend/kernel.py

@@ -532,10 +532,16 @@ def build(*, dsn: str = DEFAULT_DSN, with_versions: bool = True,
                 "SELECT to_regclass('app.research_loop_state') IS NOT NULL, "
                 "to_regclass('app.research_loop_event') IS NOT NULL, "
                 "to_regclass('app.research_loop_approval') IS NOT NULL, "
-                "to_regclass('app.research_loop_artifact') IS NOT NULL")
+                "to_regclass('app.research_loop_artifact') IS NOT NULL, "
+                "EXISTS (SELECT 1 FROM meta.migration "
+                " WHERE filename='050_research_loop_dispatch.sql'), "
+                "position('research.loop.create' in pg_get_functiondef("
+                " 'app.enforce_job_dispatch_fence()'::regprocedure)) > 0")
             loop_capability = cur.fetchone()
         if loop_capability is None or not all(loop_capability):
-            raise RuntimeError('migration 049_research_loop.sql is not fully applied')
+            raise RuntimeError(
+                'migrations 049_research_loop.sql and '
+                '050_research_loop_dispatch.sql are not fully applied')
         research_loops = ResearchLoopController(
             repository=ResearchLoopRepository(connect), service=svc,
             artifact_store=st, provider_registry=ai_providers,

backend/research/loop_controller.py

@@ -48,6 +48,7 @@ def __init__(
         self._lock = threading.Lock()
         self._running = False
         self._wake_requested = False
+        self._shutdown = False
         self._timer: threading.Timer | None = None
 
     def create(self, payload: Mapping[str, Any], actor: Mapping[str, Any]) -> dict[str, Any]:
@@ -241,6 +242,8 @@ def control(self, payload: Mapping[str, Any], actor: Mapping[str, Any]) -> dict[
 
     def wake(self) -> None:
         with self._lock:
+            if self._shutdown:
+                return
             if self._timer is not None:
                 self._timer.cancel()
                 self._timer = None
@@ -252,6 +255,8 @@ def wake(self) -> None:
 
     def shutdown(self) -> None:
         with self._lock:
+            self._shutdown = True
+            self._wake_requested = False
             if self._timer is not None:
                 self._timer.cancel()
                 self._timer = None
@@ -274,11 +279,12 @@ def _drain(self) -> None:
                 requested = self._wake_requested
                 self._wake_requested = False
                 self._running = False
-                if not requested and delay is not None:
+                shutdown = self._shutdown
+                if not shutdown and not requested and delay is not None:
                     self._timer = threading.Timer(max(0.05, delay), self.wake)
                     self._timer.daemon = True
                     self._timer.start()
-            if requested:
+            if requested and not shutdown:
                 self.wake()
 
     def _advance(self, claim: LoopClaim) -> None:

backend/tests/test_openai_compatible_provider.py

@@ -64,6 +64,29 @@ def test_valid_response_returns_only_bounded_public_fields(self):
         self.assertNotIn("reasoning", repr(result))
         self.assertNotIn("test-secret-never-leak", repr(result))
 
+    def test_action_mode_binds_to_the_request_context_digest(self):
+        self.server.mode = "action"
+        result = self.provider.complete_json(
+            self.profile,
+            system_prompt="Return one bounded JSON object and never call tools.",
+            context_json=("JSON research context:\n" + json.dumps({
+                "research_context": {
+                    "digest": "sha256:" + "9" * 64,
+                    "facts": [],
+                    "available_actions": [{
+                        "template_id": "fep.run_selected_edge.v1",
+                        "subject_refs": [{
+                            "kind": "free_energy_transformation", "id": "edge-a-b",
+                        }],
+                    }],
+                },
+            })),
+        )
+        proposal = json.loads(result.content)
+        self.assertEqual(proposal["context_digest"], "sha256:" + "9" * 64)
+        self.assertEqual(proposal["candidate_actions"][0]["subject_ref"]["id"],
+                         "edge-a-b")
+
     def test_reasoning_content_is_discarded(self):
         self.server.mode = "reasoning"
         result = self.complete()

backend/tests/test_research_loop_controller.py

@@ -171,6 +171,17 @@ def test_prepare_prerequisite_resumes_run_without_second_reasoner_call(self):
         self.assertNotIn("resume_candidate_after_snapshot",
                          snapshot_transition["updates"]["outputs"])
 
+    def test_shutdown_prevents_a_late_timer_from_resubmitting_work(self):
+        controller = self.controller()
+        controller._lock = __import__("threading").Lock()
+        controller._shutdown = True
+        controller._running = False
+        controller._wake_requested = False
+        controller._timer = None
+        controller._pool = SimpleNamespace(
+            submit=lambda *_args: self.fail("shutdown controller submitted work"))
+        controller.wake()
+
 
 if __name__ == "__main__":
     unittest.main()

backend/tests/test_research_loop_repository.py

@@ -24,6 +24,7 @@
 
 ROOT = pathlib.Path(__file__).resolve().parents[2]
 MIGRATION = ROOT / "backend/db/migrations/049_research_loop.sql"
+DISPATCH_MIGRATION = ROOT / "backend/db/migrations/050_research_loop_dispatch.sql"
 
 
 class ResearchLoopMigrationContractTests(unittest.TestCase):
@@ -51,18 +52,35 @@ def test_recorded_migration_hash_matches_normalized_file(self):
         normalized = source.replace(recorded, "PENDING")
         self.assertEqual(hashlib.sha256(normalized.encode()).hexdigest(), recorded)
 
+    def test_forward_dispatch_migration_admits_only_prepare_and_research_loop(self):
+        source = DISPATCH_MIGRATION.read_text(encoding="utf-8")
+        self.assertIn("'physics.rbfe-campaign.prepare'", source)
+        self.assertIn("'research.loop.create'", source)
+        self.assertIn("j.command_id IN", source)
+        marker = "VALUES ('050_research_loop_dispatch.sql','\\x"
+        recorded = source.split(marker, 1)[1][:64]
+        normalized = source.replace(recorded, "PENDING")
+        self.assertEqual(hashlib.sha256(normalized.encode()).hexdigest(), recorded)
+
 
 @unittest.skipUnless(os.environ.get("DIRAC_TEST_DSN") and PG_DRIVER,
-                     "requires isolated PostgreSQL DIRAC_TEST_DSN with migrations 000-049")
+                     "requires isolated PostgreSQL DIRAC_TEST_DSN with migrations 000-050")
 class ResearchLoopRepositoryPostgresTests(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
         cls.dsn = os.environ["DIRAC_TEST_DSN"]
         cls.connect = staticmethod(lambda: PG_DRIVER.connect(cls.dsn))
         cls.repository = ResearchLoopRepository(cls.connect)
-        cls.actor = {"kind": "human", "id": "chemist-loop-test"}
         with cls.connect() as conn, conn.cursor() as cur:
+            cur.execute("SELECT current_database()")
+            if "test" not in str(cur.fetchone()[0]).lower():
+                raise RuntimeError("repository tests refuse a database without 'test' in its name")
+            cur.execute(
+                "UPDATE app.research_loop_state SET state='cancelled',stage='completed',"
+                "lease_owner=NULL,lease_expires_at=NULL,finished_at=coalesce(finished_at,now()) "
+                "WHERE state IN ('active','waiting_approval','blocked','paused')")
             code = "RL-" + uuid.uuid4().hex[:12]
+            cls.actor = {"kind": "human", "id": code}
             cur.execute(
                 "INSERT INTO design.project(code,name) VALUES (%s,%s) RETURNING id",
                 (code, "Research loop repository test"),
@@ -154,6 +172,8 @@ def test_durable_create_lease_transition_approval_and_restart_read(self):
             cur.execute(
                 "INSERT INTO app.artifact(blob_sha256,media_type,role,size_bytes) "
                 "VALUES (%s,'application/json','research.action_preview',%s) "
+                "ON CONFLICT (blob_sha256,role,encoding) DO UPDATE "
+                "SET role=EXCLUDED.role "
                 "RETURNING id", (preview_digest, len(preview)),
             )
             preview_artifact_id = str(cur.fetchone()[0])

backend/tests/test_research_loop_security.py

@@ -0,0 +1,132 @@
+from __future__ import annotations
+
+import json
+import os
+import pathlib
+import tempfile
+import unittest
+import uuid
+
+import failures
+from artifacts_pg import PostgresArtifactStore
+from research.loop_repository import ResearchLoopRepository
+from research.provider_registry import FileAiProviderRegistry
+
+try:
+    import psycopg
+except ImportError:  # pragma: no cover - integration environment decides
+    psycopg = None
+
+
+ROOT = pathlib.Path(__file__).resolve().parents[2]
+EXAMPLE = json.loads(
+    (ROOT / "deploy/ai/providers.example.json").read_text(encoding="utf-8")
+)
+
+
+@unittest.skipUnless(os.environ.get("DIRAC_TEST_DSN") and psycopg,
+                     "requires isolated PostgreSQL DIRAC_TEST_DSN with migrations 000-050")
+class ResearchLoopSecurityTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.dsn = os.environ["DIRAC_TEST_DSN"]
+        cls.connect = staticmethod(lambda: psycopg.connect(cls.dsn))
+        with cls.connect() as conn, conn.cursor() as cur:
+            cur.execute("SELECT current_database()")
+            if "test" not in str(cur.fetchone()[0]).lower():
+                raise RuntimeError("security tests refuse a database without 'test' in its name")
+            code = "SEC-" + uuid.uuid4().hex[:10]
+            cls.actor = {"kind": "human", "id": code}
+            cur.execute(
+                "INSERT INTO design.project(code,name) VALUES (%s,%s) RETURNING id",
+                (code, "Research-loop security test"),
+            )
+            cls.program_id = str(cur.fetchone()[0])
+            cur.execute(
+                "INSERT INTO design.campaign "
+                "(program_id,name,objective,created_by_kind,created_by_id) "
+                "VALUES (%s,%s,%s,'human',%s) RETURNING id",
+                (cls.program_id, code, "verify private loop surfaces", code),
+            )
+            cls.campaign_id = str(cur.fetchone()[0])
+        cls.repository = ResearchLoopRepository(cls.connect)
+        cls.loop = cls.repository.create(
+            request_key="security-loop", program_id=cls.program_id,
+            campaign_id=cls.campaign_id, actor=cls.actor,
+            intent="Verify tenant isolation and provider-secret redaction",
+            autonomy_class="A2", provider_profile_id="qwen-local-isolated",
+            provider_profile_digest=b"p" * 32,
+            prompt_release_id="fep-action-proposal-v1",
+            prompt_release_digest=b"r" * 32,
+            action_catalog_digest=b"a" * 32, data_classification="restricted",
+            policy={"cloud_egress_approved": False},
+            budget_remaining={"reasoner_calls": 1, "fep_runsets": 0},
+        )
+        cls.store = PostgresArtifactStore(cls.connect)
+        cls.artifact = cls.store.put(
+            b'{"claim_boundary":"model_proposal_not_scientific_evidence"}',
+            role="research.context_snapshot", media_type="application/json",
+            metadata={"visibility": "private"},
+        )
+        cls.repository.link_artifact(
+            run_id=cls.loop["run_id"], artifact_id=cls.artifact.id,
+            role="research.context_snapshot", data_classification="restricted",
+        )
+        with cls.connect() as conn, conn.cursor() as cur:
+            cur.execute(
+                "UPDATE app.research_loop_state SET state='paused' WHERE run_id=%s",
+                (cls.loop["run_id"],),
+            )
+
+    def test_loop_artifact_uuid_and_digest_are_not_authorization_capabilities(self):
+        owner = self.store.read_authorized(self.artifact.id, self.actor)
+        self.assertEqual(owner[1],
+                         b'{"claim_boundary":"model_proposal_not_scientific_evidence"}')
+        intruder = {"kind": "human", "id": "intruder-" + uuid.uuid4().hex}
+        for address in (self.artifact.id, "sha256:" + self.artifact.sha256):
+            with self.subTest(address=address), self.assertRaises(failures.DiracNotFound):
+                self.store.read_authorized(address, intruder)
+
+    def test_provider_secret_is_absent_from_every_loop_persisted_surface(self):
+        secret = "research-loop-secret-" + uuid.uuid4().hex
+        profile = dict(EXAMPLE["profiles"][1])
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".json", encoding="utf-8") as source:
+            json.dump({"profiles": [profile]}, source)
+            source.flush()
+            registry = FileAiProviderRegistry(source.name, environ={
+                "DIRAC_LOCAL_QWEN_BASE_URL": "http://127.0.0.1:8000/v1",
+                "DIRAC_LOCAL_QWEN_API_KEY": secret,
+            })
+            resolved = registry.resolve("qwen-local-isolated")
+            public_surfaces = {
+                "repr": repr(resolved),
+                "public": resolved.to_public_dict(),
+                "attestation": registry.attest(
+                    resolved.profile_id, resolved.profile_digest, "restricted"),
+            }
+        self.assertNotIn(secret, json.dumps(public_surfaces, sort_keys=True))
+
+        with self.connect() as conn, conn.cursor() as cur:
+            cur.execute(
+                "SELECT row_to_json(ls)::text FROM app.research_loop_state ls "
+                "WHERE run_id=%s UNION ALL "
+                "SELECT row_to_json(le)::text FROM app.research_loop_event le "
+                "WHERE run_id=%s UNION ALL "
+                "SELECT row_to_json(la)::text FROM app.research_loop_artifact la "
+                "WHERE run_id=%s",
+                (self.loop["run_id"],) * 3,
+            )
+            persisted = "\n".join(str(row[0]) for row in cur.fetchall())
+            cur.execute(
+                "SELECT b.bytes,a.metadata::text FROM app.research_loop_artifact la "
+                "JOIN app.artifact a ON a.id=la.artifact_id "
+                "JOIN app.blob b ON b.sha256=a.blob_sha256 WHERE la.run_id=%s",
+                (self.loop["run_id"],),
+            )
+            for raw, metadata in cur.fetchall():
+                persisted += bytes(raw).decode("utf-8", errors="replace") + str(metadata)
+        self.assertNotIn(secret, persisted)
+
+
+if __name__ == "__main__":
+    unittest.main()