codenib.mcp

CodeNib MCP server - exposes backbone capabilities over stdio.

Provides MCP tools for semantic indexing, CodeGraph, and hybrid retrieval (vector, BM25, regex, Zoekt trigram) for external agent frameworks.

Modules:

Name	Description
context	ServerContext - loads a RepoManifest and selected index objects from disk.
explore_bounds	Deterministic byte bounds for composed MCP exploration responses.
explore_session	Bounded, fingerprint-safe context deduplication for stdio MCP sessions.
grep_jev	The source-only grep/Jev vertical shared by the CLI and MCP tool.
prompts	MCP prompt resource - guidance for calling agents.
server	CodeNib MCP server - stdio transport.
tool_surface	Configurable MCP tool visibility for agent-surface experiments.
tools	MCP tool implementations.

Classes:

Name	Description
ServerContext	Runtime context for the MCP server.

ServerContext dataclass

ServerContext(
    manifest: RepoManifest,
    symbol_graph: CodeGraph | None = None,
    bm25: BM25CodeIndexer | None = None,
    regex_index: RegexNodeIndex | None = None,
    zoekt: ZoektSearcher | None = None,
    vector: CodeVectorStore | None = None,
    lsp_provider: Any | None = None,
    lsp_provider_selection: dict[str, Any] = dict(),
    errors: dict[str, str] = dict(),
    artifact: Mapping[str, Any] | None = None,
    source_error: str | None = "source binding has not been verified",
    grep_jev: GrepJevConfig | None = None,
    _native_index_authorization: NativeIndexAuthorization | None = None,
    _artifact_binding: ContextArtifactBinding | None = None,
    _source_binding: RepositorySourceBinding | None = None,
)

Runtime context for the MCP server.

Holds the loaded manifest and runtime-loaded index objects. Missing or failed indexes stay None; tools check at call time and return descriptive errors.

Methods:

Name	Description
read_source_bytes	Read one repository file through the retained source authority.
borrow_source_reader	Borrow the exact source reader retained by this context owner.
search_grep_jev	Run the explicit remote route only against this context's source.
verify_source_status	Refresh whole-tree source truth before publishing verified status.
indexed_query	Deliver local indexed facts only while their source remains current.
load	Load a manifest and the selected runtime views.
load_views	Load additional manifest views without disturbing loaded resources.
close	Release runtime resources owned by this context.
enable_source_definition_provider	Select a lazy candidate worker only for a verified local Python source.
close_source_definition_provider	Reap the owned worker before graph compilation or source release.
begin_explore_session	Create fresh state for one stdio connection.
ensure_explore_session	Return the connection runtime, creating one for direct embeddings.
end_explore_session	Discard one connection's ledger without clearing a newer runtime.
configure_lsp_provider	Bind a runtime-only provider without mutating persisted artifacts.
validate_views	Open selected artifacts without initializing a vector query model.

Attributes:

Name	Type	Description
source_verified	bool	Return whether live reads retain exact content-byte authority.
source_verification_scope	str | None	M1 authenticates v2 content bytes, never mutable Git HEAD state.
commit_verified	bool	Mutable checkout commit provenance requires an M2 source snapshot.
loaded_views	frozenset[str]	Return the runtime views currently available in this context.

source_verified property

source_verified: bool

Return whether live reads retain exact content-byte authority.

source_verification_scope property

source_verification_scope: str | None

M1 authenticates v2 content bytes, never mutable Git HEAD state.

commit_verified property

commit_verified: bool

Mutable checkout commit provenance requires an M2 source snapshot.

loaded_views property

loaded_views: frozenset[str]

Return the runtime views currently available in this context.

read_source_bytes

read_source_bytes(relative: str, *, max_bytes: int) -> bytes

Read one repository file through the retained source authority.

Source code in codenib/mcp/context.py

def read_source_bytes(self, relative: str, *, max_bytes: int) -> bytes:
    """Read one repository file through the retained source authority."""

    if self._source_binding is None:
        raise RuntimeError(
            f"source reads are unavailable: {self.source_error or 'unverified'}"
        )
    try:
        return self._source_binding.read_bytes(relative, max_bytes=max_bytes)
    except Exception as exc:
        self.source_error = str(exc)
        raise

borrow_source_reader

borrow_source_reader() -> RepositorySourceReader

Borrow the exact source reader retained by this context owner.

Source code in codenib/mcp/context.py

def borrow_source_reader(self) -> RepositorySourceReader:
    """Borrow the exact source reader retained by this context owner."""

    if self._source_binding is None or not self._source_binding.usable:
        raise RuntimeError(
            f"source reads are unavailable: {self.source_error or 'unverified'}"
        )
    return self._source_binding.borrow_reader()

search_grep_jev

search_grep_jev(query: str, **options: Any)

Run the explicit remote route only against this context's source.

Source code in codenib/mcp/context.py

def search_grep_jev(self, query: str, **options: Any):
    """Run the explicit remote route only against this context's source."""
    from ..agent.runtime.grep_jev import GrepJevRetriever

    if self.grep_jev is None or self._source_binding is None:
        raise RuntimeError("grep → Jev requires configured, authenticated source")
    return GrepJevRetriever(self.grep_jev).search(
        self._source_binding, query, **options
    )

verify_source_status

verify_source_status() -> bool

Refresh whole-tree source truth before publishing verified status.

Source code in codenib/mcp/context.py

def verify_source_status(self) -> bool:
    """Refresh whole-tree source truth before publishing verified status."""

    source = self._source_binding
    if source is None:
        return False
    try:
        source.authenticated_identity_snapshot()
    except Exception as exc:
        self.source_error = source.failure_reason or str(exc)
        return False
    self.source_error = None
    return True

indexed_query

indexed_query() -> Iterator[None]

Deliver local indexed facts only while their source remains current.

A query owns runtime resources until its final source check accepts the result; close waits on the same query lock. Lock order is query, then view (when needed), then source verification. Session bookkeeping uses only the view lock, so workers must not retain it while calling providers: that would block the MCP event loop on session lookup. A mid-query edit discards the result. Portable artifacts retain snapshot semantics; grep/Jev captures current source independently on each request.

Source code in codenib/mcp/context.py

@contextmanager
def indexed_query(self) -> Iterator[None]:
    """Deliver local indexed facts only while their source remains current.

    A query owns runtime resources until its final source check accepts
    the result; close waits on the same query lock. Lock order is query,
    then view (when needed), then source verification. Session bookkeeping
    uses only the view lock, so workers must not retain it while calling
    providers: that would block the MCP event loop on session lookup.
    A mid-query edit discards the result. Portable artifacts retain snapshot
    semantics;
    grep/Jev captures current source independently on each request.
    """

    with self._query_lock:
        if (
            self.artifact is not None
            or self._artifact_binding is not None
            or self.grep_jev is not None
        ):
            yield
            return
        source = self._source_binding
        recovery = (
            "Run `codenib codegraph init` for this repository and retry; "
            "this MCP connection reloads rebuilt indexes. "
            if self._local_manifest_path is not None
            else "Run `codenib codegraph init` for this repository and "
            "restart its CodeNib MCP server. "
        )
        if source is None:
            raise IndexedQueryError(
                "Indexed source is unavailable. " + recovery,
                self.source_error or "source binding is unverified",
            )

        def require_current_source() -> None:
            try:
                source.verify_snapshot()
            except (OSError, RuntimeError, ValueError) as exc:
                self.source_error = source.failure_reason or str(exc)
                failure = IndexedQueryError(
                    "Indexed source changed. " + recovery, self.source_error
                )
                # Authentication revokes source authority immediately. The
                # worker owns only an immutable copy; reap it before this
                # failed query returns, rather than retaining its analysis
                # memory until a later rebuild or transport disconnect.
                try:
                    self.close_source_definition_provider()
                except Exception as cleanup:
                    failure.source_definition_cleanup_error = cleanup
                    _attach_context_cleanup_owner(failure, self)
                raise failure from exc

        require_current_source()
        try:
            yield
        finally:
            # Verify on failure too, without poisoning unchanged source
            # merely because a provider rejected a query or was cancelled.
            require_current_source()

load classmethod

load(
    manifest_path: RepoManifest | str | Path,
    *,
    views: Iterable[str] | None = None,
    artifact: Mapping[str, Any] | None = None,
    artifact_binding: ContextArtifactBinding | None = None,
    artifact_reader: PublicationDirectoryReader | None = None,
    native_index_authorization: NativeIndexAuthorization | None = None,
    source_binding: RepositorySourceBinding | None = None,
    _context_owner: Callable[[ServerContext], None] | None = None
) -> ServerContext

Load a manifest and the selected runtime views.

views=None preserves the MCP server's load-all behavior. Explicit selections avoid importing or starting unrelated view runtimes. Each selected view is loaded independently; a failure in one does not block the others. Failed views are recorded in errors.

Source code in codenib/mcp/context.py

@classmethod
def load(
    cls,
    manifest_path: RepoManifest | str | Path,
    *,
    views: Iterable[str] | None = None,
    artifact: Mapping[str, Any] | None = None,
    artifact_binding: ContextArtifactBinding | None = None,
    artifact_reader: PublicationDirectoryReader | None = None,
    native_index_authorization: NativeIndexAuthorization | None = None,
    source_binding: RepositorySourceBinding | None = None,
    _context_owner: Callable[[ServerContext], None] | None = None,
) -> ServerContext:
    """Load a manifest and the selected runtime views.

    ``views=None`` preserves the MCP server's load-all behavior. Explicit
    selections avoid importing or starting unrelated view runtimes. Each
    selected view is loaded independently; a failure in one does not block
    the others. Failed views are recorded in ``errors``.
    """
    ctx: ServerContext | None = None
    try:
        if (
            artifact_reader is not None
            and type(artifact_reader) is not PublicationDirectoryReader
        ):
            raise TypeError(
                "authenticated artifact reader must use the exact reader type"
            )
        if artifact_reader is not None and artifact_binding is None:
            raise ValueError(
                "an authenticated artifact reader requires its verified binding"
            )
        selected = _resolve_views(views)
        artifact_origin = artifact is not None or artifact_binding is not None
        if artifact_origin and native_index_authorization is not None:
            raise ValueError(
                "portable artifact contexts cannot authorize native vector parsing"
            )
        if artifact_origin:
            disallowed = selected - _PORTABLE_ARTIFACT_RUNTIME_VIEWS
            if views is not None and disallowed:
                raise ValueError(
                    "portable artifact contexts cannot load native views: "
                    + ", ".join(sorted(disallowed))
                )
            selected &= _PORTABLE_ARTIFACT_RUNTIME_VIEWS
        if artifact_binding is not None:
            if (
                not isinstance(manifest_path, RepoManifest)
                or manifest_path.to_dict() != artifact_binding.manifest.to_dict()
            ):
                raise ValueError(
                    "artifact runtime manifest must come from its verified binding"
                )
            manifest = artifact_binding.manifest
        else:
            manifest = (
                manifest_path
                if isinstance(manifest_path, RepoManifest)
                else RepoManifest.load(manifest_path)
            )

        ctx = cls(
            manifest=manifest,
            artifact=dict(artifact) if artifact is not None else None,
            _native_index_authorization=native_index_authorization,
            _artifact_binding=artifact_binding,
            _source_binding=None,
        )
        if _context_owner is not None:
            if not callable(_context_owner):
                raise TypeError("context owner must be callable")
            # Publish the context in a stable caller-owned slot before it
            # acquires any runtime resource.  A cancellation after a view
            # loader returns can then be reconciled by that owner instead
            # of dropping an unreturned live context.
            _context_owner(ctx)
        if artifact_binding is not None:
            artifact_binding.install_source_binding(
                ctx._install_repository_source,
                expected=source_binding,
                require_expected=source_binding is not None,
            )
        else:
            ctx._install_repository_source(source_binding)
        owned_source = ctx._source_binding

        if owned_source is not None:
            from ..compiler.manifest_source import require_manifest_source_identity
            from ..source_fingerprint import (
                is_secure_source_fingerprint_v2,
                lexical_repository_path,
            )

            source_identity = owned_source.authenticated_identity_snapshot()
            if (
                not owned_source.usable
                or not is_secure_source_fingerprint_v2(manifest.source_fingerprint)
                or source_identity.root
                != lexical_repository_path(manifest.repo_path)
            ):
                raise ValueError(
                    "repository source authority does not match the manifest"
                )
            require_manifest_source_identity(
                source_identity,
                manifest,
                label="repository source authority",
                mismatch_message=(
                    "repository source authority does not match the manifest"
                ),
            )

        ctx.source_error = (
            None
            if owned_source is not None
            else "source binding has not been verified"
        )

        if artifact_reader is None:
            ctx.load_views(selected)
        else:
            ctx.load_views(selected, artifact_reader=artifact_reader)
        ctx.configure_lsp_provider(
            allow_native=False,
            native_disabled_reason=(
                "portable_artifact_uses_persisted_graph"
                if artifact_origin
                else "local_source_not_verified"
            ),
        )

        cap_summary = {k: v for k, v in manifest.capabilities.items() if v}
        loaded = [
            view
            for view, _ in _VIEW_LOADERS
            if getattr(ctx, view, None) is not None
        ]
        logger.info(
            "ServerContext ready  repo=%s  commit=%s  requested=%s  loaded=%s  "
            "capabilities=%s  errors=%s",
            manifest.repo_path,
            manifest.commit[:8] if manifest.commit else "N/A",
            sorted(selected),
            loaded or "none",
            cap_summary or "none",
            list(ctx.errors) or "none",
        )
        return ctx
    except BaseException as primary:  # noqa: B036 - preserve primary
        from ..artifacts.runtime import (
            _raise_source_cleanup_failure,
            _source_cleanup_owner_is_pending,
        )

        runtime_cleanup_failure: BaseException | None = None
        if ctx is not None:
            try:
                ctx.close()
            except BaseException as exc:  # noqa: B036 - retain source priority
                _attach_context_cleanup_owner(exc, ctx)
                _attach_context_cleanup_owner(primary, ctx)
                runtime_cleanup_failure = exc
        cleanup_source = (
            ctx._source_binding
            if ctx is not None and ctx._source_binding is not None
            else (
                source_binding
                if artifact_binding is None
                and _source_cleanup_owner_is_pending(source_binding)
                else None
            )
        )
        cleanup_failure = runtime_cleanup_failure
        if cleanup_source is not None:
            try:
                cleanup_source.close()
            except BaseException as exc:  # noqa: B036 - apply shared priority
                cleanup_failure = _retain_context_cleanup_failure(
                    cleanup_failure,
                    "source cleanup retry also failed",
                    exc,
                )
        pending_owner = (
            cleanup_source
            if _source_cleanup_owner_is_pending(cleanup_source)
            else None
        )
        _raise_source_cleanup_failure(
            primary,
            cleanup_failure,
            pending_owner,
        )

load_views

load_views(
    views: Iterable[str],
    *,
    native_index_authorization: NativeIndexAuthorization | None = None,
    artifact_reader: PublicationDirectoryReader | None = None
) -> dict[str, str]

Load additional manifest views without disturbing loaded resources.

View dependencies are resolved in the same way as :meth:load. The operation is idempotent and serialized so query-time planners can safely request only the backends selected for a query. The returned mapping contains requested views that remain unavailable.

Source code in codenib/mcp/context.py

def load_views(
    self,
    views: Iterable[str],
    *,
    native_index_authorization: NativeIndexAuthorization | None = None,
    artifact_reader: PublicationDirectoryReader | None = None,
) -> Dict[str, str]:
    """Load additional manifest views without disturbing loaded resources.

    View dependencies are resolved in the same way as :meth:`load`. The
    operation is idempotent and serialized so query-time planners can
    safely request only the backends selected for a query. The returned
    mapping contains requested views that remain unavailable.
    """

    selected = _resolve_views(views)
    with self._view_lock:
        if (
            artifact_reader is not None
            and type(artifact_reader) is not PublicationDirectoryReader
        ):
            raise TypeError(
                "authenticated artifact reader must use the exact reader type"
            )
        if artifact_reader is not None and self._artifact_binding is None:
            raise ValueError(
                "an authenticated artifact reader requires its verified binding"
            )
        artifact_origin = (
            self.artifact is not None or self._artifact_binding is not None
        )
        if artifact_origin:
            disallowed = selected - _PORTABLE_ARTIFACT_RUNTIME_VIEWS
            if disallowed:
                raise ValueError(
                    "portable artifact contexts cannot load native views: "
                    + ", ".join(sorted(disallowed))
                )
            if (
                native_index_authorization is not None
                or self._native_index_authorization is not None
            ):
                raise ValueError(
                    "portable artifact contexts cannot authorize native parsing"
                )
        if native_index_authorization is not None:
            self._native_index_authorization = native_index_authorization
        for view, loader_name in _VIEW_LOADERS:
            if view not in selected or getattr(self, view, None) is not None:
                continue
            if view == "bm25" and artifact_reader is not None:
                self._load_bm25(artifact_reader=artifact_reader)
            else:
                getattr(self, loader_name)()
            if getattr(self, view, None) is not None:
                self.errors.pop(view, None)
        if "symbol_graph" in selected:
            self.configure_lsp_provider(
                allow_native=self._lsp_allow_native,
                native_disabled_reason=self._lsp_native_disabled_reason,
            )
        return {
            view: self.errors.get(view, "view did not load")
            for view in selected
            if getattr(self, view, None) is None
        }

close

close(*, _preserve_explore_session: bool = False) -> None

Release runtime resources owned by this context.

Only the serialized MCP reloader preserves a connection runtime; it transfers that ledger/barrier to the replacement after cleanup succeeds. Normal close still ends the session and releases every resource.

Source code in codenib/mcp/context.py

def close(self, *, _preserve_explore_session: bool = False) -> None:
    """Release runtime resources owned by this context.

    Only the serialized MCP reloader preserves a connection runtime; it
    transfers that ledger/barrier to the replacement after cleanup succeeds.
    Normal close still ends the session and releases every resource.
    """

    with self._query_lock, self._view_lock:
        deferred: BaseException | None = None
        if not _preserve_explore_session:
            try:
                self.end_explore_session()
            except BaseException as exc:  # noqa: B036 - visit every runtime owner
                deferred = exc
        try:
            self.close_source_definition_provider()
        except BaseException as exc:  # noqa: B036 - retain source and child
            deferred = _retain_context_cleanup_failure(
                deferred, "source definition cleanup also failed", exc
            )
        self.lsp_provider = None
        if self.zoekt is not None:
            try:
                _stop_zoekt(self.zoekt)
            except BaseException as exc:  # noqa: B036 - continue cleanup
                deferred = _retain_context_cleanup_failure(
                    deferred,
                    "Zoekt cleanup also failed",
                    exc,
                )
            else:
                self.zoekt = None
        if self.zoekt is None and self._zoekt_snapshot is not None:
            try:
                self._zoekt_snapshot.close()
            except BaseException as exc:  # noqa: B036 - continue cleanup
                deferred = _retain_context_cleanup_failure(
                    deferred,
                    "Zoekt shard snapshot cleanup also failed",
                    exc,
                )
            if self._zoekt_snapshot.closed:
                self._zoekt_snapshot = None
        if self.vector is not None:
            try:
                _close_vector(self.vector)
            except BaseException as exc:  # noqa: B036 - continue cleanup
                deferred = _retain_context_cleanup_failure(
                    deferred,
                    "vector cleanup also failed",
                    exc,
                )
            else:
                self.vector = None
        if (
            self._source_binding is not None
            and self._source_definition_provider is None
        ):
            try:
                self._source_binding.close()
            except BaseException as exc:  # noqa: B036 - preserve retry owner
                deferred = _retain_context_cleanup_failure(
                    deferred,
                    "source cleanup also failed",
                    exc,
                )
            if self._source_binding.closed:
                self._source_binding = None
                self.source_error = "source binding is closed"
        if deferred is not None:
            _attach_context_cleanup_owner(deferred, self)
            raise deferred

enable_source_definition_provider

enable_source_definition_provider() -> bool

Select a lazy candidate worker only for a verified local Python source.

The serialized MCP preparer calls this before a positional definition. Missing software simply keeps the existing complete-graph route; no install or provider network request occurs inside a source lease.

Source code in codenib/mcp/context.py

def enable_source_definition_provider(self) -> bool:
    """Select a lazy candidate worker only for a verified local Python source.

    The serialized MCP preparer calls this before a positional definition.
    Missing software simply keeps the existing complete-graph route; no
    install or provider network request occurs inside a source lease.
    """
    from ..agent.source_definitions import (
        SourceDefinitionProvider,
        managed_definition_runtime,
    )
    from ..languages import normalize_graph_language

    with self._query_lock:
        if (
            not self._defer_graph
            or not self._lsp_allow_native
            or not self.source_verified
            or self._local_manifest_path is None
            or self.symbol_graph is not None
            or self.artifact is not None
            or self._artifact_binding is not None
            or self.grep_jev is not None
            or not self.manifest.languages
            or any(
                normalize_graph_language(language) != "python"
                for language in self.manifest.languages
            )
        ):
            return False
        if self._source_definition_provider is None:
            runtime = managed_definition_runtime()
            if runtime is None:
                return False
            self._source_definition_provider = SourceDefinitionProvider(
                self._source_binding, self.manifest.source_fingerprint, runtime
            )
        if self._source_definition_provider.status not in ("available", "ok"):
            return False
        self.lsp_provider = self._source_definition_provider
        self.lsp_provider_selection = self.lsp_provider.selection()
        return True

close_source_definition_provider

close_source_definition_provider() -> None

Reap the owned worker before graph compilation or source release.

Lock order is the caller's MCP context lock, then this query lock, then bounded worker I/O. No source lease is held during shutdown. A failed close keeps this field and the source binding for retry.

Source code in codenib/mcp/context.py

def close_source_definition_provider(self) -> None:
    """Reap the owned worker before graph compilation or source release.

    Lock order is the caller's MCP context lock, then this query lock,
    then bounded worker I/O. No source lease is held during shutdown.
    A failed close keeps this field and the source binding for retry.
    """
    with self._query_lock:
        provider = self._source_definition_provider
        if provider is None:
            return
        try:
            provider.close()
        except BaseException as exc:  # noqa: B036 - retain cleanup owner
            provider.status = "cleanup_pending"
            self.lsp_provider_selection = provider.selection()
            _attach_context_cleanup_owner(exc, self)
            if isinstance(exc, Exception):
                failure = IndexedQueryError(
                    "Source definition cleanup did not finish. Retry this query; "
                    "its worker and source remain owned and graph results "
                    "were not delivered.",
                    str(exc),
                )
                _attach_context_cleanup_owner(failure, self)
                raise failure from exc
            raise
        self._source_definition_provider = None
        if self.lsp_provider is provider:
            self.lsp_provider = None
            self.lsp_provider_selection = {
                "status": "unavailable",
                "backend": "unavailable",
                "fallback_reason": "source_definition_worker_closed",
                "capabilities": {
                    "definition": False,
                    "references": False,
                    "route": False,
                },
            }

begin_explore_session

begin_explore_session() -> ExploreSessionRuntime

Create fresh state for one stdio connection.

Stdio normally has one live connection per process. Replacing any previous runtime keeps reconnects fail-closed even when an embedding application reuses the same :class:ServerContext.

Source code in codenib/mcp/context.py

def begin_explore_session(self) -> ExploreSessionRuntime:
    """Create fresh state for one stdio connection.

    Stdio normally has one live connection per process. Replacing any
    previous runtime keeps reconnects fail-closed even when an embedding
    application reuses the same :class:`ServerContext`.
    """

    from .explore_session import ExploreSessionRuntime

    with self._view_lock:
        previous = self.explore_runtime
        runtime = ExploreSessionRuntime()
        self.explore_runtime = runtime
        self._explore_loop = _running_event_loop()
        if previous is not None:
            previous.close()
        return runtime

ensure_explore_session

ensure_explore_session(
    *, _loop: AbstractEventLoop | None = None
) -> ExploreSessionRuntime

Return the connection runtime, creating one for direct embeddings.

Source code in codenib/mcp/context.py

def ensure_explore_session(
    self, *, _loop: asyncio.AbstractEventLoop | None = None
) -> ExploreSessionRuntime:
    """Return the connection runtime, creating one for direct embeddings."""

    from .explore_session import ExploreSessionRuntime

    with self._view_lock:
        loop = _loop if _loop is not None else _running_event_loop()
        if self.explore_runtime is None or (
            loop is not None
            and self._explore_loop is not None
            and loop is not self._explore_loop
        ):
            previous = self.explore_runtime
            self.explore_runtime = ExploreSessionRuntime()
            if previous is not None:
                previous.close()
        if loop is not None:
            self._explore_loop = loop
        return self.explore_runtime

end_explore_session

end_explore_session(runtime: ExploreSessionRuntime | None = None) -> None

Discard one connection's ledger without clearing a newer runtime.

Source code in codenib/mcp/context.py

def end_explore_session(self, runtime: ExploreSessionRuntime | None = None) -> None:
    """Discard one connection's ledger without clearing a newer runtime."""

    with self._view_lock:
        current = self.explore_runtime
        if current is None or (runtime is not None and current is not runtime):
            return
        current.close()
        self.explore_runtime = None
        self._explore_loop = None

configure_lsp_provider

configure_lsp_provider(
    *,
    allow_native: bool,
    native_disabled_reason: str = "native_provider_not_authorized"
) -> dict[str, Any]

Bind a runtime-only provider without mutating persisted artifacts.

Source code in codenib/mcp/context.py

def configure_lsp_provider(
    self,
    *,
    allow_native: bool,
    native_disabled_reason: str = "native_provider_not_authorized",
) -> Dict[str, Any]:
    """Bind a runtime-only provider without mutating persisted artifacts."""

    from ..agent.lsp_provider import select_checkout_lsp_provider

    self._lsp_allow_native = allow_native
    self._lsp_native_disabled_reason = native_disabled_reason
    provider, selection = select_checkout_lsp_provider(
        project_root=self.manifest.repo_path,
        languages=self.manifest.languages,
        symbol_graph=self.symbol_graph,
        source_selection=(
            self.manifest.source_selection or DEFAULT_REPOSITORY_SOURCE_SELECTION
        ),
        allow_native=allow_native,
        native_disabled_reason=native_disabled_reason,
    )
    self.lsp_provider = provider
    self.lsp_provider_selection = selection
    return dict(selection)

validate_views classmethod

validate_views(
    manifest: RepoManifest | str | Path,
    *,
    views: Iterable[str],
    native_index_authorization: NativeIndexAuthorization | None = None
) -> dict[str, str]

Open selected artifacts without initializing a vector query model.

The returned mapping contains only unavailable views. Vector indexes follow the normal FAISS/document load path with a fixed-dimension embedding probe. Temporary vector and Zoekt resources are released before returning.

Source code in codenib/mcp/context.py

@classmethod
def validate_views(
    cls,
    manifest: RepoManifest | str | Path,
    *,
    views: Iterable[str],
    native_index_authorization: NativeIndexAuthorization | None = None,
) -> Dict[str, str]:
    """Open selected artifacts without initializing a vector query model.

    The returned mapping contains only unavailable views. Vector indexes
    follow the normal FAISS/document load path with a fixed-dimension
    embedding probe. Temporary vector and Zoekt resources are released
    before returning.
    """

    selected = _resolve_views(views)
    resolved_manifest = (
        manifest
        if isinstance(manifest, RepoManifest)
        else RepoManifest.load(manifest)
    )
    ctx = cls(
        manifest=resolved_manifest,
        _native_index_authorization=native_index_authorization,
    )
    available: set[str] = set()
    try:
        for view, loader_name in _VIEW_LOADERS:
            if view not in selected:
                continue
            loader = getattr(ctx, loader_name)
            if view == "vector":
                loader(probe=True)
            else:
                loader()
            if getattr(ctx, view, None) is not None:
                available.add(view)
        return {
            view: ctx.errors.get(view, "view did not load")
            for view in selected
            if view not in available
        }
    finally:
        active_failure = sys.exc_info()[1]
        try:
            ctx.close()
        except BaseException as cleanup_failure:  # noqa: B036 - keep priority
            if active_failure is None:
                raise
            preferred = _retain_context_cleanup_failure(
                active_failure,
                "view validation cleanup also failed",
                cleanup_failure,
            )
            _attach_context_cleanup_owner(preferred, ctx)
            if preferred is cleanup_failure:
                raise cleanup_failure from active_failure