[feat.motif.Omega.D9+] make target-bound physics and result-driven design executable

WHY: Dirac could schedule GPU work and model a DMTA loop, but it could not prove a real OpenFE edge or turn governed experimental measurements into a durable, recoverable next-round portfolio without crossing its existing Job and Artifact boundaries.

Author: ivanicu

backend/db/migrations/034_motif_closed_loop_controller.sql

@@ -0,0 +1,44 @@
+-- 034 · Durable experimental-result trigger and stage ledger for Motif DMTA loops.
+
+BEGIN;
+
+CREATE TABLE design.motif_closed_loop_run (
+    id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+    request_key text NOT NULL UNIQUE,
+    specification_digest bytea NOT NULL UNIQUE CHECK (octet_length(specification_digest)=32),
+    program_id uuid NOT NULL REFERENCES design.project(id) ON DELETE RESTRICT,
+    campaign_id uuid NOT NULL REFERENCES design.campaign(id) ON DELETE RESTRICT,
+    endpoint_key text NOT NULL,
+    measurement_ids uuid[] NOT NULL CHECK (cardinality(measurement_ids) > 0),
+    specification jsonb NOT NULL,
+    state text NOT NULL DEFAULT 'pending'
+        CHECK (state IN ('pending','running','blocked','completed','cancelled')),
+    stage text NOT NULL DEFAULT 'snapshot'
+        CHECK (stage IN ('snapshot','train','predict','acquire','completed')),
+    stage_jobs jsonb NOT NULL DEFAULT '{}'::jsonb,
+    outputs jsonb NOT NULL DEFAULT '{}'::jsonb,
+    attention jsonb NOT NULL DEFAULT '{}'::jsonb,
+    actor_kind app.actor_kind NOT NULL,
+    actor_id text NOT NULL,
+    created_at timestamptz NOT NULL DEFAULT now(),
+    updated_at timestamptz NOT NULL DEFAULT now(),
+    finished_at timestamptz,
+    CHECK (finished_at IS NULL OR finished_at >= created_at)
+);
+
+CREATE INDEX motif_closed_loop_claim_idx
+    ON design.motif_closed_loop_run(state,updated_at)
+    WHERE state IN ('pending','running');
+CREATE INDEX motif_closed_loop_campaign_idx
+    ON design.motif_closed_loop_run(campaign_id,created_at DESC);
+
+COMMENT ON TABLE design.motif_closed_loop_run IS
+    'Durable, replayable result->snapshot->train->predict->acquire controller state. '
+    'Each scientific stage remains an ordinary Dirac Job; this table owns only orchestration.';
+
+INSERT INTO meta.migration (filename,sha256,content_sha256,hash_source)
+VALUES ('034_motif_closed_loop_controller.sql','\x17feb7ca24780b808b1bea2d34bd1daf6c8bd5fe9a64d9cf02b562990916cfa5'::bytea,
+        '\x17feb7ca24780b808b1bea2d34bd1daf6c8bd5fe9a64d9cf02b562990916cfa5'::bytea,'content')
+ON CONFLICT (filename) DO NOTHING;
+
+COMMIT;

backend/db/migrations/035_motif_closed_loop_retry.sql

@@ -0,0 +1,13 @@
+-- 035 · Make blocked Motif closed loops explicitly and durably retryable.
+
+BEGIN;
+
+ALTER TABLE design.motif_closed_loop_run
+    ADD COLUMN stage_attempts jsonb NOT NULL DEFAULT '{}'::jsonb;
+
+INSERT INTO meta.migration (filename,sha256,content_sha256,hash_source)
+VALUES ('035_motif_closed_loop_retry.sql','\x01da2d20120ea2601c05090bad6d2371a520aca71d8beca06eaa9bbb422e76a0'::bytea,
+        '\x01da2d20120ea2601c05090bad6d2371a520aca71d8beca06eaa9bbb422e76a0'::bytea,'content')
+ON CONFLICT (filename) DO NOTHING;
+
+COMMIT;

backend/dirac_app/handlers.py

@@ -181,6 +181,12 @@ def physics_openmm_md(input: dict, ctx) -> dict:
         actor=ctx.actor, command_id=ctx.command_id)
 
 
+def physics_openfe_edge(input: dict, ctx) -> dict:
+    return ctx.kernel.submit(
+        'physics.motif.openfe_edge', input, request_id=ctx.request_id,
+        actor=ctx.actor, command_id=ctx.command_id)
+
+
 def physics_rbfe_network(input: dict, ctx) -> dict:
     return ctx.kernel.submit(
         'physics.motif.rbfe_network', input, request_id=ctx.request_id,
@@ -228,7 +234,37 @@ def policy_release_register(input: dict, ctx) -> dict:
 
 
 def result_ingest(input: dict, ctx) -> dict:
-    return _motif_governance(ctx).ingest_measurements(input["measurements"], ctx.actor)
+    controller = getattr(ctx.kernel, "closed_loop_controller", None)
+    loop_spec = input.get("closed_loop")
+    if loop_spec is not None:
+        if controller is None:
+            raise failures.DiracFailure(
+                "DB_UNAVAILABLE", "closed-loop controller is unavailable")
+        controller.validate_spec(loop_spec, input["measurements"])
+        controller.validate_context(loop_spec)
+    result = _motif_governance(ctx).ingest_measurements(
+        input["measurements"], ctx.actor)
+    if loop_spec is not None:
+        result["closed_loop"] = controller.enqueue(
+            spec=loop_spec, measurements=input["measurements"],
+            ingest_result=result, actor=ctx.actor)
+    return result
+
+
+def campaign_closed_loop_get(input: dict, ctx) -> dict:
+    controller = getattr(ctx.kernel, "closed_loop_controller", None)
+    if controller is None:
+        raise failures.DiracFailure(
+            "DB_UNAVAILABLE", "closed-loop controller is unavailable")
+    return controller.get(input["run_ref"]["id"])
+
+
+def campaign_closed_loop_retry(input: dict, ctx) -> dict:
+    controller = getattr(ctx.kernel, "closed_loop_controller", None)
+    if controller is None:
+        raise failures.DiracFailure(
+            "DB_UNAVAILABLE", "closed-loop controller is unavailable")
+    return controller.retry(input["run_ref"]["id"])
 
 
 def _programs(ctx):

backend/executors/kubernetes_invocation.py

@@ -160,6 +160,8 @@ def _request(self, ctx: InvocationContext, payload: dict, *, attempt_id: str,
         seed = int(payload.get("seed", 0))
         output_digest = sha256_digest(json.dumps(
             spec.output_schema, sort_keys=True, separators=(",", ":")))
+        scale_profile = spec.execution.get("scale_profile") or {}
+        cpu_cores = float(scale_profile.get("cpu_cores", 4))
         return {
             "schema_version": "1.0",
             "execution_id": str(uuid4()),
@@ -175,7 +177,7 @@ def _request(self, ctx: InvocationContext, payload: dict, *, attempt_id: str,
             "input_manifest_artifact_id": input_id,
             "output_contract_digest": output_digest,
             "resource_request": {
-                "cpu_cores": 4,
+                "cpu_cores": cpu_cores,
                 "memory_bytes": 8 << 30,
                 "gpus": 1,
                 "gpu_arch": ["blackwell"],

backend/executors/kubernetes_kueue.py

@@ -79,6 +79,7 @@ class StaticPvcMount:
     claim_name: str
     mount_path: str
     read_only: bool = True
+    sub_path: str | None = None
 
     def __post_init__(self) -> None:
         for field in ("name", "claim_name"):
@@ -88,6 +89,11 @@ def __post_init__(self) -> None:
         path = PurePosixPath(self.mount_path)
         if not self.mount_path.startswith("/") or ".." in path.parts:
             raise ValueError("mount_path must be an absolute normalized path")
+        if self.sub_path is not None:
+            child = PurePosixPath(self.sub_path)
+            if (not self.sub_path or self.sub_path.startswith("/")
+                    or ".." in child.parts or str(child) != self.sub_path):
+                raise ValueError("sub_path must be a relative normalized path")
 
 
 class KubernetesKueueAdapter:
@@ -416,6 +422,8 @@ def _job_manifest(self, name: str, config_map: str,
                                     "name": mount.name,
                                     "mountPath": mount.mount_path,
                                     "readOnly": mount.read_only,
+                                    **({"subPath": getattr(mount, "sub_path")}
+                                       if getattr(mount, "sub_path", None) is not None else {}),
                                 } for mount in (*self.static_host_mounts,
                                                 *self.static_pvc_mounts)],
                             ],

backend/kernel.py

@@ -87,6 +87,9 @@ def default_executor():
         static_pvc_mounts=[
             StaticPvcMount('dirac-runtime', 'dirac-motif-runtime',
                            str(worker_repository), True),
+            StaticPvcMount(
+                'dirac-posix-shell', 'dirac-motif-runtime', '/bin/sh', True,
+                sub_path='runtime-bin/dash'),
             StaticPvcMount('dirac-exchange', 'dirac-motif-exchange',
                            str(worker_exchange), False),
         ])
@@ -340,4 +343,14 @@ def build(*, dsn: str = DEFAULT_DSN, with_versions: bool = True,
     svc.job_store_kind = getattr(js, 'kind', 'injected')  # type: ignore[attr-defined]
     svc.job_durability = getattr(js, 'durability', 'unknown')  # type: ignore[attr-defined]
     svc.executor_kind = getattr(ex, 'kind', 'injected')  # type: ignore[attr-defined]
+    try:
+        import psycopg
+        from motif.closed_loop import ClosedLoopController
+        controller = ClosedLoopController(
+            svc, lambda: psycopg.connect(dsn, autocommit=True))
+    except Exception as error:  # noqa: BLE001
+        print(f'[kernel] closed-loop controller unavailable '
+              f'({type(error).__name__}: {error})', file=sys.stderr, flush=True)
+        controller = None
+    svc.closed_loop_controller = controller  # type: ignore[attr-defined]
     return svc

backend/method_registry.py

@@ -223,6 +223,8 @@ def _qm_unit(kind: str, quantity: str, units: str) -> dict:
                              ['dock_vina', '_ligand_pdbqt', '_digest']),
     'physics.motif.openmm_md': ('motif.physics',
                                ['run_openmm_md', '_sha', '_digest']),
+    'physics.motif.openfe_edge': ('motif.openfe_runner',
+                                  ['execute_openfe_edge', '_digest']),
     'physics.motif.rbfe_network': ('motif.rbfe',
                                   ['plan_rbfe_network', '_chemistry', '_digest']),
     'physics.motif.rbfe_aggregate': ('motif.rbfe',
@@ -234,9 +236,7 @@ def _qm_unit(kind: str, quantity: str, units: str) -> dict:
     UNITS[_method_id] = {
         'module': _module,
         'fns': _functions,
-        'consts': (['PREDICTOR_SOURCE_DIGESTS'] if _module == 'motif.mesh' else
-                   ['ACQUISITION_SOURCE_DIGESTS']
-                   if _module == 'motif.advanced_acquisition' else []),
+        'consts': list((_descriptor.get('implementation') or {}).get('constants', [])),
         'descriptor_path': str(_descriptor_path),
         'exec_class': 'job',
         'in_schema': _descriptor['input']['schema'],