[feat.dirac.Ω.D9+] make Program the versioned discovery root

WHY: downstream design, motif, execution, and evidence workflows need one durable aggregate that preserves objectives, hypotheses, decisions, milestones, lineage, and reproducible snapshots.

WHAT: add migrations 020-025, governed Program atoms and commands, Motif execution foundations, connected FASCIA workspaces, voice composition surface, contract/DB gates, and an automatically regenerated function-level architecture twin.

VERIFY: Program domain 6/6; PostgreSQL artifacts 10/10; app-shell 11/11; contract generators; TypeScript; production build; digital twin 4552 nodes / 7817 edges.

Author: ivanicu

.github/workflows/dirac.yml

@@ -87,6 +87,13 @@ jobs:
         run: |
           pip install --quiet jsonschema
           python3 scripts/gen_contracts.py --check
+          python3 scripts/check_public_registry.py
+          PYTHONPATH=backend python3 -m unittest \
+            backend.tests.test_motif_contracts \
+            backend.tests.test_execution_identity \
+            backend.tests.test_local_cas \
+            backend.tests.test_execution_control \
+            backend.tests.test_motif_core
 
       - name: 'Gate 7 — contract drift (errors.json / iface.* / FIELD_META_SCHEMA)'
         if: ${{ !cancelled() }}
@@ -143,3 +150,40 @@ jobs:
           node scripts/watch_digital_twin.mjs --selftest
           python3 scripts/check_digital_twin.py --selftest
           python3 scripts/check_digital_twin.py
+
+  contract-database:
+    name: Motif contract and clean-database alignment
+    runs-on: ubuntu-latest
+    services:
+      postgres:
+        image: pgvector/pgvector:pg18
+        env:
+          POSTGRES_USER: postgres
+          POSTGRES_PASSWORD: dirac
+          POSTGRES_DB: dirac
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd "pg_isready -U postgres -d dirac"
+          --health-interval 5s
+          --health-timeout 5s
+          --health-retries 12
+    env:
+      DIRAC_DSN: postgresql://postgres:dirac@localhost:5432/dirac
+      PGPASSWORD: dirac
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+      - name: PostgreSQL client and Python
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y postgresql-client
+      - name: Clean migration 000 through current
+        run: |
+          for file in backend/db/migrations/*.sql; do
+            psql "$DIRAC_DSN" -X -v ON_ERROR_STOP=1 -f "$file"
+          done
+      - name: Canonical vocabulary matches the live schema
+        run: |
+          python3 scripts/verify_contract_db_alignment.py
+          bash backend/db/check_migration_hashes.sh

backend/artifact_backends/__init__.py

@@ -0,0 +1,6 @@
+"""Replaceable byte stores behind Dirac-owned Artifact capabilities."""
+
+from .local_cas import LocalCAS
+from .protocol import ArtifactReader, ArtifactWriter, CheckpointWriter, StoredBlob
+
+__all__ = ["ArtifactReader", "ArtifactWriter", "CheckpointWriter", "LocalCAS", "StoredBlob"]

backend/artifact_backends/local_cas.py

@@ -0,0 +1,179 @@
+"""Atomic streaming SHA-256 content-addressed storage for a local appliance."""
+from __future__ import annotations
+
+import hashlib
+import io
+import os
+import re
+import tempfile
+from pathlib import Path
+from typing import BinaryIO, Iterable
+
+from .protocol import StoredBlob
+
+_DIGEST = re.compile(r"^(?:sha256:)?([0-9a-f]{64})$")
+
+
+def _hex(digest: str) -> str:
+    match = _DIGEST.fullmatch(digest)
+    if not match:
+        raise ValueError("digest must be sha256:<64 lowercase hex>")
+    return match.group(1)
+
+
+class _BoundedReader(io.RawIOBase):
+    def __init__(self, stream: BinaryIO, remaining: int | None) -> None:
+        self._stream = stream
+        self._remaining = remaining
+
+    def readable(self) -> bool:
+        return True
+
+    def read(self, size: int = -1) -> bytes:
+        if self._remaining == 0:
+            return b""
+        if self._remaining is not None:
+            size = self._remaining if size < 0 else min(size, self._remaining)
+        data = self._stream.read(size)
+        if self._remaining is not None:
+            self._remaining -= len(data)
+        return data
+
+    def close(self) -> None:
+        self._stream.close()
+        super().close()
+
+
+class _PendingLocalWrite:
+    def __init__(self, store: "LocalCAS") -> None:
+        self._store = store
+        fd, raw_path = tempfile.mkstemp(prefix="upload-", suffix=".partial", dir=store.staging)
+        self._path = Path(raw_path)
+        self._stream = os.fdopen(fd, "wb")
+        self._hash = hashlib.sha256()
+        self._size = 0
+        self._closed = False
+
+    def write(self, chunk: bytes) -> int:
+        if self._closed:
+            raise ValueError("write session is closed")
+        if not isinstance(chunk, bytes):
+            raise TypeError("artifact chunks must be bytes")
+        written = self._stream.write(chunk)
+        self._hash.update(chunk[:written])
+        self._size += written
+        return written
+
+    def commit(self) -> StoredBlob:
+        if self._closed:
+            raise ValueError("write session is closed")
+        self._stream.flush()
+        os.fsync(self._stream.fileno())
+        self._stream.close()
+        self._closed = True
+        digest_hex = self._hash.hexdigest()
+        target = self._store._path(digest_hex)
+        target.parent.mkdir(parents=True, exist_ok=True)
+        if target.exists():
+            self._path.unlink()
+            if target.stat().st_size != self._size:
+                raise IOError("content-address collision has a different stored size")
+        else:
+            os.replace(self._path, target)
+            directory_fd = os.open(target.parent, os.O_RDONLY)
+            try:
+                os.fsync(directory_fd)
+            finally:
+                os.close(directory_fd)
+        return StoredBlob(
+            digest=f"sha256:{digest_hex}",
+            size_bytes=self._size,
+            locator=str(target),
+            backend=self._store.kind,
+        )
+
+    def abort(self) -> None:
+        if not self._closed:
+            self._stream.close()
+            self._closed = True
+        self._path.unlink(missing_ok=True)
+
+    def __enter__(self) -> "_PendingLocalWrite":
+        return self
+
+    def __exit__(self, exc_type, exc, traceback) -> bool:
+        if exc_type is not None:
+            self.abort()
+        elif not self._closed:
+            self.abort()
+        return False
+
+
+class LocalCAS:
+    kind = "local_cas"
+
+    def __init__(self, root: str | Path) -> None:
+        self.root = Path(root).resolve()
+        self.blobs = self.root / "blobs" / "sha256"
+        self.staging = self.root / ".staging"
+        self.blobs.mkdir(parents=True, exist_ok=True)
+        self.staging.mkdir(parents=True, exist_ok=True)
+
+    def _path(self, digest: str) -> Path:
+        value = _hex(digest)
+        return self.blobs / value[:2] / value[2:4] / value
+
+    def begin(self) -> _PendingLocalWrite:
+        return _PendingLocalWrite(self)
+
+    def put_chunks(self, chunks: Iterable[bytes]) -> StoredBlob:
+        pending = self.begin()
+        try:
+            for chunk in chunks:
+                pending.write(chunk)
+            return pending.commit()
+        except BaseException:
+            pending.abort()
+            raise
+
+    def open(self, digest: str, *, offset: int = 0, length: int | None = None) -> BinaryIO:
+        if offset < 0 or (length is not None and length < 0):
+            raise ValueError("offset and length must be non-negative")
+        path = self._path(digest)
+        stream = path.open("rb")
+        size = path.stat().st_size
+        if offset > size:
+            stream.close()
+            raise ValueError("offset exceeds artifact size")
+        stream.seek(offset)
+        return _BoundedReader(stream, length)
+
+    def iter_bytes(
+        self, digest: str, *, offset: int = 0, length: int | None = None,
+        chunk_size: int = 1024 * 1024,
+    ) -> Iterable[bytes]:
+        if chunk_size <= 0:
+            raise ValueError("chunk_size must be positive")
+        with self.open(digest, offset=offset, length=length) as stream:
+            while True:
+                chunk = stream.read(chunk_size)
+                if not chunk:
+                    break
+                yield chunk
+
+    def verify(self, digest: str) -> StoredBlob:
+        expected = _hex(digest)
+        hasher = hashlib.sha256()
+        size = 0
+        for chunk in self.iter_bytes(expected):
+            hasher.update(chunk)
+            size += len(chunk)
+        if hasher.hexdigest() != expected:
+            raise IOError(f"artifact digest verification failed for sha256:{expected}")
+        path = self._path(expected)
+        return StoredBlob(
+            digest=f"sha256:{expected}",
+            size_bytes=size,
+            locator=str(path),
+            backend=self.kind,
+        )

backend/artifact_backends/protocol.py

@@ -0,0 +1,41 @@
+"""Narrow streaming capabilities injected into scientific workers."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import BinaryIO, Iterable, Protocol
+
+
+@dataclass(frozen=True)
+class StoredBlob:
+    digest: str
+    size_bytes: int
+    locator: str
+    backend: str
+
+
+class PendingWrite(Protocol):
+    def write(self, chunk: bytes) -> int: ...
+    def commit(self) -> StoredBlob: ...
+    def abort(self) -> None: ...
+
+
+class ArtifactReader(Protocol):
+    def open(self, digest: str, *, offset: int = 0, length: int | None = None) -> BinaryIO: ...
+    def iter_bytes(
+        self, digest: str, *, offset: int = 0, length: int | None = None,
+        chunk_size: int = 1024 * 1024,
+    ) -> Iterable[bytes]: ...
+    def verify(self, digest: str) -> StoredBlob: ...
+
+
+class ArtifactWriter(Protocol):
+    def begin(self) -> PendingWrite: ...
+
+
+class CheckpointWriter(ArtifactWriter, Protocol):
+    """A separate capability so workers cannot mistake partial state for output."""
+
+
+__all__ = [
+    "ArtifactReader", "ArtifactWriter", "CheckpointWriter", "PendingWrite", "StoredBlob"
+]

backend/cache_fields.py

@@ -71,7 +71,8 @@ def __init__(self, get_cube, *, put_cube=None, prepare_mol=None,
                          'skipped_uncacheable': 0, 'error': 0,
                          'write_queued': 0, 'write_ok': 0, 'write_error': 0}
 
-    def lookup(self, method_id: str, payload: dict) -> HandlerResult | None:
+    def lookup(self, method_id: str, payload: dict, *,
+               execution_digest: str | None = None) -> HandlerResult | None:
         kind = CACHEABLE_KIND.get(method_id)
         if kind is None:
             self.counters['skipped_uncacheable'] += 1
@@ -109,7 +110,8 @@ def lookup(self, method_id: str, payload: dict) -> HandlerResult | None:
 
     def store(self, method_id: str, payload: dict, out: HandlerResult, *,
               seconds: float, job_id: str | None = None,
-              envelope: dict | None = None) -> None:
+              envelope: dict | None = None,
+              execution_digest: str | None = None) -> None:
         """Queue a validated computed field for durable persistence.
 
         The invocation never waits for PostgreSQL. This preserves the existing interaction

backend/cache_results.py

@@ -13,7 +13,14 @@
 from invocation import HandlerResult
 
 
-def request_digest(payload: dict) -> bytes:
+def request_digest(payload: dict, *, execution_digest: str | None = None) -> bytes:
+    canonical = json.dumps(
+        {"schema_version": "2.0", "execution_digest": execution_digest,
+         "payload": payload}, sort_keys=True, separators=(',', ':'))
+    return hashlib.sha256(canonical.encode()).digest()
+
+
+def legacy_request_digest(payload: dict) -> bytes:
     canonical = json.dumps(payload, sort_keys=True, separators=(',', ':'))
     return hashlib.sha256(canonical.encode()).digest()
 
@@ -24,9 +31,11 @@ class CompositeCache:
     def __init__(self, *repositories: Any) -> None:
         self.repositories = tuple(r for r in repositories if r is not None)
 
-    def lookup(self, method_id: str, payload: dict) -> HandlerResult | None:
+    def lookup(self, method_id: str, payload: dict, *,
+               execution_digest: str | None = None) -> HandlerResult | None:
         for repository in self.repositories:
-            hit = repository.lookup(method_id, payload)
+            hit = repository.lookup(method_id, payload,
+                                    execution_digest=execution_digest)
             if hit is not None:
                 return hit
         return None
@@ -46,7 +55,8 @@ def __init__(self, connect, artifact_store, *, excluded_methods=()) -> None:
         self.counters = {'looked_up': 0, 'hit': 0, 'miss': 0,
                          'stored': 0, 'deduplicated': 0}
 
-    def lookup(self, method_id: str, payload: dict) -> HandlerResult | None:
+    def lookup(self, method_id: str, payload: dict, *,
+               execution_digest: str | None = None) -> HandlerResult | None:
         if method_id in self._excluded:
             return None
         self.counters['looked_up'] += 1
@@ -56,8 +66,19 @@ def lookup(self, method_id: str, payload: dict) -> HandlerResult | None:
                 'FROM app.v_result_cache_servable c '
                 'JOIN meta.method m ON m.id = c.method_row_id '
                 'WHERE m.method_id = %s AND c.request_digest = %s',
-                (method_id, request_digest(payload)))
+                (method_id, request_digest(payload, execution_digest=execution_digest)))
             row = cur.fetchone()
+            if row is None and execution_digest is not None and '.motif.' not in method_id:
+                # Compatibility window for results written before cache identity v2.
+                # Motif never falls back because checkpoint/calibration/runtime identity
+                # is part of correctness, not only freshness.
+                cur.execute(
+                    'SELECT c.id, c.result, c.provenance, c.warnings, c.parameters_used '
+                    'FROM app.v_result_cache_servable c '
+                    'JOIN meta.method m ON m.id = c.method_row_id '
+                    'WHERE m.method_id = %s AND c.request_digest = %s',
+                    (method_id, legacy_request_digest(payload)))
+                row = cur.fetchone()
             if row is None:
                 self.counters['miss'] += 1
                 return None
@@ -78,7 +99,8 @@ def lookup(self, method_id: str, payload: dict) -> HandlerResult | None:
 
     def store(self, method_id: str, payload: dict, out: HandlerResult, *,
               seconds: float, job_id: str | None = None,
-              envelope: dict | None = None) -> None:
+              envelope: dict | None = None,
+              execution_digest: str | None = None) -> None:
         if method_id in self._excluded or out.cache != 'computed' or envelope is None:
             return
         references = list(envelope.get('artifacts') or [])
@@ -94,7 +116,7 @@ def store(self, method_id: str, payload: dict, out: HandlerResult, *,
                 ' parameters_used, source_job_id, compute_seconds) '
                 'VALUES (%s,%s,%s,%s,%s,%s,%s,%s) '
                 'ON CONFLICT (method_row_id, request_digest) DO NOTHING RETURNING id',
-                (method[0], request_digest(payload), json.dumps(out.result),
+                (method[0], request_digest(payload, execution_digest=execution_digest), json.dumps(out.result),
                  json.dumps(out.provenance), json.dumps(out.warnings),
                  json.dumps(out.parameters_used), job_id, seconds))
             inserted = cur.fetchone()