codenib.compiler ¶

Compiler infrastructure for CodeNib.

The package provides:

Index compilation (Phase 1): IndexCompiler builds all indexes for a repository and writes a RepoManifest recording what was built, where, and when.
Resource management: ResourceResolver checks index freshness and produces ResourcePlan decisions used by ResourceGuard to filter available skills at query time.
Parameter resolution: SessionContext + resolve_params merge config defaults, session adjustments, and query-time overrides.

Modules:

Name	Description
`artifact_fingerprints`	Dependency-free fingerprints for persisted compiler artifacts.
`artifact_quality`	Commit-surface and completeness gates for reusable benchmark artifacts.
`cache_lock`	Cooperative serialization for a private compiler cache.
`checkout_identity`	Validate that a repository checkout matches a compiled manifest.
`graph_artifact`	Authenticated loading for persisted symbol-graph artifacts.
`index_builders`	Index builder protocol, registry, and concrete implementations.
`index_compiler`	Index Compiler — Phase 1 of the two-phase compilation architecture.
`manifest`	Repo manifest — the linking protocol between index compilation (Phase 1)
`manifest_source`	Bind live repository reads to a manifest's exact source-selection era.
`params`	Layered parameter resolution for skill execution.
`resources`	Index resource management for the skill compiler.
`skill_context`	Skill-aware context construction.
`snapshot_store`	Content-addressed repository snapshots and reusable analysis artifacts.
`verification`	Admission control for incrementally updated indexes.
`zoekt_artifact`	Authenticate Zoekt shards before handing them to an external process.

Classes:

Name	Description
`IndexBuilderRegistry`	Maps index_type names to their builder implementations.
`IndexCompiler`	Orchestrate Phase 1: build all indexes for a repository and write
`IndexCompilerConfig`	Configuration for the IndexCompiler.
`ManifestIndexStateStore`	Read-only `IndexStateStore` backed by a `RepoManifest`.
`RepoManifest`	Top-level manifest describing a compiled repository.
`ResolvedParams`	Final resolved parameter set after merging all layers.
`SessionContext`	Layer 3: Runtime session/repo context injected once per session.
`IndexRequirement`	Declaration of an index dependency from a skill config.
`IndexState`	Lifecycle state of an index.
`ResourcePlan`	Aggregate resource decisions for all indexes needed by a plan.
`ResourceResolver`	Resolves index requirements against current index state.

Functions:

Name	Description
`resolve_params`	Merge parameter layers with later layers taking precedence.
`build_skill_contexts`	Build the union of indexes required by skill_ids and package them.
`load_contexts_from_manifest`	Load index artifacts directly from a pre-built `RepoManifest`.
`required_index_types`	Union of `index_type` strings the given skills must have built.

IndexBuilderRegistry ¶

IndexBuilderRegistry()

Maps index_type names to their builder implementations.

Source code in codenib/compiler/index_builders.py

def __init__(self) -> None:
    self._builders: Dict[str, IndexBuilder] = {}

IndexCompiler ¶

IndexCompiler(
    builder_registry: IndexBuilderRegistry, config: IndexCompilerConfig | None = None
)

Orchestrate Phase 1: build all indexes for a repository and write a manifest.

The manifest (repo_manifest.json) is the linking protocol that Phase 2 (AgentRunner + ResourceGuard) reads to know what indexes are available at query time.

Methods:

Name	Description
`compile_repo`	Build all requested indexes and write the repo manifest.
`update_repo`	Advance an existing manifest to the repo's current HEAD.

Source code in codenib/compiler/index_compiler.py

def __init__(
    self,
    builder_registry: IndexBuilderRegistry,
    config: Optional[IndexCompilerConfig] = None,
) -> None:
    self._builders = builder_registry
    self._config = config or IndexCompilerConfig()

compile_repo ¶

compile_repo(
    repo_path: str,
    *,
    index_types: list[str] | None = None,
    cache_dir: str | None = None
) -> RepoManifest

Build all requested indexes and write the repo manifest.

Parameters:

Name	Type	Description	Default
`repo_path`	`str`	Absolute path to the repository root.	required
`index_types`	`list[str] \| None`	Which indexes to build (default: config.index_types).	`None`
`cache_dir`	`str \| None`	Override for the cache directory (default: `repo_path/.codenib_cache`).	`None`

Returns:

Type	Description
`RepoManifest`	The completed `RepoManifest`.

Source code in codenib/compiler/index_compiler.py

def compile_repo(
    self,
    repo_path: str,
    *,
    index_types: Optional[List[str]] = None,
    cache_dir: Optional[str] = None,
) -> RepoManifest:
    """
    Build all requested indexes and write the repo manifest.

    Args:
        repo_path: Absolute path to the repository root.
        index_types: Which indexes to build (default: config.index_types).
        cache_dir: Override for the cache directory
            (default: ``repo_path/.codenib_cache``).

    Returns:
        The completed ``RepoManifest``.
    """
    repo_path = os.path.abspath(repo_path)
    cache = self._resolve_cache_dir(repo_path, cache_dir)
    with compiler_cache_lock(cache):
        existing = self._load_existing_manifest(cache)
        source_selection = self._snapshot_source_selection(existing)
        return self._compile(
            repo_path,
            index_types=index_types,
            cache_dir=cache,
            existing_manifest=existing,
            source_selection=source_selection,
            force_rebuild=True,
        )

update_repo ¶

update_repo(
    repo_path: str,
    *,
    index_types: list[str] | None = None,
    cache_dir: str | None = None
) -> RepoManifest

Advance an existing manifest to the repo's current HEAD.

Requested views are rebuilt from the captured source generation. An incremental delta is unsafe until the Git cleanliness/HEAD observation can be bound to the same pinned repository authority as the source fingerprint; a path-based status check can otherwise observe a temporary replacement repository and authorize reuse for different bytes.

Falls back to a full :meth:compile_repo when there is no existing manifest, or when the previously indexed commit cannot be determined. Returns the existing manifest untouched when HEAD has not moved.

Source code in codenib/compiler/index_compiler.py

def update_repo(
    self,
    repo_path: str,
    *,
    index_types: Optional[List[str]] = None,
    cache_dir: Optional[str] = None,
) -> RepoManifest:
    """
    Advance an existing manifest to the repo's current HEAD.

    Requested views are rebuilt from the captured source generation.  An
    incremental delta is unsafe until the Git cleanliness/HEAD observation
    can be bound to the same pinned repository authority as the source
    fingerprint; a path-based status check can otherwise observe a
    temporary replacement repository and authorize reuse for different
    bytes.

    Falls back to a full :meth:`compile_repo` when there is no existing
    manifest, or when the previously indexed commit cannot be determined.
    Returns the existing manifest untouched when HEAD has not moved.
    """
    repo_path = os.path.abspath(repo_path)
    cache = self._resolve_cache_dir(repo_path, cache_dir)
    with compiler_cache_lock(cache):
        return self._update_repo_locked(
            repo_path,
            index_types=index_types,
            cache=cache,
        )

IndexCompilerConfig `dataclass` ¶

IndexCompilerConfig(
    cache_dir_name: str = REPO_INDEX_DIRNAME,
    index_types: list[str] = (lambda: ["bm25", "vector", "symbol_graph"])(),
    languages: list[str] = (lambda: ["python"])(),
    source_selection: RepositorySourceSelection | None = None,
)

Configuration for the IndexCompiler.

ManifestIndexStateStore ¶

ManifestIndexStateStore(manifest: RepoManifest)

Read-only IndexStateStore backed by a RepoManifest.

Maps manifest IndexEntry objects to IndexStatus objects that ResourceResolver can consume. Does not support set_status — the manifest is written only by IndexCompiler.

Source code in codenib/compiler/manifest.py

def __init__(self, manifest: RepoManifest) -> None:
    self._manifest = manifest

RepoManifest `dataclass` ¶

RepoManifest(
    version: str = MANIFEST_VERSION,
    repo_path: str = "",
    commit: str = "",
    last_indexed_commit: str = "",
    source_fingerprint: str = "",
    last_indexed_source_fingerprint: str = "",
    source_selection: RepositorySourceSelection | None = (
        lambda: DEFAULT_REPOSITORY_SOURCE_SELECTION
    )(),
    last_indexed_source_selection_digest: str = "",
    languages: list[str] = list(),
    file_count: int = 0,
    indexes: dict[str, IndexEntry] = dict(),
    capabilities: dict[str, bool] = dict(),
    compiled_at: str = "",
    compiled_at_epoch: float = 0.0,
)

Top-level manifest describing a compiled repository.

Written by IndexCompiler after Phase 1. Read by ManifestIndexStateStore at query time (Phase 2).

Methods:

Name	Description
`save`	Atomically replace the manifest after flushing the JSON payload.
`load`	Load a manifest from a JSON file.
`derive_capabilities`	Compute capabilities from the set of available indexes.
`index_is_current`	Whether a view was built for the manifest's current source state.

Attributes:

Name	Type	Description
`source_selection_digest`	`str \| None`	Canonical current selection identity, or the v1.1 legacy sentinel.

source_selection_digest `property` ¶

source_selection_digest: str | None

Canonical current selection identity, or the v1.1 legacy sentinel.

save ¶

save(path: str | Path) -> None

Atomically replace the manifest after flushing the JSON payload.

Source code in codenib/compiler/manifest.py

def save(self, path: str | Path) -> None:
    """Atomically replace the manifest after flushing the JSON payload."""

    p = Path(path)
    p.parent.mkdir(parents=True, exist_ok=True)
    fd, temporary = tempfile.mkstemp(
        dir=str(p.parent),
        prefix=f".{p.name}.",
        suffix=".tmp",
    )
    temporary_path = Path(temporary)
    try:
        try:
            mode = p.stat().st_mode & 0o777
        except FileNotFoundError:
            mode = None
        with os.fdopen(fd, "w", encoding="utf-8") as handle:
            fchmod = getattr(os, "fchmod", None)
            if fchmod is not None and mode is not None:
                fchmod(handle.fileno(), mode)
            json.dump(self.to_dict(), handle, indent=2, allow_nan=False)
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temporary_path, p)
    finally:
        temporary_path.unlink(missing_ok=True)

load `classmethod` ¶

load(path: str | Path) -> RepoManifest

Load a manifest from a JSON file.

Source code in codenib/compiler/manifest.py

@classmethod
def load(cls, path: str | Path) -> RepoManifest:
    """Load a manifest from a JSON file."""
    with open(path, "rb") as source:
        payload = source.read(_MAX_MANIFEST_BYTES + 1)
    if len(payload) > _MAX_MANIFEST_BYTES:
        raise ValueError(f"repository manifest exceeds {_MAX_MANIFEST_BYTES} bytes")
    if payload.startswith(
        (
            b"\xef\xbb\xbf",
            b"\xff\xfe",
            b"\xfe\xff",
            b"\x00\x00\xfe\xff",
            b"\xff\xfe\x00\x00",
        )
    ):
        raise ValueError("repository manifest must use BOM-free UTF-8")
    validate_bounded_json_stream(
        io.BytesIO(payload),
        label="repository manifest",
        max_bytes=_MAX_MANIFEST_BYTES,
    )
    try:
        text = payload.decode("utf-8", errors="strict")
    except UnicodeDecodeError as exc:
        raise ValueError("repository manifest must use UTF-8") from exc
    data = json.loads(
        text,
        object_pairs_hook=_reject_duplicate_json_keys,
        parse_constant=_reject_nonfinite_json_constant,
    )
    return cls.from_dict(data)

derive_capabilities ¶

derive_capabilities() -> None

Compute capabilities from the set of available indexes.

Source code in codenib/compiler/manifest.py

def derive_capabilities(self) -> None:
    """Compute capabilities from the set of available indexes."""
    has_bm25 = self.index_is_current("bm25")
    has_vector = self.index_is_current("vector")
    has_graph = self.index_is_current("symbol_graph")

    self.capabilities = {
        "sparse_search": has_bm25,
        "dense_search": has_vector,
        "hybrid_search": has_bm25 and has_vector,
        "symbol_navigation": has_graph,
    }

index_is_current ¶

index_is_current(index_type: str) -> bool

Whether a view was built for the manifest's current source state.

Source code in codenib/compiler/manifest.py

def index_is_current(self, index_type: str) -> bool:
    """Whether a view was built for the manifest's current source state."""

    entry = self.indexes.get(index_type)
    if entry is None or entry.status != "fresh":
        return False
    if self.version == MANIFEST_VERSION:
        if not is_secure_source_fingerprint_v2(self.source_fingerprint):
            return False
        if entry.commit != self.commit and not (
            not entry.commit
            and not self.source_fingerprint
            and self.source_selection == DEFAULT_REPOSITORY_SOURCE_SELECTION
        ):
            return False
        if entry.source_fingerprint != self.source_fingerprint:
            return False
    elif self.commit and entry.commit and entry.commit != self.commit:
        return False
    if (
        self.version != MANIFEST_VERSION
        and self.source_fingerprint
        and (entry.source_fingerprint != self.source_fingerprint)
    ):
        return False
    if self.version == MANIFEST_VERSION:
        selection_digest = self.source_selection_digest
        if selection_digest is None or (
            entry.source_selection_digest != selection_digest
            and not (
                not entry.source_selection_digest
                and not self.source_fingerprint
                and self.source_selection == DEFAULT_REPOSITORY_SOURCE_SELECTION
            )
        ):
            return False
    return True

ResolvedParams `dataclass` ¶

ResolvedParams(params: dict[str, Any] = dict(), provenance: dict[str, str] = dict())

Final resolved parameter set after merging all layers.

SessionContext `dataclass` ¶

SessionContext(
    repo_path: str | None = None,
    repo_size: int | None = None,
    primary_language: str | None = None,
    index_freshness: dict[str, float] = dict(),
    user_expertise: str = "intermediate",
    extras: dict[str, Any] = dict(),
)

Layer 3: Runtime session/repo context injected once per session.

IndexRequirement `dataclass` ¶

IndexRequirement(
    index_type: str,
    max_age_seconds: float = 3600.0,
    scope: str = "current_repo",
    required: bool = True,
)

Declaration of an index dependency from a skill config.

IndexState ¶

Bases: str, Enum

Lifecycle state of an index.

ResourcePlan `dataclass` ¶

ResourcePlan(
    decisions: list[ResourceDecision] = list(),
    can_execute: bool = True,
    blocking_builds: list[str] = list(),
    async_updates: list[str] = list(),
    warnings: list[str] = list(),
)

Aggregate resource decisions for all indexes needed by a plan.

ResourceResolver ¶

ResourceResolver(state_store: IndexStateStore)

Resolves index requirements against current index state.

Source code in codenib/compiler/resources.py

def __init__(self, state_store: IndexStateStore) -> None:
    self._state_store = state_store

resolve_params ¶

resolve_params(
    defaults: dict[str, Any],
    session_ctx: SessionContext | None = None,
    query_params: dict[str, Any] | None = None,
    *,
    session_rules: SessionParamRules | None = None
) -> ResolvedParams

Merge parameter layers with later layers taking precedence.

Parameters:

Name	Type	Description	Default
`defaults`	`dict[str, Any]`	Layer 1+2 defaults from config.yaml.	required
`session_ctx`	`SessionContext \| None`	Layer 3 runtime session context.	`None`
`query_params`	`dict[str, Any] \| None`	Layer 4 query-time overrides (invocation inputs).	`None`
`session_rules`	`SessionParamRules \| None`	Optional rules for session-based adjustment.	`None`

Returns:

Type	Description
`ResolvedParams`	`ResolvedParams` with merged params and provenance tracking.

Source code in codenib/compiler/params.py

def resolve_params(
    defaults: Dict[str, Any],
    session_ctx: Optional[SessionContext] = None,
    query_params: Optional[Dict[str, Any]] = None,
    *,
    session_rules: Optional[SessionParamRules] = None,
) -> ResolvedParams:
    """
    Merge parameter layers with later layers taking precedence.

    Args:
        defaults: Layer 1+2 defaults from config.yaml.
        session_ctx: Layer 3 runtime session context.
        query_params: Layer 4 query-time overrides (invocation inputs).
        session_rules: Optional rules for session-based adjustment.

    Returns:
        ``ResolvedParams`` with merged params and provenance tracking.
    """
    result: Dict[str, Any] = {}
    provenance: Dict[str, str] = {}

    # Layer 1+2: config defaults
    for k, v in defaults.items():
        result[k] = v
        provenance[k] = "defaults"

    # Layer 3: session context adjustments
    if session_ctx is not None:
        rules = session_rules or DefaultSessionRules()
        adjusted = rules.adjust(result, session_ctx)
        for k, v in adjusted.items():
            if k not in result or result[k] != v:
                result[k] = v
                provenance[k] = "session"

    # Layer 4: query-time overrides (highest priority)
    if query_params:
        for k, v in query_params.items():
            result[k] = v
            provenance[k] = "query"

    return ResolvedParams(params=result, provenance=provenance)

build_skill_contexts ¶

build_skill_contexts(
    repo_path: str,
    skill_ids: Iterable[str],
    *,
    languages: Sequence[str] = ("python",),
    cache_dir: str | None = None,
    skills_dir: str | None = None,
    skill_registry: SkillRegistry | None = None,
    builder_registry: IndexBuilderRegistry | None = None,
    embedding_model: str = DEFAULT_EMBEDDING_MODEL,
    embedding_revision: str | None = None,
    embedding_dimension: int = DEFAULT_EMBEDDING_DIMENSION,
    default_top_k: int = 10,
    default_level: str = "l2",
    rebuild: bool = False,
    trust_remote_code: bool | None = None,
    embedding_batch_size: int | None = None,
    embedding_max_seq_length: int | None = None,
    native_index_authorization: NativeIndexAuthorization | None = None,
    native_index_authorization_resolver: NativeIndexAuthorizationResolver | None = None,
    source_selection: RepositorySourceSelection | None = None
) -> dict[str, Any]

Build the union of indexes required by skill_ids and package them.

Returns the contexts dict accepted by SkillLoader.load_all / SkillLoader.load_skill: {"retrieve": RetrieveContext, "expand": ExpandContext, ...}. Keys are only present when at least one skill of that type was requested AND its index dependencies could be loaded.

Build behaviour

Already-populated index directories under cache_dir/<index_type> are loaded, not rebuilt.
Missing types trigger IndexCompiler.compile_repo for that subset.
rebuild=True forces a fresh build of every requested type.

The function does no agent-side work; it returns context objects ready to hand to SkillLoader. If a requested skill has no index requirements (e.g. code_to_query), it will not contribute a key here — its skill type still gets a context if a sibling skill needs one (e.g. embedding_search shares the retrieve key with bm25_search).

Skill-metadata resolution: when skills_dir is given, each skill's index_requirements / skill_type are read directly from its config.yaml on disk, so callers no longer have to pre-populate the SkillRegistry before calling this (see #154). The registry is only consulted for skills that are not present on disk.

Existing native vector views require either a capability already bound to the intended cache or a resolver that receives its current manifest entry. When this function itself compiles new bytes, it may mint only for that compiler-owned captured output.

Source code in codenib/compiler/skill_context.py

def build_skill_contexts(
    repo_path: str,
    skill_ids: Iterable[str],
    *,
    languages: Sequence[str] = ("python",),
    cache_dir: Optional[str] = None,
    skills_dir: Optional[str] = None,
    skill_registry: Optional[SkillRegistry] = None,
    builder_registry: Optional[IndexBuilderRegistry] = None,
    embedding_model: str = DEFAULT_EMBEDDING_MODEL,
    embedding_revision: Optional[str] = None,
    embedding_dimension: int = DEFAULT_EMBEDDING_DIMENSION,
    default_top_k: int = 10,
    default_level: str = "l2",
    rebuild: bool = False,
    trust_remote_code: Optional[bool] = None,
    embedding_batch_size: Optional[int] = None,
    embedding_max_seq_length: Optional[int] = None,
    native_index_authorization: NativeIndexAuthorization | None = None,
    native_index_authorization_resolver: NativeIndexAuthorizationResolver | None = None,
    source_selection: RepositorySourceSelection | None = None,
) -> Dict[str, Any]:
    """Build the union of indexes required by *skill_ids* and package them.

    Returns the ``contexts`` dict accepted by ``SkillLoader.load_all`` /
    ``SkillLoader.load_skill``: ``{"retrieve": RetrieveContext, "expand":
    ExpandContext, ...}``. Keys are only present when at least one skill of
    that type was requested AND its index dependencies could be loaded.

    Build behaviour:
      - Already-populated index directories under ``cache_dir/<index_type>``
        are loaded, not rebuilt.
      - Missing types trigger ``IndexCompiler.compile_repo`` for that subset.
      - ``rebuild=True`` forces a fresh build of every requested type.

    The function does no agent-side work; it returns context objects ready
    to hand to ``SkillLoader``. If a requested skill has no index
    requirements (e.g. ``code_to_query``), it will not contribute a key
    here — its skill type still gets a context if a *sibling* skill needs
    one (e.g. ``embedding_search`` shares the ``retrieve`` key with
    ``bm25_search``).

    Skill-metadata resolution: when ``skills_dir`` is given, each skill's
    ``index_requirements`` / ``skill_type`` are read directly from its
    ``config.yaml`` on disk, so callers no longer have to pre-populate the
    ``SkillRegistry`` before calling this (see #154). The registry is only
    consulted for skills that are not present on disk.

    Existing native vector views require either a capability already bound to
    the intended cache or a resolver that receives its current manifest entry.
    When this function itself compiles new bytes, it may mint only for that
    compiler-owned captured output.
    """
    skill_ids = list(skill_ids)
    if (
        native_index_authorization is not None
        and native_index_authorization_resolver is not None
    ):
        raise ValueError(
            "provide either native_index_authorization or "
            "native_index_authorization_resolver, not both"
        )
    skill_registry = skill_registry or SkillRegistry()

    cache_dir = cache_dir or os.path.join(
        os.path.abspath(repo_path), _DEFAULT_CACHE_DIR_NAME
    )
    os.makedirs(cache_dir, exist_ok=True)
    selected_source = resolve_compiler_source_selection(cache_dir, source_selection)

    needed = required_index_types(
        skill_ids, skills_dir=skills_dir, skill_registry=skill_registry
    )
    optional = optional_index_types(
        skill_ids, skills_dir=skills_dir, skill_registry=skill_registry
    )
    skill_types = _skill_types_for(
        skill_ids, skills_dir=skills_dir, skill_registry=skill_registry
    )
    load_policy = resolve_embedding_load_policy(
        embedding_model,
        revision=embedding_revision,
        trust_remote_code=trust_remote_code,
    )

    compiled_manifest: RepoManifest | None = None
    compiled_types: Set[str] = set()
    current_manifest = _current_skill_manifest(
        repo_path,
        cache_dir,
        selected_source,
    )
    if needed:
        missing = sorted(
            needed
            if rebuild or current_manifest is None
            else {
                index_type
                for index_type in needed
                if _current_skill_view_path(current_manifest, index_type) is None
            }
        )
        # An existing native vector cache is not self-authorizing. Without a
        # caller-issued capability, rebuild it from the repository source so
        # this compiler invocation owns the bytes it will authorize below.
        if (
            "vector" in needed
            and native_index_authorization is None
            and native_index_authorization_resolver is None
        ):
            missing = sorted(set(missing) | {"vector"})
        if missing:
            registry = builder_registry
            if registry is None:
                registry = IndexBuilderRegistry()
                register_default_builders(
                    registry,
                    languages=list(languages),
                    embedding_model=embedding_model,
                    embedding_revision=load_policy.revision,
                    embedding_dimension=embedding_dimension,
                    trust_remote_code=load_policy.trust_remote_code,
                    embedding_batch_size=embedding_batch_size,
                    embedding_max_seq_length=embedding_max_seq_length,
                    source_selection=selected_source,
                )
            logger.info("Building missing indexes %s under %s", missing, cache_dir)
            compiled_manifest = _run_compiler(
                repo_path,
                missing,
                cache_dir,
                languages=languages,
                builder_registry=registry,
                source_selection=selected_source,
            )
            compiled_types.update(missing)

    active_manifest = compiled_manifest or current_manifest

    # Load required types plus any optional type already present on disk
    # (optional types are never built here — used-if-present only).
    loadable: Set[str] = set(needed)
    for t in optional:
        if _current_skill_view_path(active_manifest, t) is not None:
            if (
                t == "vector"
                and native_index_authorization is None
                and native_index_authorization_resolver is None
            ):
                logger.warning(
                    "Skipping optional native vector cache without external "
                    "authorization"
                )
                continue
            loadable.add(t)

    loaded: Dict[str, Any] = {}
    if "bm25" in loadable:
        bm25_path = _current_skill_view_path(active_manifest, "bm25")
        if bm25_path is None:
            raise ValueError("required BM25 view is not current for this source")
        loaded["bm25"] = _load_bm25(bm25_path)
    if "vector" in loadable:
        vector_path = _current_skill_view_path(active_manifest, "vector")
        if vector_path is None:
            raise ValueError("required vector view is not current for this source")
        vector_artifact_metadata: Dict[str, Any] | None = None
        vector_authorization = native_index_authorization
        if "vector" in compiled_types and compiled_manifest is not None:
            vector_entry = compiled_manifest.indexes.get("vector")
            if (
                vector_entry is not None
                and vector_entry.status == "fresh"
                and compiled_manifest.index_is_current("vector")
            ):
                from ..index.embedding.artifact_integrity import (
                    capture_authenticated_vector_view,
                )
                from ..native_index_authorization import (
                    _mint_trusted_local_admin_authorization,
                )

                vector_path = vector_entry.path
                vector_artifact_metadata = dict(vector_entry.config)
                if native_index_authorization_resolver is not None:
                    vector_authorization = _resolve_native_vector_authorization(
                        vector_entry,
                        native_index_authorization_resolver,
                    )
                else:
                    with capture_authenticated_vector_view(vector_path) as vector_view:
                        vector_authorization = _mint_trusted_local_admin_authorization(
                            vector_view.ownership,
                            view_type="vector",
                            semantic_contract=vector_artifact_metadata,
                            evidence=(
                                "skill-context-compiler-owned-build",
                                "captured-vector-tree-subject",
                            ),
                        )
        elif vector_authorization is not None:
            cached_entry = (
                active_manifest.indexes.get("vector")
                if active_manifest is not None
                else None
            )
            if cached_entry is not None:
                vector_artifact_metadata = dict(cached_entry.config)
        elif native_index_authorization_resolver is not None:
            cached_entry = (
                active_manifest.indexes.get("vector")
                if active_manifest is not None
                else None
            )
            if cached_entry is None:
                raise ValueError(
                    "cannot resolve native-index authorization without a current "
                    "vector manifest entry"
                )
            vector_artifact_metadata = dict(cached_entry.config)
            vector_authorization = _resolve_native_vector_authorization(
                cached_entry,
                native_index_authorization_resolver,
            )
        # Required vectors always reach the fail-closed loader, even when a
        # broken compiler result omitted its authority. Optional vectors may
        # still degrade when no external capability was supplied.
        if vector_authorization is not None or "vector" in needed:
            loaded["vector"] = _load_vector(
                vector_path,
                embedding_model=embedding_model,
                embedding_provider="huggingface",
                embedding_dimension=embedding_dimension,
                embedding_revision=load_policy.revision,
                embedding_kwargs=(
                    {"max_seq_length": embedding_max_seq_length}
                    if embedding_max_seq_length is not None
                    else None
                ),
                trust_remote_code=load_policy.trust_remote_code,
                default_batch_size=embedding_batch_size,
                artifact_metadata=vector_artifact_metadata,
                native_index_authorization=vector_authorization,
            )
    if "symbol_graph" in loadable:
        graph_path = _current_skill_view_path(active_manifest, "symbol_graph")
        graph_entry = (
            active_manifest.indexes.get("symbol_graph")
            if active_manifest is not None
            else None
        )
        if graph_path is None or graph_entry is None:
            raise ValueError("required symbol graph is not current for this source")
        try:
            loaded["symbol_graph"] = _load_symbol_graph(graph_entry)
        except Exception as exc:  # noqa: BLE001
            # A REQUIRED graph must surface its load error; an OPTIONAL one
            # (``required: false``, e.g. bm25_search's relabel graph) must
            # never crash the build — a present-but-stale/corrupt graph dir
            # (schema bump, partial SCIP build) degrades to "no graph".
            if "symbol_graph" in needed:
                raise
            logger.warning(
                "Skipping stale/corrupt optional symbol_graph under %s: %s",
                cache_dir,
                exc,
            )

    lsp_provider = None
    if SkillType.EXPAND in skill_types or "symbol_graph" in loaded:
        lsp_provider = _select_cpp_lsp_provider(
            repo_path=repo_path,
            languages=languages,
            symbol_graph=loaded.get("symbol_graph"),
            source_selection=selected_source,
        )
    return _package_contexts(
        loaded,
        skill_types=skill_types,
        default_top_k=default_top_k,
        default_level=default_level,
        lsp_provider=lsp_provider,
    )

load_contexts_from_manifest ¶

load_contexts_from_manifest(
    manifest: RepoManifest,
    *,
    skill_ids: Iterable[str],
    skill_registry: SkillRegistry | None = None,
    default_top_k: int = 10,
    default_level: str = "l2",
    native_index_authorization: NativeIndexAuthorization | None = None,
    native_index_authorization_resolver: NativeIndexAuthorizationResolver | None = None
) -> dict[str, Any]

Load index artifacts directly from a pre-built RepoManifest.

The AoT counterpart to :func:build_skill_contexts: instead of deciding what to build from skill_ids and then building, this function consumes a manifest written by a previous :class:~codenib.compiler.index_compiler.IndexCompiler.compile_repo run, validates that the indexes the requested skill_ids need exist in it, and packages them into the same context dict shape that :class:~codenib.agent.skills.loader.SkillLoader consumes.

Source of truth for paths and configs is the manifest itself — manifest.indexes[t].path is loaded directly, ignoring any conventional <cache_dir>/<t> layout. The vector index's embedding model/dimension are read from manifest.indexes["vector"].config when present.

native_index_authorization_resolver, when supplied, is invoked with that exact vector IndexEntry and must return an out-of-band capability bound to the same tree and config. Manifest data never mints authority.

Raises:

Type	Description
`ValueError`	if any `skill_ids` requires an `index_type` that isn't present in the manifest or whose `status != "fresh"`. Requirements marked `required: false` are loaded when fresh and skipped silently when absent. Loud failure for required indexes beats the deferred "Skill not available" at tool-call time.

Source code in codenib/compiler/skill_context.py

def load_contexts_from_manifest(
    manifest: RepoManifest,
    *,
    skill_ids: Iterable[str],
    skill_registry: Optional[SkillRegistry] = None,
    default_top_k: int = 10,
    default_level: str = "l2",
    native_index_authorization: NativeIndexAuthorization | None = None,
    native_index_authorization_resolver: NativeIndexAuthorizationResolver | None = None,
) -> Dict[str, Any]:
    """Load index artifacts directly from a pre-built ``RepoManifest``.

    The AoT counterpart to :func:`build_skill_contexts`: instead of
    deciding what to build from ``skill_ids`` and then building, this
    function consumes a manifest written by a previous
    :class:`~codenib.compiler.index_compiler.IndexCompiler.compile_repo`
    run, validates that the indexes the requested ``skill_ids`` need
    exist in it, and packages them into the same context dict shape that
    :class:`~codenib.agent.skills.loader.SkillLoader` consumes.

    Source of truth for paths and configs is the manifest itself —
    ``manifest.indexes[t].path`` is loaded directly, ignoring any
    conventional ``<cache_dir>/<t>`` layout. The vector index's embedding
    model/dimension are read from ``manifest.indexes["vector"].config``
    when present.

    ``native_index_authorization_resolver``, when supplied, is invoked with
    that exact vector ``IndexEntry`` and must return an out-of-band capability
    bound to the same tree and config. Manifest data never mints authority.

    Raises:
        ValueError: if any ``skill_ids`` *requires* an ``index_type`` that
            isn't present in the manifest or whose ``status != "fresh"``.
            Requirements marked ``required: false`` are loaded when fresh and
            skipped silently when absent. Loud failure for required indexes
            beats the deferred "Skill not available" at tool-call time.
    """
    skill_ids = list(skill_ids)
    if (
        native_index_authorization is not None
        and native_index_authorization_resolver is not None
    ):
        raise ValueError(
            "provide either native_index_authorization or "
            "native_index_authorization_resolver, not both"
        )
    registry = skill_registry or SkillRegistry()

    # Map skill_id → list of index_types to load, validating required ones.
    needed: Set[str] = set()
    for sid in skill_ids:
        meta = registry.get(sid)
        if meta is None:
            continue
        for req in meta.index_requirements or []:
            entry = manifest.indexes.get(req.index_type)
            fresh = entry is not None and manifest.index_is_current(req.index_type)
            if fresh:
                if (
                    req.index_type == "vector"
                    and not getattr(req, "required", True)
                    and native_index_authorization is None
                    and native_index_authorization_resolver is None
                ):
                    logger.warning(
                        "Skipping optional manifest vector view without external "
                        "native-index authorization"
                    )
                    continue
                needed.add(req.index_type)
            elif getattr(req, "required", True):
                raise ValueError(
                    f"Manifest is missing required index {req.index_type!r} "
                    f"for skill {sid!r}. Rebuild the manifest with this "
                    f"index_type, or drop {sid!r} from allowed_skills."
                )
            # else: optional + not fresh -> skip silently (used-if-present)

    skill_types = _skill_types_for(skill_ids, skill_registry=registry)

    loaded: Dict[str, Any] = {}
    if "bm25" in needed:
        bm25_entry = manifest.indexes["bm25"]
        require_bm25_manifest_artifact(bm25_entry)
        loaded["bm25"] = _load_bm25(bm25_entry.path)
    if "vector" in needed:
        vec_entry = manifest.indexes["vector"]
        vector_authorization = native_index_authorization
        if native_index_authorization_resolver is not None:
            vector_authorization = _resolve_native_vector_authorization(
                vec_entry,
                native_index_authorization_resolver,
            )
        route = resolve_embedding_artifact_route(vec_entry.config)
        embedding_kwargs = route.embedding_backend_kwargs()
        embedding_kwargs.update(route.client_kwargs())
        loaded["vector"] = _load_vector(
            vec_entry.path,
            embedding_model=route.model,
            embedding_provider=route.provider,
            embedding_dimension=route.dimension,
            index_metric=vec_entry.config.get("index_metric", "ip"),
            embedding_kwargs=embedding_kwargs,
            artifact_metadata=vec_entry.config,
            native_index_authorization=vector_authorization,
        )
    if "symbol_graph" in needed:
        loaded["symbol_graph"] = _load_symbol_graph(manifest.indexes["symbol_graph"])

    lsp_provider = None
    if SkillType.EXPAND in skill_types or "symbol_graph" in loaded:
        lsp_provider = _select_cpp_lsp_provider(
            repo_path=manifest.repo_path,
            languages=manifest.languages,
            symbol_graph=loaded.get("symbol_graph"),
            source_selection=(
                manifest.source_selection or DEFAULT_REPOSITORY_SOURCE_SELECTION
            ),
        )
    return _package_contexts(
        loaded,
        skill_types=skill_types,
        default_top_k=default_top_k,
        default_level=default_level,
        lsp_provider=lsp_provider,
    )

required_index_types ¶

required_index_types(
    skill_ids: Iterable[str],
    *,
    skills_dir: str | None = None,
    skill_registry: SkillRegistry | None = None
) -> set[str]

Union of index_type strings the given skills must have built.

index_requirements is declarative skill metadata that lives on disk in each skill's config.yaml. The SkillRegistry is a runtime binding layer built on top of that metadata — and it is empty until SkillLoader has loaded the very contexts this resolution feeds into. To break that chicken-and-egg, prefer the on-disk configs.

Resolution order

If skills_dir is given, read each skill's config.yaml from disk. This is the authoritative source for bundled skills.
Fall back to skill_registry (or the singleton) for skills that don't live on disk — e.g. registered via the @skill decorator at import time.

Requirements marked required: false are excluded — those are "use if present, don't force a build" (e.g. bm25_search's symbol_graph, used only to relabel names). Use :func:optional_index_types for those.

Skills found in neither source are ignored — callers that need to fail loudly should check membership themselves.

Source code in codenib/compiler/skill_context.py

def required_index_types(
    skill_ids: Iterable[str],
    *,
    skills_dir: Optional[str] = None,
    skill_registry: Optional[SkillRegistry] = None,
) -> Set[str]:
    """Union of ``index_type`` strings the given skills *must* have built.

    ``index_requirements`` is declarative skill metadata that lives on disk in
    each skill's ``config.yaml``. The ``SkillRegistry`` is a *runtime* binding
    layer built on top of that metadata — and it is empty until ``SkillLoader``
    has loaded the very contexts this resolution feeds into. To break that
    chicken-and-egg, prefer the on-disk configs.

    Resolution order:
      1. If ``skills_dir`` is given, read each skill's ``config.yaml`` from
         disk. This is the authoritative source for bundled skills.
      2. Fall back to ``skill_registry`` (or the singleton) for skills that
         don't live on disk — e.g. registered via the ``@skill`` decorator at
         import time.

    Requirements marked ``required: false`` are excluded — those are "use if
    present, don't force a build" (e.g. ``bm25_search``'s ``symbol_graph``,
    used only to relabel names). Use :func:`optional_index_types` for those.

    Skills found in neither source are ignored — callers that need to fail
    loudly should check membership themselves.
    """
    skill_ids = list(skill_ids)
    needed: Set[str] = set()
    seen_on_disk: Set[str] = set()

    if skills_dir:
        configs = _read_skill_configs(skills_dir)
        for sid in skill_ids:
            cfg = configs.get(sid)
            if cfg is None:
                continue
            seen_on_disk.add(sid)
            for req in cfg.get("index_requirements") or []:
                # Mirror the registry branch: ``required: false`` reqs are
                # "use if present, don't force a build" — see
                # :func:`optional_index_types`.
                if not isinstance(req, dict) or not req.get("required", True):
                    continue
                index_type = req.get("type")
                if index_type:
                    needed.add(index_type)

    registry = skill_registry or SkillRegistry()
    for sid in skill_ids:
        if sid in seen_on_disk:
            continue
        meta = registry.get(sid)
        if meta is None:
            continue
        for req in meta.index_requirements or []:
            if getattr(req, "required", True):
                needed.add(req.index_type)
    return needed

codenib.compiler ¶

IndexBuilderRegistry ¶

IndexCompiler ¶

compile_repo ¶

update_repo ¶

IndexCompilerConfig dataclass ¶

ManifestIndexStateStore ¶

RepoManifest dataclass ¶

source_selection_digest property ¶

save ¶

load classmethod ¶

derive_capabilities ¶

index_is_current ¶

ResolvedParams dataclass ¶

SessionContext dataclass ¶

IndexRequirement dataclass ¶

IndexState ¶

ResourcePlan dataclass ¶

ResourceResolver ¶

resolve_params ¶

build_skill_contexts ¶

load_contexts_from_manifest ¶

required_index_types ¶

IndexCompilerConfig `dataclass` ¶

RepoManifest `dataclass` ¶

source_selection_digest `property` ¶

load `classmethod` ¶

ResolvedParams `dataclass` ¶

SessionContext `dataclass` ¶

IndexRequirement `dataclass` ¶

ResourcePlan `dataclass` ¶