feat: embed fragments before publication

README.md

@@ -145,7 +145,13 @@ All dispatch modes read **only** a role-specific scheduler, worker URL,
 WorkerLease signing key, the server-side JSON root registry
 (`CONTEXT_ENGINE_WORKER_FILE_ROOTS_JSON`), and an optional bounded per-file byte
 ceiling (`CONTEXT_ENGINE_WORKER_MAX_FILE_BYTES`, default 1 MiB, accepted only
-within 1–64 MiB). Markdown files are discovered recursively. **A caller may not
+within 1–64 MiB). They also require an explicit embedding provider mode and the
+schema-pinned dimension (`CONTEXT_ENGINE_WORKER_EMBEDDING_PROVIDER` and
+`CONTEXT_ENGINE_WORKER_EMBEDDING_DIMENSION`). CI uses the network-free `twin`
+mode. Real deployments select `external` and supply endpoint, model, and API key
+only through the corresponding `CONTEXT_ENGINE_WORKER_EMBEDDING_*` environment
+variables, including a required batch size bounded to 1–256 inputs per request.
+Markdown files are discovered recursively. **A caller may not
 supply Organization, Source, job, or token** — that is the point of the boundary.
 Output is limited to `dispatched` / `no_work` / `refused`.
 

STATUS.md

@@ -93,6 +93,7 @@ Follow the ADR for its exact evidence boundary.
 | [0059](./docs/decisions/0059-dispatch-scheduled-file-imports-through-exact-leases.md) | Dispatch scheduled File imports through exact leases |
 | [0060](./docs/decisions/0060-reclaim-expired-file-imports-with-bounded-retries.md) | Reclaim expired File imports with bounded retries |
 | [0065](./docs/decisions/0065-recurse-file-discovery-with-anchored-descriptors.md) | Recurse File discovery through anchored descriptors under one bounded byte ceiling |
+| [0066](./docs/decisions/0066-embed-fragments-before-publication.md) | Embed newly published Fragments before activation through an explicit provider |
 
 ADR-0065 extends the active File Provider boundary from a flat root to
 deterministic recursive discovery of canonical nested Markdown paths. Each
@@ -106,6 +107,15 @@ PostgreSQL evidence covers nested publication plus mixed flat/nested replay.
 This does **not** activate provider polling/watchers, a full-resync mechanism,
 new delete authority, or any non-Markdown carrier.
 
+ADR-0066 adds one Supply-owned embedding seam to File publication. New Fragment
+rows receive validated 384-dimensional float32 vectors in the same durable
+publication boundary before activation; unchanged acquisitions and recovery
+past preparation do not call the provider again. The partial HNSW index is a
+future candidate-discovery implementation detail and has no authorization role.
+
+This does **not** activate vector retrieval, query embedding, historical
+backfill, or any Runtime/AuthorizationKernel change.
+
 ### Wire contract, SDK, and trusted delivery
 
 | ADR | Activates |

adapters/embeddings.py

@@ -0,0 +1,231 @@
+"""Network-free and external adapters for the Supply embedding seam."""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Callable
+from contextlib import closing
+from dataclasses import dataclass, field
+from hashlib import shake_256
+from math import sqrt
+from typing import IO, BinaryIO, cast
+from urllib.error import HTTPError
+from urllib.parse import urlsplit
+from urllib.request import HTTPRedirectHandler, Request, build_opener
+
+from engine.supply.embeddings import (
+    CONTEXT_FRAGMENT_EMBEDDING_DIMENSION,
+    EmbeddingProfile,
+    EmbeddingProviderUnavailable,
+    EmbeddingVector,
+    validate_embedding_batch,
+)
+
+_MAX_EXTERNAL_RESPONSE_BYTES = 64 * 1024 * 1024
+_DEFAULT_TIMEOUT_SECONDS = 30.0
+EmbeddingTransport = Callable[[Request, float, int], bytes]
+
+
+class _RejectRedirectHandler(HTTPRedirectHandler):
+    """Keep the configured endpoint as the only bearer-credential recipient."""
+
+    def redirect_request(
+        self,
+        request: Request,
+        fp: IO[bytes],
+        code: int,
+        message: str,
+        headers: object,
+        new_url: str,
+    ) -> Request:
+        del message, new_url
+        raise HTTPError(
+            request.full_url,
+            code,
+            "Embedding redirect is unavailable",
+            headers,  # type: ignore[arg-type]
+            fp,
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class ExternalEmbeddingConfiguration:
+    """Environment-derived external provider configuration."""
+
+    endpoint: str = field(repr=False)
+    model: str
+    api_key: str = field(repr=False)
+    dimension: int
+    batch_size: int
+    timeout_seconds: float = _DEFAULT_TIMEOUT_SECONDS
+
+    def __post_init__(self) -> None:
+        parsed = urlsplit(self.endpoint)
+        if (
+            type(self.endpoint) is not str
+            or not self.endpoint
+            or self.endpoint != self.endpoint.strip()
+            or parsed.scheme != "https"
+            or not parsed.hostname
+            or parsed.username is not None
+            or parsed.password is not None
+            or bool(parsed.query)
+            or bool(parsed.fragment)
+            or type(self.model) is not str
+            or not self.model
+            or self.model != self.model.strip()
+            or type(self.api_key) is not str
+            or not self.api_key
+            or self.api_key != self.api_key.strip()
+            or type(self.timeout_seconds) not in {int, float}
+            or not 0 < float(self.timeout_seconds) <= 120
+            or type(self.batch_size) is not int
+            or not 1 <= self.batch_size <= 256
+        ):
+            raise ValueError("Embedding configuration is not available")
+        EmbeddingProfile(self.dimension)
+
+
+def _default_transport(request: Request, timeout: float, maximum_bytes: int) -> bytes:
+    with closing(
+        cast(
+            BinaryIO,
+            build_opener(_RejectRedirectHandler()).open(  # noqa: S310
+                request,
+                timeout=timeout,
+            ),
+        )
+    ) as response:
+        payload = response.read(maximum_bytes + 1)
+    if len(payload) > maximum_bytes:
+        raise OSError("embedding response exceeded the configured bound")
+    return payload
+
+
+class ExternalEmbeddingProvider:
+    """Call one environment-configured JSON embedding endpoint."""
+
+    __slots__ = ("_configuration", "_transport")
+
+    def __init__(
+        self,
+        configuration: ExternalEmbeddingConfiguration,
+        *,
+        transport: EmbeddingTransport = _default_transport,
+    ) -> None:
+        if type(configuration) is not ExternalEmbeddingConfiguration:
+            raise TypeError("External embedding configuration is required")
+        if not callable(transport):
+            raise TypeError("External embedding transport is required")
+        self._configuration = configuration
+        self._transport = transport
+
+    @property
+    def profile(self) -> EmbeddingProfile:
+        return EmbeddingProfile(self._configuration.dimension)
+
+    def embed(self, inputs: tuple[str, ...]) -> tuple[EmbeddingVector, ...]:
+        if (
+            type(inputs) is not tuple
+            or not inputs
+            or any(type(value) is not str or not value for value in inputs)
+        ):
+            raise EmbeddingProviderUnavailable("Embedding provider is unavailable")
+        try:
+            vectors: list[EmbeddingVector] = []
+            for offset in range(0, len(inputs), self._configuration.batch_size):
+                batch = inputs[offset : offset + self._configuration.batch_size]
+                vectors.extend(self._embed_batch(batch))
+            return tuple(vectors)
+        except Exception:
+            raise EmbeddingProviderUnavailable(
+                "Embedding provider is unavailable"
+            ) from None
+
+    def _embed_batch(self, inputs: tuple[str, ...]) -> tuple[EmbeddingVector, ...]:
+        body = json.dumps(
+                {
+                    "dimensions": self.profile.dimension,
+                    "encoding_format": "float",
+                    "input": list(inputs),
+                    "model": self._configuration.model,
+                },
+                ensure_ascii=False,
+                separators=(",", ":"),
+            ).encode("utf-8")
+        request = Request(
+            self._configuration.endpoint,
+            data=body,
+            headers={
+                "Accept": "application/json",
+                "Authorization": f"Bearer {self._configuration.api_key}",
+                "Content-Type": "application/json",
+            },
+            method="POST",
+        )
+        raw_response = self._transport(
+            request,
+            float(self._configuration.timeout_seconds),
+            _MAX_EXTERNAL_RESPONSE_BYTES,
+        )
+        response = json.loads(raw_response)
+        raw_data = response["data"]
+        if type(raw_data) is not list or len(raw_data) != len(inputs):
+            raise ValueError
+        ordered: list[list[object] | None] = [None] * len(inputs)
+        for item in raw_data:
+            if type(item) is not dict:
+                raise ValueError
+            index = item.get("index")
+            vector = item.get("embedding")
+            if (
+                type(index) is not int
+                or not 0 <= index < len(inputs)
+                or ordered[index] is not None
+                or type(vector) is not list
+            ):
+                raise ValueError
+            ordered[index] = cast(list[object], vector)
+        if any(vector is None for vector in ordered):
+            raise ValueError
+        return validate_embedding_batch(
+            inputs,
+            cast(list[list[object]], ordered),
+            self.profile,
+        )
+
+
+class DeterministicEmbeddingTwin:
+    """Stable content-derived vectors for tests without network egress."""
+
+    __slots__ = ("_profile",)
+
+    def __init__(
+        self,
+        dimension: int = CONTEXT_FRAGMENT_EMBEDDING_DIMENSION,
+    ) -> None:
+        self._profile = EmbeddingProfile(dimension)
+
+    @property
+    def profile(self) -> EmbeddingProfile:
+        return self._profile
+
+    def embed(self, inputs: tuple[str, ...]) -> tuple[EmbeddingVector, ...]:
+        if (
+            type(inputs) is not tuple
+            or not inputs
+            or any(type(value) is not str or not value for value in inputs)
+        ):
+            raise EmbeddingProviderUnavailable("Embedding provider is unavailable")
+        vectors: list[EmbeddingVector] = []
+        for value in inputs:
+            raw = shake_256(
+                b"context-engine.embedding-twin.v1\x00" + value.encode("utf-8")
+            ).digest(self.profile.dimension * 2)
+            unscaled = tuple(
+                (int.from_bytes(raw[offset : offset + 2], "big") - 32767.5) / 32767.5
+                for offset in range(0, len(raw), 2)
+            )
+            norm = sqrt(sum(component * component for component in unscaled))
+            vectors.append(tuple(component / norm for component in unscaled))
+        return validate_embedding_batch(inputs, vectors, self.profile)

applications/worker.py

@@ -15,6 +15,11 @@
 from sqlalchemy import Engine, text
 from sqlalchemy.exc import SQLAlchemyError
 
+from adapters.embeddings import (
+    DeterministicEmbeddingTwin,
+    ExternalEmbeddingConfiguration,
+    ExternalEmbeddingProvider,
+)
 from adapters.file_source import FileReadLimits, FileRootRegistry
 from engine import BUILD_IDENTIFIER
 from engine.control import FileImportReceiver, FileRootRef, SourceRef
@@ -38,6 +43,8 @@
 from engine.runtime import Runtime
 from engine.runtime.construction import required_kernel_dependencies
 from engine.supply import (
+    CONTEXT_FRAGMENT_EMBEDDING_DIMENSION,
+    EmbeddingProvider,
     MarkdownCompilerConfig,
     WorkerLeaseCodec,
     WorkerLeaseKeyring,
@@ -48,6 +55,8 @@
 _FILE_DISPATCH_POLL_SECONDS = 1.0
 DEFAULT_WORKER_MAX_FILE_BYTES = 1_048_576
 _WORKER_MAX_FILE_BYTES_ENV = "CONTEXT_ENGINE_WORKER_MAX_FILE_BYTES"
+_WORKER_EMBEDDING_PROVIDER_ENV = "CONTEXT_ENGINE_WORKER_EMBEDDING_PROVIDER"
+_WORKER_EMBEDDING_DIMENSION_ENV = "CONTEXT_ENGINE_WORKER_EMBEDDING_DIMENSION"
 
 
 class WorkerNoOpCompletionAuthority(Protocol):
@@ -140,6 +149,24 @@ def _required_environment(name: str) -> str:
     return value
 
 
+def _required_bounded_integer_environment(
+    name: str,
+    *,
+    minimum: int,
+    maximum: int,
+) -> int:
+    raw_value = _required_environment(name)
+    if not raw_value.isascii() or not raw_value.isdecimal():
+        raise ValueError("Supply worker configuration is not available")
+    try:
+        value = int(raw_value)
+    except ValueError:
+        raise ValueError("Supply worker configuration is not available") from None
+    if not minimum <= value <= maximum:
+        raise ValueError("Supply worker configuration is not available")
+    return value
+
+
 def _file_read_limits() -> FileReadLimits:
     raw_limit = os.environ.get(_WORKER_MAX_FILE_BYTES_ENV)
     if raw_limit is None:
@@ -152,6 +179,47 @@ def _file_read_limits() -> FileReadLimits:
         raise ValueError("Supply worker configuration is not available") from None
 
 
+def _embedding_provider() -> EmbeddingProvider:
+    """Compose the explicit CI twin or one environment-only external provider."""
+
+    mode = _required_environment(_WORKER_EMBEDDING_PROVIDER_ENV)
+    raw_dimension = _required_environment(_WORKER_EMBEDDING_DIMENSION_ENV)
+    if not raw_dimension.isdecimal():
+        raise ValueError("Supply worker configuration is not available")
+    try:
+        dimension = int(raw_dimension)
+    except ValueError:
+        raise ValueError("Supply worker configuration is not available") from None
+    if dimension != CONTEXT_FRAGMENT_EMBEDDING_DIMENSION:
+        raise ValueError("Supply worker configuration is not available")
+    if mode == "twin":
+        return DeterministicEmbeddingTwin(dimension)
+    if mode != "external":
+        raise ValueError("Supply worker configuration is not available")
+    raw_timeout = os.environ.get("CONTEXT_ENGINE_WORKER_EMBEDDING_TIMEOUT_SECONDS")
+    if raw_timeout is None:
+        timeout_seconds = 30.0
+    else:
+        try:
+            timeout_seconds = float(raw_timeout)
+        except ValueError:
+            raise ValueError("Supply worker configuration is not available") from None
+    return ExternalEmbeddingProvider(
+        ExternalEmbeddingConfiguration(
+            endpoint=_required_environment("CONTEXT_ENGINE_WORKER_EMBEDDING_ENDPOINT"),
+            model=_required_environment("CONTEXT_ENGINE_WORKER_EMBEDDING_MODEL"),
+            api_key=_required_environment("CONTEXT_ENGINE_WORKER_EMBEDDING_API_KEY"),
+            dimension=dimension,
+            batch_size=_required_bounded_integer_environment(
+                "CONTEXT_ENGINE_WORKER_EMBEDDING_BATCH_SIZE",
+                minimum=1,
+                maximum=256,
+            ),
+            timeout_seconds=timeout_seconds,
+        )
+    )
+
+
 def _run_one_file_import() -> int:
     """Consume one exact, signed File job in the independent Supply process."""
 
@@ -180,6 +248,7 @@ def _run_one_file_import() -> int:
             ),
             roots,
             MarkdownCompilerConfig("markdown-config-v1"),
+            embedding_provider=_embedding_provider(),
             clock=lambda: datetime.now(UTC).replace(microsecond=0),
         ).run(
             FileImportLeaseRedemption(
@@ -289,6 +358,7 @@ def _worker_database_time(engine: Engine) -> datetime:
 def _run_file_dispatch(*, single_cycle: bool) -> int:
     """Run configured autonomous File dispatch without caller routing facts."""
 
+    embedding_provider = _embedding_provider()
     codec = WorkerLeaseCodec(
         WorkerLeaseKeyring(active_version=1, keys={1: _worker_signing_key()})
     )
@@ -319,6 +389,7 @@ def worker_factory(receiver: FileImportReceiver) -> PostgreSQLFileImportWorker:
                 receiver,
                 roots,
                 MarkdownCompilerConfig("markdown-config-v1"),
+                embedding_provider=embedding_provider,
                 clock=lambda: _worker_database_time(worker_engine),
             )