[fix.infra.Ω.D8+] make RBFE execution scientifically and operationally fail closed

The FEP workflow previously let client-local state, stale scientific artifacts,
and process-local submission timing stand in for durable identity and evidence.
That made a visually plausible review insufficient proof of the exact receptor,
ligand, mapping, campaign generation, or execution that would later run.

Bind every scientific object to an owner-scoped campaign generation and exact
content digest; require structured receptor, pose, mapping, charge, stereo, and
chemistry witnesses before execution; invalidate descendants atomically when
inputs change; and isolate Job/RunSet access by principal.  Campaign preparation
is now a durable, idempotent Job whose transactional dispatch outbox survives an
API-process crash and rejects stale workers with fencing tokens.

Migrations 039-047 provide the durable campaign, ownership, RunSet,
idempotency, and dispatch state.  Contract schemas and generated clients expose
the same strict surface, while readiness remains fail closed until every required
migration and executor capability is present.

No physical FEP or GPU work was launched while establishing these guarantees.

Author: ivanicu

backend/artifacts_pg.py

@@ -48,13 +48,16 @@ def __init__(self, connect, *, base_path: str = '/v2/artifacts') -> None:
         self._connect = connect
         self.base_path = base_path
         self.counters = {'put': 0, 'put_deduplicated': 0, 'read': 0, 'miss': 0,
-                         'digest_mismatch': 0, 'unavailable': 0}
+                         'digest_mismatch': 0, 'unavailable': 0,
+                         'authorization_denied': 0,
+                         'authorization_schema_unavailable': 0}
 
     # ── writes ────────────────────────────────────────────────────────────────
     def put(self, data: bytes, *, role: str, media_type: str | None = None,
             metadata: dict[str, Any] | None = None,
             method_version: str | None = None,
-            method_id: str | None = None) -> A.Artifact:
+            method_id: str | None = None,
+            cursor=None) -> A.Artifact:
         """Store bytes and name their role. Idempotent by (content, role).
 
         The digest is computed HERE and passed as the primary key, and the database
@@ -74,7 +77,8 @@ def put(self, data: bytes, *, role: str, media_type: str | None = None,
         digest = A.sha256_hex(data)
         mt = media_type or A.MEDIA_TYPES.get(role, 'application/octet-stream')
         self.counters['put'] += 1
-        with self._connect() as conn, conn.cursor() as cur:
+
+        def persist(cur) -> A.Artifact:
             cur.execute(
                 'INSERT INTO app.blob (sha256, media_type, byte_len, bytes) '
                 'VALUES (decode(%s, %s), %s, %s, %s) ON CONFLICT (sha256) DO NOTHING',
@@ -111,8 +115,17 @@ def put(self, data: bytes, *, role: str, media_type: str | None = None,
                               metadata=metadata or {},
                               method_version=method_version)
 
+        # A caller that already owns the authoritative transaction may lend its
+        # cursor.  This is not a convenience: campaign preparation has one CAS
+        # commit barrier, and opening a private connection here used to make the
+        # artifact survive when that later CAS rolled back.
+        if cursor is not None:
+            return persist(cursor)
+        with self._connect() as conn, conn.cursor() as cur:
+            return persist(cur)
+
     def link_to_job(self, job_id: str, artifact_id: str, role: str,
-                    ordinal: int = 0) -> None:
+                    ordinal: int = 0, *, cursor=None) -> None:
         """Record which invocation produced this artifact.
 
         Separate from `put` because an artifact can exist before its job row does
@@ -122,12 +135,43 @@ def link_to_job(self, job_id: str, artifact_id: str, role: str,
         """
         if psycopg is None:
             return
-        with self._connect() as conn, conn.cursor() as cur:
+
+        def link(cur) -> None:
             cur.execute(
                 'INSERT INTO app.job_artifact (job_id, artifact_id, role, ordinal) '
                 'VALUES (%s, %s, %s, %s) ON CONFLICT DO NOTHING',
                 (job_id, artifact_id, role, ordinal))
 
+        if cursor is not None:
+            link(cursor)
+            return
+        with self._connect() as conn, conn.cursor() as cur:
+            link(cur)
+
+    def link_to_campaign(self, campaign_id: str, artifact_id: str, role: str,
+                         ordinal: int = 0, *, cursor=None) -> None:
+        """Make campaign ownership explicit and independently queryable.
+
+        Artifact metadata is immutable first-writer context, not an ownership
+        relation: content deduplication can reuse one artifact in several campaigns.
+        The join table is therefore the authorization/provenance fact.
+        """
+        if psycopg is None:
+            return
+
+        def link(cur) -> None:
+            cur.execute(
+                'INSERT INTO app.rbfe_campaign_artifact '
+                '(campaign_id, artifact_id, role, ordinal) '
+                'VALUES (%s, %s, %s, %s) ON CONFLICT DO NOTHING',
+                (campaign_id, artifact_id, role, ordinal))
+
+        if cursor is not None:
+            link(cursor)
+            return
+        with self._connect() as conn, conn.cursor() as cur:
+            link(cur)
+
     # ── reads ─────────────────────────────────────────────────────────────────
     _SELECT = (
         'SELECT a.id, encode(a.blob_sha256, %s), a.role, a.media_type, '
@@ -144,6 +188,105 @@ def _row_to_artifact(self, row) -> A.Artifact:
                           size_bytes=row[4], encoding=row[5], id=str(row[0]),
                           metadata=dict(row[6] or {}), method_version=row[7])
 
+    @staticmethod
+    def _principal(actor: dict[str, str] | None) -> tuple[str, str]:
+        if not isinstance(actor, dict):
+            raise failures.DiracInvalidParameters(
+                'artifact access requires an authenticated actor')
+        kind = str(actor.get('kind') or '')
+        actor_id = str(actor.get('id') or '').strip()
+        if kind not in {'human', 'agent', 'service'} or not actor_id:
+            raise failures.DiracInvalidParameters(
+                'artifact actor must be a human, agent, or service')
+        return kind, actor_id
+
+    def _require_authorization_schema(self, cur) -> None:
+        """Refuse before compiling ACL SQL against a partially migrated schema.
+
+        PostgreSQL resolves every relation in a statement before evaluating its
+        WHERE clause.  Without this capability probe, a daemon started before 040
+        or 045 turns an intentional deny into an undefined-table 500.  A legacy
+        private Artifact is never assigned an inferred owner or made public as a
+        compatibility shortcut; it remains indistinguishable from absence.
+        """
+        cur.execute(
+            "SELECT to_regclass('app.rbfe_campaign') IS NOT NULL, "
+            "to_regclass('app.rbfe_campaign_artifact') IS NOT NULL, "
+            "EXISTS (SELECT 1 FROM pg_constraint "
+            "WHERE conrelid=to_regclass('app.rbfe_campaign_artifact') "
+            "AND contype='f' "
+            "AND conname='rbfe_campaign_artifact_role_fk' "
+            "AND pg_get_constraintdef(oid) LIKE "
+            "'%%FOREIGN KEY (artifact_id, role)%%')")
+        capability = tuple(cur.fetchone() or (False, False, False))
+        if all(capability):
+            return
+        missing = []
+        if not capability[0]:
+            missing.append('040_rbfe_campaign_state.sql')
+        if not all(capability[1:3]):
+            missing.append('045_rbfe_campaign_artifact_ownership.sql')
+        self.counters['authorization_schema_unavailable'] += 1
+        raise failures.DiracFailure(
+            'DB_UNAVAILABLE',
+            'artifact authorization schema is incomplete; private artifacts '
+            'remain fail-closed until explicit ownership can be verified',
+            details={
+                'required_migrations': missing,
+                'legacy_unowned_policy': 'fail_closed',
+                'owner_inference': False,
+                'implicit_public': False,
+            })
+
+    @staticmethod
+    def _access_clause(alias: str = 'a') -> str:
+        """SQL ownership witness for an HTTP-visible Artifact.
+
+        A UUID or digest is an integrity capability, never an authorization
+        capability.  Legacy/unlinked rows therefore fail closed.  Internal
+        science paths continue to use ``head/read`` because their authorization
+        happened when the server resolved the owning Job/Campaign reference.
+        """
+        return f'''(
+            coalesce({alias}.metadata->>'visibility','') = 'public'
+            OR EXISTS (
+                SELECT 1 FROM app.job_artifact ja
+                JOIN app.job j ON j.id=ja.job_id
+                WHERE ja.artifact_id={alias}.id
+                  AND j.actor_kind=%s AND j.actor_id=%s
+            )
+            OR EXISTS (
+                SELECT 1 FROM app.rbfe_campaign c
+                WHERE c.id::text={alias}.metadata->>'campaign_id'
+                  AND c.created_by_kind=%s AND c.created_by_id=%s
+            )
+            OR EXISTS (
+                SELECT 1 FROM app.rbfe_campaign_artifact ca
+                JOIN app.rbfe_campaign c ON c.id=ca.campaign_id
+                WHERE ca.artifact_id={alias}.id
+                  AND c.created_by_kind=%s AND c.created_by_id=%s
+            )
+            OR EXISTS (
+                SELECT 1
+                FROM design.motif_scientific_object o
+                JOIN app.rbfe_campaign c ON EXISTS (
+                    SELECT 1
+                    FROM jsonb_array_elements(
+                        CASE WHEN jsonb_typeof(c.state->'owned_object_refs')='array'
+                             THEN c.state->'owned_object_refs' ELSE '[]'::jsonb END
+                    ) owned_ref
+                    WHERE owned_ref->>'id'=o.id::text
+                )
+                WHERE o.document_artifact_id={alias}.id
+                  AND c.created_by_kind=%s AND c.created_by_id=%s
+            )
+        )'''
+
+    @staticmethod
+    def _access_params(kind: str, actor_id: str) -> tuple[str, ...]:
+        return (kind, actor_id, kind, actor_id, kind, actor_id,
+                kind, actor_id)
+
     def head(self, address: str) -> A.Artifact:
         """Metadata without bytes — the request a client makes to decide.
 
@@ -184,6 +327,56 @@ def head(self, address: str) -> A.Artifact:
                                                      'roles_at_digest': n}})
             return art
 
+    def head_authorized(self, address: str,
+                        actor: dict[str, str]) -> A.Artifact:
+        """Resolve metadata only when the authenticated actor owns its lineage.
+
+        Digest lookup is filtered before ordering; otherwise the newest artifact
+        at shared bytes could belong to another actor and make an owned role look
+        absent (or, worse, disclose which role was newest).
+        """
+        if psycopg is None:
+            raise failures.DiracFailure('DB_UNAVAILABLE', 'no database driver')
+        kind, actor_id = self._principal(actor)
+        by_digest = A.is_digest(address) or address.startswith('sha256:')
+        digest = address[7:] if address.startswith('sha256:') else address
+        access = self._access_clause('a')
+        access_params = self._access_params(kind, actor_id)
+        with self._connect() as conn, conn.cursor() as cur:
+            self._require_authorization_schema(cur)
+            if by_digest:
+                cur.execute(
+                    self._SELECT
+                    + ' WHERE a.blob_sha256=decode(%s,%s) AND ' + access
+                    + ' ORDER BY a.created_at DESC LIMIT 1',
+                    ("hex", digest, "hex", *access_params))
+            else:
+                cur.execute(
+                    self._SELECT + ' WHERE a.id=%s AND ' + access,
+                    ("hex", address, *access_params))
+            row = cur.fetchone()
+            if row is None:
+                self.counters['miss'] += 1
+                self.counters['authorization_denied'] += 1
+                # Deliberately indistinguishable from a nonexistent address.
+                raise failures.DiracNotFound(
+                    f'no artifact at {address!r}',
+                    details={'address': address, 'looked_up_by':
+                             'digest' if by_digest else 'id'})
+            art = self._row_to_artifact(row)
+            if by_digest:
+                cur.execute(
+                    'SELECT count(*) FROM app.artifact a '
+                    'WHERE a.blob_sha256=decode(%s,%s) AND ' + access,
+                    (digest, 'hex', *access_params))
+                count = int(cur.fetchone()[0])
+                if count > 1:
+                    art = A.Artifact(**{
+                        **art.__dict__,
+                        'metadata': {**art.metadata, 'roles_at_digest': count},
+                    })
+            return art
+
     def read(self, address: str) -> tuple[A.Artifact, bytes]:
         """Bytes, verified against the digest they were stored under.
 
@@ -213,6 +406,25 @@ def read(self, address: str) -> tuple[A.Artifact, bytes]:
         self.counters['read'] += 1
         return art, data
 
+    def read_authorized(self, address: str,
+                        actor: dict[str, str]) -> tuple[A.Artifact, bytes]:
+        art = self.head_authorized(address, actor)
+        with self._connect() as conn, conn.cursor() as cur:
+            cur.execute('SELECT bytes FROM app.blob WHERE sha256 = decode(%s, %s)',
+                        (art.sha256, 'hex'))
+            row = cur.fetchone()
+        if row is None:
+            self.counters['miss'] += 1
+            raise failures.DiracNotFound(f'no artifact at {address!r}')
+        data = bytes(row[0])
+        try:
+            A.verify_bytes(data, art.sha256)
+        except failures.DiracFailure:
+            self.counters['digest_mismatch'] += 1
+            raise
+        self.counters['read'] += 1
+        return art, data
+
     def read_range(self, address: str, range_header: str | None
                    ) -> tuple[A.Artifact, bytes, tuple[int, int] | None]:
         """A byte range, sliced AFTER the whole object is verified.
@@ -229,3 +441,13 @@ def read_range(self, address: str, range_header: str | None
             return art, data, None
         lo, hi = rng
         return art, data[lo:hi + 1], rng
+
+    def read_range_authorized(
+            self, address: str, range_header: str | None,
+            actor: dict[str, str]) -> tuple[A.Artifact, bytes, tuple[int, int] | None]:
+        art, data = self.read_authorized(address, actor)
+        rng = A.parse_range(range_header, len(data))
+        if rng is None:
+            return art, data, None
+        lo, hi = rng
+        return art, data[lo:hi + 1], rng

backend/catalog.py

@@ -43,6 +43,43 @@
 CONTRACTS = pathlib.Path(__file__).resolve().parent.parent / 'contracts'
 
 
+class _FrozenDict(dict):
+    """JSON object whose contract bytes cannot be changed after catalog load."""
+
+    def _immutable(self, *_args, **_kwargs):
+        raise TypeError('Method descriptor is immutable')
+
+    __setitem__ = __delitem__ = clear = pop = popitem = setdefault = update = _immutable
+    __ior__ = _immutable
+
+
+class _FrozenList(list):
+    """JSON array companion to :class:`_FrozenDict`."""
+
+    def _immutable(self, *_args, **_kwargs):
+        raise TypeError('Method descriptor is immutable')
+
+    __setitem__ = __delitem__ = append = clear = extend = insert = pop = remove = _immutable
+    __iadd__ = __imul__ = reverse = sort = _immutable
+
+
+def _freeze_json(value: Any) -> Any:
+    if isinstance(value, dict):
+        return _FrozenDict({key: _freeze_json(child)
+                            for key, child in value.items()})
+    if isinstance(value, (list, tuple)):
+        return _FrozenList(_freeze_json(child) for child in value)
+    return value
+
+
+def _thaw_json(value: Any) -> Any:
+    if isinstance(value, dict):
+        return {key: _thaw_json(child) for key, child in value.items()}
+    if isinstance(value, (list, tuple)):
+        return [_thaw_json(child) for child in value]
+    return value
+
+
 @dataclass(frozen=True)
 class ArtifactDeclaration:
     role: str
@@ -172,7 +209,17 @@ class MethodCatalog:
     """
 
     def __init__(self, specs: dict[str, MethodSpec]) -> None:
-        self._specs = specs
+        # MethodSpec is frozen, but a frozen dataclass containing a mutable dict
+        # is not immutable.  Normalize even injected test/plugin specs here so no
+        # caller can rewrite execution routing after admission.
+        self._specs = {
+            mid: MethodSpec(
+                method_id=spec.method_id, summary=spec.summary,
+                descriptor=_freeze_json(spec.descriptor),
+                handler_ref=spec.handler_ref, estimate_ref=spec.estimate_ref,
+                artifacts=tuple(spec.artifacts), version=spec.version)
+            for mid, spec in specs.items()
+        }
 
     # ── loading ───────────────────────────────────────────────────────────────
     @classmethod
@@ -226,7 +273,7 @@ def bind_versions(self, versions: dict[str, str]) -> MethodCatalog:
         for mid, spec in self._specs.items():
             out[mid] = MethodSpec(
                 method_id=spec.method_id, summary=spec.summary,
-                descriptor=spec.descriptor, handler_ref=spec.handler_ref,
+                descriptor=_freeze_json(spec.descriptor), handler_ref=spec.handler_ref,
                 estimate_ref=spec.estimate_ref, artifacts=spec.artifacts,
                 version=versions.get(mid))
         return MethodCatalog(out)
@@ -337,14 +384,14 @@ def describe(self, method_id: str) -> dict:
             'version': spec.version,
             'summary': spec.summary,
             'description': spec.descriptor.get('description', ''),
-            'input_schema': spec.input_schema,
-            'output_schema': spec.output_schema,
-            'refusals': spec.refusals,
+            'input_schema': _thaw_json(spec.input_schema),
+            'output_schema': _thaw_json(spec.output_schema),
+            'refusals': _thaw_json(spec.refusals),
             'artifacts': [{'role': a.role, 'media_type': a.media_type,
                            'required': a.required, 'typical_bytes': a.typical_bytes}
                           for a in spec.artifacts],
-            'execution': spec.execution,
-            'exposure': spec.exposure,
+            'execution': _thaw_json(spec.execution),
+            'exposure': _thaw_json(spec.exposure),
             'executable': spec.is_executable,
         }
 
@@ -368,7 +415,7 @@ def as_tool_list(self, *, surface: str = 'mcp') -> list[dict]:
                 'method_id': spec.method_id,
                 'description': (spec.summary + ' ' +
                                 spec.descriptor.get('description', '')).strip(),
-                'input_schema': spec.input_schema,
+                'input_schema': _thaw_json(spec.input_schema),
                 'curated': allowed == 'curated',
             })
         return out