Skip to content

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
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)
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.

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 
68
69
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 
77
78
79
80
81
82
83
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 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
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``.
    """
    return self._compile(
        repo_path,
        index_types=index_types,
        cache_dir=cache_dir,
        existing_manifest=None,
    )

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.

Each builder is asked for an incremental update rather than a full build. Builders without a real delta path fall back to a rebuild internally, so the result is always correct -- only the cost differs.

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 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
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.

    Each builder is asked for an incremental update rather than a full
    build. Builders without a real delta path fall back to a rebuild
    internally, so the result is always correct -- only the cost differs.

    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 = cache_dir or os.path.join(repo_path, self._config.cache_dir_name)
    manifest_path = os.path.join(cache, MANIFEST_FILENAME)

    if not os.path.isfile(manifest_path):
        logger.info(
            "No manifest at %s; falling back to a full build", manifest_path
        )
        return self.compile_repo(
            repo_path, index_types=index_types, cache_dir=cache_dir
        )

    try:
        existing = RepoManifest.load(manifest_path)
    except Exception as exc:  # noqa: BLE001 - unreadable manifest: rebuild
        logger.warning(
            "Manifest at %s unusable (%s); rebuilding", manifest_path, exc
        )
        return self.compile_repo(
            repo_path, index_types=index_types, cache_dir=cache_dir
        )

    types_to_update = index_types or self._config.index_types
    head_commit = self._get_head_commit(repo_path)
    source = fingerprint_repository(repo_path, exclude_roots=(cache,))
    incomplete = [
        idx_type
        for idx_type in types_to_update
        if idx_type not in existing.indexes
        or not self._entry_matches_source(
            existing.indexes[idx_type],
            commit=head_commit,
            source_fingerprint=source.value,
        )
        or not self._entry_matches_builder(
            existing.indexes[idx_type],
            self._builders.get(idx_type),
        )
    ]
    if not incomplete:
        identity = head_commit[:8] if head_commit else source.value[:19]
        logger.info("Indexes already at %s; nothing to update", identity)
        return existing
    if head_commit and existing.commit == head_commit:
        logger.info(
            "Retrying incomplete indexes at %s: %s",
            head_commit[:8],
            incomplete,
        )
    else:
        logger.info(
            "Updating indexes %s -> %s",
            (existing.commit or "unknown")[:8],
            (head_commit or "HEAD")[:8],
        )
    return self._compile(
        repo_path,
        index_types=index_types,
        cache_dir=cache_dir,
        existing_manifest=existing,
        source=source,
    )

IndexCompilerConfig dataclass 

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

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 
195
196
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 = "",
    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

Write the manifest to a JSON file.
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.

save 

save(path: str | Path) -> None

Write the manifest to a JSON file.

Source code in codenib/compiler/manifest.py 
141
142
143
144
145
146
def save(self, path: str | Path) -> None:
    """Write the manifest to a JSON file."""
    p = Path(path)
    p.parent.mkdir(parents=True, exist_ok=True)
    with open(p, "w") as f:
        json.dump(self.to_dict(), f, indent=2)

load classmethod 

load(path: str | Path) -> RepoManifest

Load a manifest from a JSON file.

Source code in codenib/compiler/manifest.py 
148
149
150
151
152
153
@classmethod
def load(cls, path: str | Path) -> RepoManifest:
    """Load a manifest from a JSON file."""
    with open(path) as f:
        data = json.load(f)
    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 
155
156
157
158
159
160
161
162
163
164
165
166
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 
168
169
170
171
172
173
174
175
176
177
178
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.commit and entry.commit and entry.commit != self.commit:
        return False
    if self.source_fingerprint:
        return entry.source_fingerprint == self.source_fingerprint
    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 
196
197
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 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
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
) -> 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.

Source code in codenib/compiler/skill_context.py 
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
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,
) -> 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.
    """
    skill_ids = list(skill_ids)
    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)

    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,
    )

    if needed:
        missing = _missing_index_types(needed, cache_dir, rebuild=rebuild)
        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,
                )
            logger.info("Building missing indexes %s under %s", missing, cache_dir)
            _run_compiler(
                repo_path,
                missing,
                cache_dir,
                languages=languages,
                builder_registry=registry,
            )

    # 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 _looks_built(cache_dir, t):
            loadable.add(t)

    loaded: Dict[str, Any] = {}
    if "bm25" in loadable:
        loaded["bm25"] = _load_bm25(_index_dir(cache_dir, "bm25"))
    if "vector" in loadable:
        loaded["vector"] = _load_vector(
            _index_dir(cache_dir, "vector"),
            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,
        )
    if "symbol_graph" in loadable:
        try:
            loaded["symbol_graph"] = _load_symbol_graph(
                _index_dir(cache_dir, "symbol_graph")
            )
        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,
            )

    return _package_contexts(
        loaded,
        skill_types=skill_types,
        default_top_k=default_top_k,
        default_level=default_level,
    )

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"
) -> 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.

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 
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
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",
) -> 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.

    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)
    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:
                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:
        loaded["bm25"] = _load_bm25(manifest.indexes["bm25"].path)
    if "vector" in needed:
        vec_entry = manifest.indexes["vector"]
        # Prefer the embedding config that was used at build time so the
        # load uses a compatible tokenizer/dimension.
        emb_model = vec_entry.config.get("embedding_model")
        emb_provider = vec_entry.config.get("embedding_provider")
        emb_dim = vec_entry.config.get(
            "dimension", vec_entry.config.get("embedding_dimension")
        )
        embedding_kwargs = vec_entry.config.get("embedding_kwargs") or {}
        if not isinstance(embedding_kwargs, dict):
            raise ValueError("vector manifest has invalid embedding kwargs")
        if (
            not emb_model
            or not emb_provider
            or not isinstance(emb_dim, int)
            or emb_dim <= 0
        ):
            raise ValueError("vector manifest has incomplete embedding identity")
        loaded["vector"] = _load_vector(
            vec_entry.path,
            embedding_model=emb_model,
            embedding_provider=emb_provider,
            embedding_dimension=emb_dim,
            index_metric=vec_entry.config.get("index_metric", "ip"),
            embedding_kwargs=embedding_kwargs,
        )
    if "symbol_graph" in needed:
        loaded["symbol_graph"] = _load_symbol_graph(
            manifest.indexes["symbol_graph"].path
        )

    return _package_contexts(
        loaded,
        skill_types=skill_types,
        default_top_k=default_top_k,
        default_level=default_level,
    )

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
  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.

Source code in codenib/compiler/skill_context.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
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