codenib.agent

Agent package consolidating agent utilities and implementations.

Modules:

Name	Description
agent_types	Data types for the agent runner.
boundary	Agent-boundary line-numbering conversion (issue #153).
compile	Agent compile: query-time skill selection.
extract_agent	Keyword extraction agent for problem statements.
harness	Declarative agent harness configuration.
history	Token-budgeted chat history for the agent loop (issue #109, phase 4).
lsp_graph	Graph-backed LSP-shaped navigation helpers.
lsp_provider	LSP-compatible providers over CodeNib static indexes.
rerank_agent	Rerank agent for ranking code nodes based on relevance to a query.
resource_guard	Manifest-aware resource checking for the agent.
route_context	Initial static-graph route context for agent runs.
runner	Lightweight agent runner using LLM tool calling.
runtime	Runtime support types for agent execution.
skills	Declarative skill definitions for CodeNib agents.
tool_schema	Convert skills and tools to OpenAI function-calling tool schemas.
tools	Always-on default tool primitives for the agent.

Classes:

Name	Description
AgentResult	Outcome of AgentRunner.run().
ToolCallRecord	Record of a single tool invocation during an agent run.
KeywordExtraction	Model for keyword extraction output.
KeywordExtractor	Agent for extracting keywords from problem statements.
AgentHarnessSpec	Stable runner-facing harness contract.
AgentRunAccumulator	Accumulate usage and turn counts across related agent runs.
PlainChatHistory	Unbounded chat history with the same interface as the budgeted one.
TokenBudgetedChatHistory	Chat-history container that caps total context tokens.
LSPProviderMetadata	Trace-safe metadata describing how an LSP-shaped call was served.
LSPProviderNodes	List-compatible LSP result carrying provider metadata for traces.
StaticLSPProvider	LSP-shaped provider backed by a loaded CodeNib symbol graph.
RerankAgent	Agent for reranking code nodes based on query relevance using LLM APIs.
RerankResult	Model for reranking output.
LSPRouteContext	Rendered startup context produced by the static graph route.
AgentRunner	LLM-driven agent loop over the CodeNib skill registry.
CodeNibAgentOptions	Configuration for a single query() invocation.
AgentRunTrace	Durable event log for one AgentRunner.run() invocation.
AgentTraceEvent	A replay-oriented event emitted by an agent run.
ContextLedger	Append-only collection of context ledger entries for one agent run.
ContextLedgerEntry	Bounded summary of context produced, retained, or consumed by a run.
RepositoryContextExplorer	Plan and execute repository search over manifest-backed CodeNib views.
RepositoryEvidence	One ranked source span in CodeNib's 0-based coordinate system.
RepositoryExplorePreparation	A deterministic query plan with all selected views loaded.
RepositoryExplorerCapabilityError	The selected exploration policy cannot run over the manifest.
RepositoryExploreResult	Ranked repository evidence and the plan that produced it.
RepositoryExploreTrace	Planner and execution provenance for one exploration request.

Functions:

Name	Description
extract_keywords_from_statement	Extract keywords from a problem statement.
agent_working_directory	Temporarily run agent default tools relative to cwd.
run_agent_in_directory	Run an AgentRunner-like object with repo-relative default tools.
count_message_tokens	Best-effort token count for a list of chat messages.
lsp_result_metadata	Return provider metadata from a list-compatible LSP result.
rerank_nodes_with_query	Convenience function to rerank nodes with a query.
build_lsp_route_context	Run an lsp_route executor and render startup context.
canonical_lsp_route_args	Return the canonical argument shape for a static LSP route call.
extract_lsp_symbol_seeds	Extract explicit symbol-like seeds from task text.
filter_lsp_symbol_seeds	Apply a route startup seed policy while preserving seed order.
fingerprint_lsp_route_nodes	Return a stable fingerprint for an ordered route-node result.
is_specific_lsp_symbol_seed	Return whether seed is specific enough for gated startup routing.
normalize_lsp_route_seed_policy	Normalize and validate the startup route seed policy.
render_lsp_route_context	Render route nodes as compact, unverified prompt context.
compile_repo	Compile indexes for repo_path ahead of time and return the manifest.
has_localization_contract	Return true when an answer carries the localization output contract.
query	Run one agent turn over a repo and return the result.
normalize_repository_explorer_policy	Return the canonical repository-explorer policy name.
repository_explorer_build_views	Return the deterministic materialization set for a benchmark policy.
registry_to_tools	Convert all skills in the registry to OpenAI tool schemas.
skill_to_tool_schema	Convert a single SkillMetadata to an OpenAI function tool dict.

AgentResult dataclass

AgentResult(
    answer: str,
    tool_calls: list[ToolCallRecord] = list(),
    messages: list[dict[str, Any]] = list(),
    total_turns: int = 0,
    total_duration_ms: float = 0.0,
    usage: TokenUsage | None = None,
    usage_records: list[UsageRecord] = list(),
    trace: AgentRunTrace | None = None,
)

Outcome of AgentRunner.run().

ToolCallRecord dataclass

ToolCallRecord(
    tool_call_id: str,
    skill_id: str,
    arguments: dict[str, Any],
    resolved_arguments: dict[str, Any] | None = None,
    result: Any = None,
    duration_ms: float = 0.0,
    error: str | None = None,
)

Record of a single tool invocation during an agent run.

KeywordExtraction

Bases: BaseModel

Model for keyword extraction output.

KeywordExtractor

KeywordExtractor(llm: LiteLLMChat)

Agent for extracting keywords from problem statements.

Parameters:

Name	Type	Description	Default
llm	LiteLLMChat	A configured LiteLLMChat instance.	required

Methods:

Name	Description
extract_keywords	Extract keywords from a problem statement.

Source code in codenib/agent/extract_agent.py

def __init__(
    self,
    llm: LiteLLMChat,
):
    """Initialize the keyword extractor.

    Args:
        llm: A configured LiteLLMChat instance.
    """
    self.llm = llm
    self.structured_llm = self.llm.with_structured_output(KeywordExtraction)

extract_keywords

extract_keywords(problem_statement: str) -> KeywordExtraction

Extract keywords from a problem statement.

Parameters:

Name	Type	Description	Default
problem_statement	str	The problem statement to extract keywords from	required

Returns:

Name	Type	Description
KeywordExtraction	KeywordExtraction	Structured output with extracted keywords

Source code in codenib/agent/extract_agent.py

def extract_keywords(self, problem_statement: str) -> KeywordExtraction:
    """
    Extract keywords from a problem statement.

    Args:
        problem_statement (str): The problem statement to extract keywords from

    Returns:
        KeywordExtraction: Structured output with extracted keywords
    """
    # Create prompt with detailed instructions
    prompt = (
        "You are a keyword extraction specialist. "
        "Your task is to extract important keywords "
        "from problem statements. Focus on identifying "
        "technical terms, function names, class names, "
        "modules, file paths, and concepts that would be "
        "useful for searching in a codebase. "
        "\n\n"
        "Guidelines for extraction:\n"
        "1. Extract file paths and file names "
        "(e.g., 'django/db/models/expressions.py'"
        "-> and 'expressions.py')\n"
        "2. Extract function and method names "
        "(e.g., 'separability_matrix', 'run_validators')\n"
        "3. Extract class names and module names\n"
        "4. Prefer precise terms over general ones\n"
        "5. Remove common stopwords and general "
        "programming terms\n"
        "\n\n"
        "Please extract the key technical terms and "
        "concepts from the following problem statement:"
        f"\n\n{problem_statement}\n\n"
        "Return only the essential terms that would be "
        "most useful for searching in a codebase."
    )

    # Use structured LLM to get output directly as a KeywordExtraction object
    input_msg = human_message(prompt)
    result = self.structured_llm.invoke([input_msg])
    logger.debug(f"Extracted keywords: {result}")
    return result

AgentHarnessSpec dataclass

AgentHarnessSpec(
    max_turns: int = 10,
    max_context_tokens: int | None = None,
    allow_skills: Collection[str] | None = None,
    exclude_skills: Collection[str] | None = None,
    include_default_tools: bool = True,
    default_tool_ids: Collection[str] | None = None,
    system_prompt: str | None = None,
    first_turn_tool_choice: str | None = None,
    force_first_turn_only: bool = False,
    force_localization_contract: bool = False,
    compact_after_read: bool = False,
    compact_keep_reads: int = 0,
    enable_lsp_route_context: bool = False,
    lsp_route_seed_limit: int = 8,
    lsp_route_seed_policy: str = "all",
    lsp_route_query_fallback: bool = False,
    lsp_route_top_k: int = 12,
    lsp_route_include_neighbors: bool = True,
)

Stable runner-facing harness contract.

This deliberately mirrors generic :class:AgentRunner controls only. It does not encode benchmark arms, scorer fields, model names, or dataset identifiers, so experiments can compare policies without leaking those policies into the core runtime API.

Methods:

Name	Description
with_overrides	Return a copy with selected fields replaced and revalidated.
to_runner_kwargs	Return AgentRunner keyword arguments for this harness.
create_runner	Build an AgentRunner using this harness.

with_overrides

with_overrides(**overrides: Any) -> 'AgentHarnessSpec'

Return a copy with selected fields replaced and revalidated.

Source code in codenib/agent/harness.py

def with_overrides(self, **overrides: Any) -> "AgentHarnessSpec":
    """Return a copy with selected fields replaced and revalidated."""
    data = {field.name: getattr(self, field.name) for field in fields(self)}
    unknown = set(overrides) - set(data)
    if unknown:
        raise TypeError(f"Unknown harness option(s): {sorted(unknown)}")
    data.update(overrides)
    return type(self)(**data)

to_runner_kwargs

to_runner_kwargs(
    *,
    session_ctx: Any | None = None,
    manifest: Any | None = None,
    compile_table: Any | None = None,
    extra: Mapping[str, Any] | None = None
) -> dict[str, Any]

Return AgentRunner keyword arguments for this harness.

Collection fields are copied into mutable sets because AgentRunner treats them as constructor inputs and may derive internal sets from them. The spec remains immutable and reusable across cells/subagents.

Source code in codenib/agent/harness.py

def to_runner_kwargs(
    self,
    *,
    session_ctx: Optional[Any] = None,
    manifest: Optional[Any] = None,
    compile_table: Optional[Any] = None,
    extra: Optional[Mapping[str, Any]] = None,
) -> Dict[str, Any]:
    """Return ``AgentRunner`` keyword arguments for this harness.

    Collection fields are copied into mutable sets because ``AgentRunner``
    treats them as constructor inputs and may derive internal sets from
    them. The spec remains immutable and reusable across cells/subagents.
    """
    kwargs: Dict[str, Any] = {
        "max_turns": self.max_turns,
        "max_context_tokens": self.max_context_tokens,
        "allow_skills": (
            set(self.allow_skills) if self.allow_skills is not None else None
        ),
        "exclude_skills": (
            set(self.exclude_skills) if self.exclude_skills is not None else None
        ),
        "include_default_tools": self.include_default_tools,
        "default_tool_ids": (
            set(self.default_tool_ids)
            if self.default_tool_ids is not None
            else None
        ),
        "system_prompt": self.system_prompt,
        "first_turn_tool_choice": self.first_turn_tool_choice,
        "force_first_turn_only": self.force_first_turn_only,
        "force_localization_contract": self.force_localization_contract,
        "compact_after_read": self.compact_after_read,
        "compact_keep_reads": self.compact_keep_reads,
        "enable_lsp_route_context": self.enable_lsp_route_context,
        "lsp_route_seed_limit": self.lsp_route_seed_limit,
        "lsp_route_seed_policy": self.lsp_route_seed_policy,
        "lsp_route_query_fallback": self.lsp_route_query_fallback,
        "lsp_route_top_k": self.lsp_route_top_k,
        "lsp_route_include_neighbors": self.lsp_route_include_neighbors,
        "session_ctx": session_ctx,
        "manifest": manifest,
        "compile_table": compile_table,
    }
    if extra:
        kwargs.update(extra)
    return kwargs

create_runner

create_runner(
    *,
    llm: Any | None = None,
    model: str | None = None,
    registry: Any | None = None,
    session_ctx: Any | None = None,
    manifest: Any | None = None,
    compile_table: Any | None = None,
    **overrides: Any
) -> Any

Build an AgentRunner using this harness.

overrides are one-off runner keyword overrides for cases like subagents with a shorter turn budget. They do not mutate the spec.

Source code in codenib/agent/harness.py

def create_runner(
    self,
    *,
    llm: Optional[Any] = None,
    model: Optional[str] = None,
    registry: Optional[Any] = None,
    session_ctx: Optional[Any] = None,
    manifest: Optional[Any] = None,
    compile_table: Optional[Any] = None,
    **overrides: Any,
) -> Any:
    """Build an ``AgentRunner`` using this harness.

    ``overrides`` are one-off runner keyword overrides for cases like
    subagents with a shorter turn budget. They do not mutate the spec.
    """
    from .runner import AgentRunner

    kwargs = self.to_runner_kwargs(
        session_ctx=session_ctx,
        manifest=manifest,
        compile_table=compile_table,
        extra=overrides,
    )
    return AgentRunner(llm=llm, model=model, registry=registry, **kwargs)

AgentRunAccumulator dataclass

AgentRunAccumulator(
    _usage_values: dict[str, list[float]] = (
        lambda: {key: [] for key in USAGE_TOTAL_KEYS}
    )(),
    _turns: list[int] = list(),
)

Accumulate usage and turn counts across related agent runs.

A harness may charge one logical cell for several LLM interactions: routing gates, isolated subagents, verify retries, or a final convergence run. This helper keeps that accounting generic so experiment scripts do not each reimplement token/turn summation.

Methods:

Name	Description
add_usage	Add a TokenUsage or mapping with flat/nested token fields.
add_turns	Add one run's turn count when it is known.
add_result	Add accounting fields from an AgentResult-like object.
usage_sum	Return the sum for one usage field, or None if never recorded.
usage_totals	Return flat totals with None for fields never observed.
total_turns	Return summed turns, or fallback when no turn count was added.

add_usage

add_usage(usage: Any | None) -> None

Add a TokenUsage or mapping with flat/nested token fields.

Source code in codenib/agent/harness.py

def add_usage(self, usage: Optional[Any]) -> None:
    """Add a ``TokenUsage`` or mapping with flat/nested token fields."""
    if usage is None:
        return
    if isinstance(usage, TokenUsage):
        payload: Mapping[str, Any] = usage.to_dict()
    elif isinstance(usage, Mapping):
        nested = usage.get("token_usage")
        payload = nested if isinstance(nested, Mapping) else usage
    else:
        payload = {
            key: getattr(usage, key, None)
            for key in USAGE_TOTAL_KEYS
            if getattr(usage, key, None) is not None
        }

    for key in USAGE_TOTAL_KEYS:
        value = payload.get(key)
        if value is None:
            continue
        try:
            self._usage_values[key].append(float(value))
        except (TypeError, ValueError):
            continue

add_turns

add_turns(turns: int | None) -> None

Add one run's turn count when it is known.

Source code in codenib/agent/harness.py

def add_turns(self, turns: Optional[int]) -> None:
    """Add one run's turn count when it is known."""
    if turns is None:
        return
    self._turns.append(int(turns))

add_result

add_result(result: Any) -> None

Add accounting fields from an AgentResult-like object.

Source code in codenib/agent/harness.py

def add_result(self, result: Any) -> None:
    """Add accounting fields from an ``AgentResult``-like object."""
    self.add_usage(getattr(result, "usage", None))
    self.add_turns(getattr(result, "total_turns", None))

usage_sum

usage_sum(key: str) -> float | None

Return the sum for one usage field, or None if never recorded.

Source code in codenib/agent/harness.py

def usage_sum(self, key: str) -> Optional[float]:
    """Return the sum for one usage field, or ``None`` if never recorded."""
    if key not in self._usage_values:
        raise KeyError(f"Unknown usage field: {key}")
    values = self._usage_values[key]
    if not values:
        return None
    total = sum(values)
    if key != "cost_usd" and total.is_integer():
        return int(total)
    return total

usage_totals

usage_totals() -> dict[str, float | None]

Return flat totals with None for fields never observed.

Source code in codenib/agent/harness.py

def usage_totals(self) -> Dict[str, Optional[float]]:
    """Return flat totals with ``None`` for fields never observed."""
    return {key: self.usage_sum(key) for key in USAGE_TOTAL_KEYS}

total_turns

total_turns(*, fallback: int | None = None) -> int | None

Return summed turns, or fallback when no turn count was added.

Source code in codenib/agent/harness.py

def total_turns(self, *, fallback: Optional[int] = None) -> Optional[int]:
    """Return summed turns, or ``fallback`` when no turn count was added."""
    if not self._turns:
        return fallback
    return sum(self._turns)

PlainChatHistory

PlainChatHistory(messages: list[dict[str, Any]] | None = None)

Unbounded chat history with the same interface as the budgeted one.

Lets :class:~codenib.agent.runner.AgentRunner use a single code path whether or not a token budget is configured. Behaviour is identical to a bare list of message dicts — nothing is ever evicted.

Source code in codenib/agent/history.py

def __init__(self, messages: Optional[List[Dict[str, Any]]] = None) -> None:
    self._messages: List[Dict[str, Any]] = list(messages or [])

TokenBudgetedChatHistory

TokenBudgetedChatHistory(
    max_tokens: int, *, model: str | None = None, keep_last: int = 2
)

Chat-history container that caps total context tokens.

Pinned system messages are always retained. When appending a message would push the estimated token total above max_tokens, the oldest non-system messages are evicted one at a time (oldest first) until the history fits again, so the system prompt plus the most recent turns are preserved.

keep_last guards the tail: the most recent keep_last non-system messages are never evicted even under budget pressure, so the model always sees the immediate context it needs to act. A single message that on its own exceeds the budget is kept (we never drop the message just added) and a warning is logged.

The container is intentionally minimal — add_message / get_messages / clear / __len__ / __iter__ — so the runner can use it in place of a bare list.

Methods:

Name	Description
get_messages	Return the live message list (the object the LLM is called with).
clear	Drop all messages.
add_message	Append message, then evict oldest non-system messages if over budget.
extend	Append several messages, enforcing the budget after each one.
total_tokens	Estimated token total of the current history.

Source code in codenib/agent/history.py

def __init__(
    self,
    max_tokens: int,
    *,
    model: Optional[str] = None,
    keep_last: int = 2,
) -> None:
    if max_tokens <= 0:
        raise ValueError(f"max_tokens must be positive, got {max_tokens}")
    self.max_tokens = max_tokens
    self.model = model
    self.keep_last = max(0, keep_last)
    self._messages: List[Dict[str, Any]] = []

get_messages

get_messages() -> list[dict[str, Any]]

Return the live message list (the object the LLM is called with).

Source code in codenib/agent/history.py

def get_messages(self) -> List[Dict[str, Any]]:
    """Return the live message list (the object the LLM is called with)."""
    return self._messages

clear

clear() -> None

Drop all messages.

Source code in codenib/agent/history.py

def clear(self) -> None:
    """Drop all messages."""
    self._messages.clear()

add_message

add_message(message: dict[str, Any]) -> None

Append message, then evict oldest non-system messages if over budget.

Source code in codenib/agent/history.py

def add_message(self, message: Dict[str, Any]) -> None:
    """Append *message*, then evict oldest non-system messages if over budget."""
    self._messages.append(message)
    self._enforce_budget()

extend

extend(messages: list[dict[str, Any]]) -> None

Append several messages, enforcing the budget after each one.

Source code in codenib/agent/history.py

def extend(self, messages: List[Dict[str, Any]]) -> None:
    """Append several messages, enforcing the budget after each one."""
    for message in messages:
        self.add_message(message)

total_tokens

total_tokens() -> int

Estimated token total of the current history.

Source code in codenib/agent/history.py

def total_tokens(self) -> int:
    """Estimated token total of the current history."""
    return count_message_tokens(self._messages, model=self.model)

LSPProviderMetadata dataclass

LSPProviderMetadata(
    provider: str,
    capability: str,
    status: str,
    lsp_method: str,
    backend: str | None = None,
    index_snapshot: str | None = None,
    fallback_reason: str | None = None,
    behavior_contract: str = _GRAPH_BEHAVIOR_CONTRACT,
    position_granularity: str = "line",
    position_encoding: str | None = None,
)

Trace-safe metadata describing how an LSP-shaped call was served.

LSPProviderNodes

LSPProviderNodes(nodes: Iterable[Any] = (), *, metadata: LSPProviderMetadata)

Bases: list

List-compatible LSP result carrying provider metadata for traces.

Source code in codenib/agent/lsp_provider.py

def __init__(
    self,
    nodes: Iterable[Any] = (),
    *,
    metadata: LSPProviderMetadata,
) -> None:
    super().__init__(nodes)
    self.lsp_provider_metadata = metadata

StaticLSPProvider

StaticLSPProvider(
    graph: Any,
    *,
    snapshot_id: str | None = None,
    occurrence_index: Any | None = None,
    backend: str | None = None,
    fallback_reason: str | None = None
)

LSP-shaped provider backed by a loaded CodeNib symbol graph.

Methods:

Name	Description
can_serve	Return a non-throwing fast-path decision for one capability.
definition	Serve textDocument/definition from the static graph.
references	Serve textDocument/references from the static graph.
route	Serve CodeNib's LSP-shaped route extension from the static graph.

Source code in codenib/agent/lsp_provider.py

def __init__(
    self,
    graph: Any,
    *,
    snapshot_id: Optional[str] = None,
    occurrence_index: Optional[Any] = None,
    backend: Optional[str] = None,
    fallback_reason: Optional[str] = None,
) -> None:
    self.graph = graph
    self.occurrence_index = occurrence_index or getattr(
        graph, "lsp_occurrence_index", None
    )
    self.snapshot_id = snapshot_id or _graph_snapshot_id(graph)
    self.backend = backend
    self.fallback_reason = fallback_reason
    self.position_encoding = getattr(
        self.occurrence_index, "position_encoding", None
    )

can_serve

can_serve(capability: str) -> LSPProviderMetadata

Return a non-throwing fast-path decision for one capability.

Source code in codenib/agent/lsp_provider.py

def can_serve(self, capability: str) -> LSPProviderMetadata:
    """Return a non-throwing fast-path decision for one capability."""

    normalized = _normalize_capability(capability)
    if normalized not in _SUPPORTED_CAPABILITIES:
        return _metadata(
            normalized,
            status="unsupported",
            snapshot_id=self.snapshot_id,
            backend=self.backend,
            fallback_reason="unsupported_capability",
        )
    if (
        normalized in {CAPABILITY_DEFINITION, CAPABILITY_REFERENCES}
        and self.occurrence_index is not None
    ):
        return _metadata(
            normalized,
            status="ok",
            snapshot_id=self.snapshot_id,
            backend=self.backend,
            fallback_reason=self.fallback_reason,
            behavior_contract=self._occurrence_behavior_contract(),
            position_granularity="character",
            position_encoding=self.position_encoding,
        )
    if (
        normalized == CAPABILITY_ROUTE
        and getattr(self.graph, "supports_graph_routes", True) is False
    ):
        return _metadata(
            normalized,
            status="unavailable",
            snapshot_id=self.snapshot_id,
            backend=self.backend,
            fallback_reason="fact_index_does_not_support_routes",
        )
    if self.graph is None:
        return _metadata(
            normalized,
            status="unavailable",
            snapshot_id=None,
            backend=self.backend,
            fallback_reason="symbol_graph_unavailable",
        )
    return _metadata(
        normalized,
        status="ok",
        snapshot_id=self.snapshot_id,
        backend=self.backend,
        fallback_reason=self.fallback_reason,
    )

definition

definition(
    *,
    file_path: str | None = None,
    line: int | None = None,
    character: int | None = None,
    symbol: str | None = None,
    top_k: int = 8
) -> LSPProviderNodes

Serve textDocument/definition from the static graph.

Source code in codenib/agent/lsp_provider.py

def definition(
    self,
    *,
    file_path: Optional[str] = None,
    line: Optional[int] = None,
    character: Optional[int] = None,
    symbol: Optional[str] = None,
    top_k: int = 8,
) -> LSPProviderNodes:
    """Serve ``textDocument/definition`` from the static graph."""

    self._require(CAPABILITY_DEFINITION)
    if (
        self.occurrence_index is not None
        and file_path is not None
        and line is not None
        and character is not None
        and symbol is None
    ):
        try:
            locations = self.occurrence_index.definitions(
                file_path=file_path,
                line=line,
                character=character,
                top_k=top_k,
            )
        except LSPPositionFallback:
            raise
        except ValueError:
            locations = []
        if locations or getattr(
            self.occurrence_index, "authoritative_empty", False
        ):
            return self._wrap(
                CAPABILITY_DEFINITION,
                _nodes_from_locations(locations, capability=CAPABILITY_DEFINITION),
                behavior_contract=self._occurrence_behavior_contract(),
                position_granularity="character",
                position_encoding=self.position_encoding,
            )
    nodes = lsp_definition(
        self.graph,
        file_path=file_path,
        line=line,
        character=character,
        symbol=symbol,
        top_k=top_k,
    )
    position_query = (
        file_path is not None
        and line is not None
        and character is not None
        and symbol is None
    )
    return self._wrap(
        CAPABILITY_DEFINITION,
        nodes,
        behavior_contract=(
            _GRAPH_POSITION_BEHAVIOR_CONTRACT
            if position_query
            else _GRAPH_BEHAVIOR_CONTRACT
        ),
        position_granularity="character" if position_query else "line",
        position_encoding="UTF32" if position_query else None,
    )

references

references(
    *,
    file_path: str | None = None,
    line: int | None = None,
    character: int | None = None,
    symbol: str | None = None,
    include_declaration: bool = True,
    top_k: int = 40
) -> LSPProviderNodes

Serve textDocument/references from the static graph.

Source code in codenib/agent/lsp_provider.py

def references(
    self,
    *,
    file_path: Optional[str] = None,
    line: Optional[int] = None,
    character: Optional[int] = None,
    symbol: Optional[str] = None,
    include_declaration: bool = True,
    top_k: int = 40,
) -> LSPProviderNodes:
    """Serve ``textDocument/references`` from the static graph."""

    self._require(CAPABILITY_REFERENCES)
    if (
        self.occurrence_index is not None
        and file_path is not None
        and line is not None
        and character is not None
        and symbol is None
    ):
        try:
            locations = self.occurrence_index.references(
                file_path=file_path,
                line=line,
                character=character,
                include_declaration=include_declaration,
                top_k=top_k,
            )
        except LSPPositionFallback:
            raise
        except ValueError:
            locations = []
        if locations or getattr(
            self.occurrence_index, "authoritative_empty", False
        ):
            return self._wrap(
                CAPABILITY_REFERENCES,
                _nodes_from_locations(locations, capability=CAPABILITY_REFERENCES),
                behavior_contract=self._occurrence_behavior_contract(),
                position_granularity="character",
                position_encoding=self.position_encoding,
            )
    nodes = lsp_references(
        self.graph,
        file_path=file_path,
        line=line,
        character=character,
        symbol=symbol,
        include_declaration=include_declaration,
        top_k=top_k,
    )
    position_query = (
        file_path is not None
        and line is not None
        and character is not None
        and symbol is None
    )
    return self._wrap(
        CAPABILITY_REFERENCES,
        nodes,
        behavior_contract=(
            _GRAPH_POSITION_BEHAVIOR_CONTRACT
            if position_query
            else _GRAPH_BEHAVIOR_CONTRACT
        ),
        position_granularity="character" if position_query else "line",
        position_encoding="UTF32" if position_query else None,
    )

route

route(
    *,
    symbols: Sequence[str],
    query: str | None = None,
    top_k: int = 12,
    include_neighbors: bool = True
) -> LSPProviderNodes

Serve CodeNib's LSP-shaped route extension from the static graph.

Source code in codenib/agent/lsp_provider.py

def route(
    self,
    *,
    symbols: Sequence[str],
    query: Optional[str] = None,
    top_k: int = 12,
    include_neighbors: bool = True,
) -> LSPProviderNodes:
    """Serve CodeNib's LSP-shaped route extension from the static graph."""

    self._require(CAPABILITY_ROUTE)
    nodes = lsp_route(
        self.graph,
        symbols=symbols,
        query=query,
        top_k=top_k,
        include_neighbors=include_neighbors,
    )
    return self._wrap(CAPABILITY_ROUTE, nodes, position_granularity="symbol")

RerankAgent

RerankAgent(
    llm: LiteLLMChat, listwise_format: Literal["structured", "rankgpt"] = "structured"
)

Agent for reranking code nodes based on query relevance using LLM APIs.

Parameters:

Name	Type	Description	Default
llm	LiteLLMChat	A configured LiteLLMChat instance.	required
listwise_format	Literal['structured', 'rankgpt']	How to ask the LLM to format the ranking. "structured" (default) uses a JSON schema enforced via with_structured_output; suitable for general-purpose LLMs that follow JSON-mode prompts well. "rankgpt" uses SweRank/RankGPT-style text output ([3] > [5] > [1] > ...) and a regex parser. Required for listwise rerankers fine-tuned on this format (Salesforce/SweRankLLM-*, RankZephyr, etc.) — forcing JSON on them collapses the output to the first index only.	'structured'

Methods:

Name	Description
rerank_nodes	Rerank nodes based on their relevance to the query.

Source code in codenib/agent/rerank_agent.py

def __init__(
    self,
    llm: LiteLLMChat,
    listwise_format: Literal["structured", "rankgpt"] = "structured",
):
    """Initialize the rerank agent.

    Args:
        llm: A configured LiteLLMChat instance.
        listwise_format: How to ask the LLM to format the ranking.
            ``"structured"`` (default) uses a JSON schema enforced via
            ``with_structured_output``; suitable for general-purpose LLMs
            that follow JSON-mode prompts well.
            ``"rankgpt"`` uses SweRank/RankGPT-style text output
            (``[3] > [5] > [1] > ...``) and a regex parser. Required for
            listwise rerankers fine-tuned on this format
            (Salesforce/SweRankLLM-*, RankZephyr, etc.) — forcing JSON on
            them collapses the output to the first index only.
    """
    if listwise_format not in _LISTWISE_FORMATS:
        supported = ", ".join(sorted(_LISTWISE_FORMATS))
        raise ValueError(f"listwise_format must be one of: {supported}")

    self.llm = llm
    self.listwise_format = listwise_format
    self.structured_llm = (
        self.llm.with_structured_output(RerankResult)
        if listwise_format == "structured"
        else None
    )
    logger.info(
        "Initialized rerank agent with model=%s listwise_format=%s",
        self.llm.model,
        listwise_format,
    )

rerank_nodes

rerank_nodes(
    query: str,
    nodes: list[NodeInfo],
    top_k: int | None = None,
    window_size: int | None = None,
    window_step: int | None = None,
    include_content: bool = False,
) -> list[QueriedNode]

Rerank nodes based on their relevance to the query.

Parameters:

Name	Type	Description	Default
query	str	The query to rank nodes against	required
nodes	list[NodeInfo]	List of nodes with content to rank	required
top_k	int | None	Maximum number of results to return (None for all)	None
window_size	int | None	Number of nodes per rerank window. None -> all nodes.	None
window_step	int | None	Step size between sliding windows. Defaults to window_size.	None
include_content	bool	Whether to include node content in the result objects.	False

Returns:

Type	Description
list[QueriedNode]	List[QueriedNode]: Ranked nodes with relevance scores (optionally with content)

Notes

When a sliding window is configured, each window is reranked independently and the averaged scores across all windows determine the final ordering.

Source code in codenib/agent/rerank_agent.py

def rerank_nodes(
    self,
    query: str,
    nodes: List[NodeInfo],
    top_k: Optional[int] = None,
    window_size: Optional[int] = None,
    window_step: Optional[int] = None,
    include_content: bool = False,
) -> List[QueriedNode]:
    """
    Rerank nodes based on their relevance to the query.

    Args:
        query (str): The query to rank nodes against
        nodes (List[NodeInfo]): List of nodes with content to rank
        top_k (Optional[int]): Maximum number of results to return (None for all)
        window_size (Optional[int]): Number of nodes per rerank window. None -> all nodes.
        window_step (Optional[int]): Step size between sliding
            windows. Defaults to window_size.
        include_content (bool): Whether to include node content in the result objects.

    Returns:
        List[QueriedNode]: Ranked nodes with relevance scores (optionally with content)

    Notes:
        When a sliding window is configured, each window is reranked independently and
        the averaged scores across all windows determine the final ordering.
    """
    if top_k is not None and (
        isinstance(top_k, bool) or not isinstance(top_k, int) or top_k < 0
    ):
        raise ValueError("top_k must be a non-negative integer or None")
    if top_k == 0:
        return []
    if not nodes:
        logger.warning("No nodes provided for reranking")
        return []

    if not query.strip():
        logger.warning("Empty query provided for reranking")
        return []

    try:
        # Filter nodes with content
        valid_nodes: List[Tuple[int, NodeInfo]] = []
        for i, node in enumerate(nodes):
            if node.content and node.content.strip():
                valid_nodes.append((i, node))

        if not valid_nodes:
            logger.warning(
                "No nodes with content found for reranking; preserving "
                "the first-stage candidate order."
            )
            return self._preserve_first_stage(
                nodes,
                top_k=top_k,
                include_content=include_content,
            )

        logger.info(
            f"Reranking {len(valid_nodes)} nodes with query: {query[:100]}..."
        )

        node_lookup: Dict[int, NodeInfo] = dict(enumerate(nodes))

        total_nodes = len(valid_nodes)
        window_size = window_size or total_nodes
        if window_size <= 0:
            window_size = total_nodes

        window_step = window_step or window_size
        if window_step <= 0:
            window_step = window_size

        aggregated_scores: Dict[int, float] = defaultdict(float)
        appearance_count: Counter[int] = Counter()

        window_starts = list(range(0, total_nodes, window_step))
        tail_start = max(total_nodes - window_size, 0)
        if tail_start not in window_starts:
            window_starts.append(tail_start)
        window_starts = sorted(set(window_starts))

        window_id = 0
        for window_start in window_starts:
            window_nodes = valid_nodes[window_start : window_start + window_size]
            if not window_nodes:
                continue

            window_id += 1
            logger.debug(
                "Processing rerank window %s (nodes %s-%s)",
                window_id,
                window_start,
                window_start + len(window_nodes) - 1,
            )

            window_results = self._rerank_window(query, window_nodes)
            for original_idx, score in window_results:
                aggregated_scores[original_idx] += float(score)
                appearance_count[original_idx] += 1

        if not aggregated_scores:
            logger.warning(
                "Reranker did not return any usable scores; preserving "
                "the first-stage candidate order."
            )

        averaged_scores = {
            idx: aggregated_scores[idx] / appearance_count[idx]
            for idx in aggregated_scores
            if appearance_count[idx] > 0
        }

        sorted_indices = sorted(
            averaged_scores.items(), key=lambda item: item[1], reverse=True
        )
        ranked_ids = {idx for idx, _score in sorted_indices}
        sorted_indices.extend(
            (
                original_idx,
                float(node.score) if node.score is not None else 0.0,
            )
            for original_idx, node in enumerate(nodes)
            if original_idx not in ranked_ids
        )

        ranked_nodes: List[QueriedNode] = []
        for original_idx, score in sorted_indices:
            node = node_lookup.get(original_idx)
            if not node:
                continue
            ranked_nodes.append(
                self._to_queried_node(
                    node,
                    score=float(score),
                    include_content=include_content,
                )
            )
            if top_k and len(ranked_nodes) >= top_k:
                break

        logger.info(
            "Successfully reranked %s nodes across %s window(s)",
            len(ranked_nodes),
            window_id or 1,
        )
        return ranked_nodes

    except Exception as e:
        logger.error(f"Error during reranking: {e}")
        return self._preserve_first_stage(
            nodes,
            top_k=top_k,
            include_content=include_content,
        )

RerankResult

Bases: BaseModel

Model for reranking output.

LSPRouteContext dataclass

LSPRouteContext(
    seeds: tuple[str, ...],
    nodes: tuple[Any, ...],
    text: str = "",
    arguments: Mapping[str, Any] | None = None,
    route_fingerprint: str | None = None,
)

Rendered startup context produced by the static graph route.

AgentRunner

AgentRunner(
    llm: LiteLLMChat | None = None,
    registry: SkillRegistry | None = None,
    *,
    model: str | None = None,
    temperature: float = 0.0,
    max_tokens: int = 512,
    system_prompt: str | None = None,
    max_turns: int = 10,
    max_context_tokens: int | None = None,
    allow_skills: set[str] | None = None,
    exclude_skills: set[str] | None = None,
    manifest: Any | None = None,
    session_ctx: Any | None = None,
    compile_table: Any | None = None,
    include_default_tools: bool = True,
    default_tool_ids: set[str] | None = None,
    sandbox: "SandboxSession" | None = None,
    retry: RetryConfig | None = None,
    force_localization_contract: bool = False,
    force_final_answer: bool = False,
    review_final_answer: bool = False,
    first_turn_tool_choice: str | None = None,
    force_first_turn_only: bool = False,
    compact_after_read: bool = False,
    compact_keep_reads: int = 0,
    enable_lsp_route_context: bool = False,
    lsp_route_seed_limit: int = 8,
    lsp_route_seed_policy: str = "all",
    lsp_route_query_fallback: bool = False,
    lsp_route_top_k: int = 12,
    lsp_route_include_neighbors: bool = True
)

LLM-driven agent loop over the CodeNib skill registry.

Usage::

from codenib.agent.skills.registry import SkillRegistry

runner = AgentRunner(model="gpt-4o", registry=SkillRegistry())
result = runner.run("How does authentication work in this repo?")
print(result.answer)

Methods:

Name	Description
run	Execute the agent loop and return the result.

Source code in codenib/agent/runner.py

def __init__(
    self,
    llm: Optional[LiteLLMChat] = None,
    registry: Optional[SkillRegistry] = None,
    *,
    model: Optional[str] = None,
    temperature: float = 0.0,
    max_tokens: int = 512,
    system_prompt: Optional[str] = None,
    max_turns: int = 10,
    max_context_tokens: Optional[int] = None,
    allow_skills: Optional[Set[str]] = None,
    exclude_skills: Optional[Set[str]] = None,
    manifest: Optional[Any] = None,
    session_ctx: Optional[Any] = None,
    compile_table: Optional[Any] = None,
    include_default_tools: bool = True,
    default_tool_ids: Optional[Set[str]] = None,
    sandbox: Optional["SandboxSession"] = None,
    retry: Optional[RetryConfig] = None,
    force_localization_contract: bool = False,
    force_final_answer: bool = False,
    review_final_answer: bool = False,
    first_turn_tool_choice: Optional[str] = None,
    force_first_turn_only: bool = False,
    compact_after_read: bool = False,
    compact_keep_reads: int = 0,
    enable_lsp_route_context: bool = False,
    lsp_route_seed_limit: int = 8,
    lsp_route_seed_policy: str = "all",
    lsp_route_query_fallback: bool = False,
    lsp_route_top_k: int = 12,
    lsp_route_include_neighbors: bool = True,
) -> None:
    if llm is not None:
        self.llm = llm
    elif model is not None:
        self.llm = LiteLLMChat(
            model=model,
            temperature=temperature,
            max_tokens=max_tokens,
            retry=retry or RetryConfig(),
        )
    else:
        raise ValueError("Either 'llm' or 'model' must be provided")
    # Optional per-run context-window cap. When set, ``run()`` keeps the
    # conversation in a :class:`TokenBudgetedChatHistory` that evicts the
    # oldest non-system messages once the budget would be exceeded, so a
    # long tool-calling loop can't overflow the model context.
    self.max_context_tokens = max_context_tokens
    self.registry = registry or SkillRegistry()
    self._sandbox = sandbox
    if sandbox is not None and manifest is None and self.registry.list_skills():
        raise ValueError(
            "sandbox execution without a manifest cannot expose retrieval "
            "skills because their source snapshot is unbound"
        )
    if sandbox is not None and manifest is not None:
        manifest_revision = str(getattr(manifest, "commit", "") or "")
        sandbox_revision = str(sandbox.metadata.source_revision or "")
        if not manifest_revision:
            raise ValueError(
                "sandbox execution requires a revision-bearing RepoManifest"
            )
        if manifest_revision != sandbox_revision:
            raise ValueError(
                "sandbox source revision does not match RepoManifest.commit: "
                f"{sandbox_revision!r} != {manifest_revision!r}"
            )
        manifest_fingerprint = str(
            getattr(manifest, "source_fingerprint", "") or ""
        )
        sandbox_fingerprint = str(sandbox.metadata.source_fingerprint or "")
        if not manifest_fingerprint or not sandbox_fingerprint:
            raise ValueError(
                "sandbox execution requires matching source fingerprints in "
                "the manifest and sandbox metadata"
            )
        if not (
            is_secure_source_fingerprint_v2(manifest_fingerprint)
            and is_secure_source_fingerprint_v2(sandbox_fingerprint)
        ):
            raise ValueError(
                "sandbox execution requires matching secure source "
                "fingerprint v2 identities"
            )
        if manifest_fingerprint != sandbox_fingerprint:
            raise ValueError(
                "sandbox source fingerprint does not match RepoManifest: "
                f"{sandbox_fingerprint!r} != {manifest_fingerprint!r}"
            )
    # Default tools (read + grep + glob + bash) are a SEPARATE type from
    # retrieval skills: they live in their own ``ToolRegistry`` and never
    # pass through the skill allow/exclude/compile_table funnel. Registered
    # unconditionally so every query — and every agent-compile subset — has
    # the filesystem primitives the model is pretrained on.
    # ``include_default_tools=False`` withholds them — used to force the
    # structured (retrieval + graph) path in cost-comparison experiments.
    self.tool_registry = ToolRegistry()
    self._include_defaults = include_default_tools
    self._default_ids: Set[str] = set()
    if include_default_tools:
        if sandbox is None:
            ensure_default_tools_registered(self.tool_registry)
        else:
            # Import only for the explicit isolated path. Ordinary local
            # queries do not import a container backend or contact Docker.
            from .tools.sandbox import ensure_sandbox_tools_registered

            ensure_sandbox_tools_registered(self.tool_registry, sandbox)
        # Which of the registered defaults to actually expose. Defaults to
        # ALL (read + grep + glob + bash). Restricting to a subset — e.g.
        # ``{"read"}`` — yields a graph-primary / LocAgent-style harness
        # that can read code but has NO grep escape hatch, so the structured
        # (bm25 + call-graph) tools must carry navigation.
        allowed = set(DEFAULT_TOOL_IDS)
        # A non-empty restrictor narrows to that subset; ``None`` *or* an
        # empty set means "all defaults" (the off-switch is
        # ``include_default_tools=False``, not an empty restrictor) — this
        # mirrors the empty->all skills contract in ``_resolve_allow_set``.
        if default_tool_ids:
            allowed &= set(default_tool_ids)
        self._default_ids = allowed

    # Tools and skills share the model-facing namespace: a skill whose id
    # equals a default-tool id would emit a duplicate function name (which
    # providers reject) and shadow the tool at dispatch. Fail loudly here.
    collisions = self._default_ids & set(self.registry.list_skills())
    if collisions:
        raise ValueError(
            f"skill id(s) {sorted(collisions)} collide with default tool "
            f"id(s); rename the skill(s) — tools and skills share the "
            f"model-facing tool namespace."
        )
    self.max_turns = max_turns
    self.session_ctx = session_ctx
    # The localization eval needs answers to end in the Files:/Symbols:/
    # Locations: contract; QA callers (web demo) keep prose, so they turn
    # this off and skip the schema-forcing final turn entirely.
    self._force_contract = force_localization_contract
    # Chat callers (the web demo) want a usable prose answer even when the
    # exploration budget runs out: one extra tool-free turn asks the model
    # to commit to an answer from what it already read. Opt-in so eval
    # behaviour is untouched; an empty forced answer raises instead of
    # silently returning mid-exploration chatter.
    self._final_answer = force_final_answer
    # Grounded QA callers can require one evidence-audit pass after the
    # first prose draft. The audit retains tool access, so it can repair a
    # missing call-site trace instead of merely rephrasing the same answer.
    # Opt-in keeps localization experiments and their turn budgets stable.
    self._review_final_answer = review_final_answer
    self.first_turn_tool_choice = first_turn_tool_choice
    # When True, only turn 0 is forced (legacy single-turn behaviour);
    # default False = force until the agent reads a file.
    self._force_first_turn_only = force_first_turn_only
    # eager_compact: after the first successful read anchors the agent, stub
    # the bulky tool outputs already in history (full file/grep text) down to
    # short references so later turns stop re-carrying them — the dominant
    # prompt-token cost (prompt is ~97% of spend and grows every turn).
    # Loss-safe for localization (answers need path/symbol/line, not source).
    self._compact_after_read = compact_after_read
    # Seed richness for eager_compact: how many successful read outputs to
    # keep VERBATIM in the collapse seed. For backward compatibility, 0 is
    # an alias for the minimum of one read; values 2+ retain more history.
    self._compact_keep_reads = compact_keep_reads
    # Optional startup context over the same static graph exposed by the
    # lsp_route skill. This is opt-in harness policy: it makes graph route
    # hints available on turn 1 without depending on the model to discover
    # the tool, while still requiring normal reads before answering.
    self._enable_lsp_route_context = bool(enable_lsp_route_context)
    self._lsp_route_seed_limit = max(1, int(lsp_route_seed_limit or 8))
    self._lsp_route_seed_policy = normalize_lsp_route_seed_policy(
        lsp_route_seed_policy
    )
    self._lsp_route_query_fallback = bool(lsp_route_query_fallback)
    self._lsp_route_top_k = max(1, int(lsp_route_top_k or 12))
    self._lsp_route_include_neighbors = bool(lsp_route_include_neighbors)

    # Resource guard: filter unavailable skills and collect warnings.
    # The "base" allow / exclude are stored so we can recompute the
    # tool list per-query when a compile_table is in play.
    self._base_allow: Optional[Set[str]] = (
        set(allow_skills) if allow_skills is not None else None
    )
    self._base_exclude: Set[str] = set(exclude_skills) if exclude_skills else set()
    resource_warnings: List[str] = []

    if manifest is not None:
        from .resource_guard import ResourceGuard

        guard = ResourceGuard(manifest, self.registry)
        report = guard.preflight()
        self._base_exclude |= report.unavailable
        resource_warnings = report.warnings

    # ``_base_exclude`` applies to *skills* only. Default tools live in a
    # separate ``ToolRegistry`` (and are gated by ``_default_ids``), so no
    # skill exclude — caller's ``exclude_skills`` or the ResourceGuard's
    # ``report.unavailable`` — can ever reach them.

    self._compile_table = compile_table

    # Pre-compute the static tool list. When a compile_table is set,
    # ``run()`` recomputes per-query against the resolved allow set.
    self.tools = self._tools_for(self._base_allow)

    # Keep prompt policy coupled to the model-facing tool list. A compile
    # table can narrow that list per query, so run() renders it again after
    # resolving the query-specific subset.
    self._custom_system_prompt = system_prompt or None
    self._environment_prompt_block = self._build_environment_block(
        self.session_ctx, sandbox=self._sandbox
    )
    self._resource_warnings = list(resource_warnings)
    self.system_prompt = self._render_system_prompt(self.tools)

run

run(
    query: str,
    *,
    max_turns: int | None = None,
    chat_history: list[dict[str, str]] | None = None
) -> AgentResult

Execute the agent loop and return the result.

chat_history seeds prior conversation turns (text-only {"role": "user"|"assistant", "content": ...} dicts, no tool messages) between the system prompt and query, so follow-up questions can reference earlier answers.

Source code in codenib/agent/runner.py

def run(
    self,
    query: str,
    *,
    max_turns: Optional[int] = None,
    chat_history: Optional[List[Dict[str, str]]] = None,
) -> AgentResult:
    """Execute the agent loop and return the result.

    ``chat_history`` seeds prior conversation turns (text-only
    ``{"role": "user"|"assistant", "content": ...}`` dicts, no tool
    messages) between the system prompt and *query*, so follow-up
    questions can reference earlier answers.
    """
    max_turns = max_turns or self.max_turns

    # CAR / agent_compile: when a compile_table is set, classify the
    # query and intersect the table-resolved subset with the
    # constructor-time allow set. The table can *narrow* allowed
    # skills, never broaden them, so the user's explicit
    # ``allow_skills`` remains the upper bound.
    tools = self.tools
    if self._compile_table is not None:
        from .compile import agent_compile

        table_allow = agent_compile(query, self.session_ctx, self._compile_table)
        if table_allow is not None:
            effective: Optional[Set[str]]
            if self._base_allow is None:
                effective = set(table_allow)
            else:
                effective = set(table_allow) & self._base_allow
                # A compile_table subset that is disjoint from a
                # non-empty allow_skills upper bound must NOT broaden
                # back to the full registry via the empty→full
                # fallback. The table can only narrow, never broaden,
                # so fall back to the upper bound itself.
                if not effective and self._base_allow:
                    logger.warning(
                        "AgentRunner: compile_table scenario is disjoint "
                        "from the allow_skills upper bound %s; keeping "
                        "allow_skills (table narrows, never broadens)",
                        sorted(self._base_allow),
                    )
                    effective = set(self._base_allow)
            tools = self._tools_for(effective)
    active_tool_ids = self._tool_names(tools)

    all_tool_calls: List[ToolCallRecord] = []
    usage_tracker = UsageTracker()
    start = time.monotonic()
    trace = AgentRunTrace()
    trace.add(
        "run_start",
        0,
        query_chars=len(query or ""),
        max_turns=max_turns,
        tool_count=len(tools or []),
        sandbox=(
            self._sandbox.metadata.to_dict() if self._sandbox is not None else None
        ),
    )
    user_query = query
    if self._enable_lsp_route_context:
        route_start = time.monotonic()
        route_context, route_skip_reason = self._initial_lsp_route_context(query)
        route_duration_ms = (time.monotonic() - route_start) * 1000
        if route_context is not None and route_context.text:
            user_query = f"{query}\n\n{route_context.text}"
            seeds = list(route_context.seeds)
            seed_source = "symbol" if seeds else "query"
            trace.add(
                "lsp_route_context",
                0,
                status="offered",
                seeds=seeds,
                seed_source=seed_source,
                seed_policy=self._lsp_route_seed_policy,
                route_count=len(route_context.nodes),
                context_chars=len(route_context.text),
                duration_ms=route_duration_ms,
                route_args=dict(route_context.arguments or {}),
                route_fingerprint=route_context.route_fingerprint,
                route_preview=summarize_lsp_route_nodes(route_context.nodes),
            )
            trace.add_context(
                "lsp_route",
                "offered",
                0,
                summary=(
                    "initial static LSP route context from "
                    + ("seeds: " + ", ".join(seeds) if seeds else "query text")
                ),
                provenance={
                    "kind": "initial_context",
                    "tool": "lsp_route",
                },
                freshness="fresh",
                token_estimate=_estimate_context_tokens(route_context.text),
                metadata={
                    "seeds": seeds,
                    "seed_source": seed_source,
                    "seed_policy": self._lsp_route_seed_policy,
                    "route_count": len(route_context.nodes),
                    "top_k": self._lsp_route_top_k,
                    "include_neighbors": self._lsp_route_include_neighbors,
                    "query_fallback": self._lsp_route_query_fallback,
                    "duration_ms": route_duration_ms,
                    "route_args": dict(route_context.arguments or {}),
                    "route_fingerprint": route_context.route_fingerprint,
                },
            )
        else:
            trace.add(
                "lsp_route_context",
                0,
                status="skipped",
                reason=route_skip_reason or "empty_route_context",
                duration_ms=route_duration_ms,
            )

    history = self._new_history()
    run_system_prompt = self._render_system_prompt(tools)
    history.add_message({"role": "system", "content": run_system_prompt})
    for msg in chat_history or []:
        history.add_message({"role": msg["role"], "content": msg["content"]})
    history.add_message({"role": "user", "content": user_query})
    has_read = False  # has the agent read a file yet (gates forced tools)
    read_paths: List[str] = []  # files the agent read (for schema salvage)
    read_outputs: List[Dict[str, str]] = []  # successful read transcripts
    compacted = False  # eager_compact: collapsed to direction seed yet?
    answer_reviewed = False

    def _finish_result(
        answer: str,
        *,
        total_turns: int,
        answer_source: str,
    ) -> AgentResult:
        elapsed = (time.monotonic() - start) * 1000
        trace.add(
            "final_answer",
            total_turns,
            source=answer_source,
            answer_chars=len(answer or ""),
            has_contract=_has_localization_contract(answer or ""),
        )
        trace.add(
            "run_end",
            total_turns,
            duration_ms=elapsed,
            tool_calls=len(all_tool_calls),
            usage_records=len(usage_tracker.records),
        )
        return AgentResult(
            answer=answer,
            tool_calls=all_tool_calls,
            messages=history.get_messages(),
            total_turns=total_turns,
            total_duration_ms=elapsed,
            usage=usage_tracker.totals(),
            usage_records=list(usage_tracker.records),
            trace=trace,
        )

    for turn in range(max_turns):
        logger.debug("agent turn %d/%d", turn + 1, max_turns)

        call_kwargs: Dict[str, Any] = {
            "usage_tracker": usage_tracker,
            "usage_turn": turn + 1,
        }
        # Last-turn salvage: if this is the final allowed turn and the agent
        # still hasn't emitted the Files:/Symbols:/Locations: contract, spend
        # it producing a formatted answer instead of one more (truncated)
        # tool call. This is contract preservation: a useful localization can
        # be lost if the run ends mid-search or in unstructured prose.
        is_last_turn = self._force_contract and turn == max_turns - 1
        last_assistant = ""
        for _m in reversed(history.get_messages()):
            if _m.get("role") == "assistant" and _m.get("content"):
                last_assistant = _m["content"]
                break
        if is_last_turn and not _has_localization_contract(last_assistant):
            history.add_message(
                {
                    "role": "user",
                    "content": (
                        "Stop exploring. Give your final answer NOW in exactly "
                        f"this format:\n{LOCALIZATION_SCHEMA}"
                    ),
                }
            )
            if tools:
                call_kwargs["tools"] = tools
                call_kwargs["tool_choice"] = "none"
        elif tools:
            call_kwargs["tools"] = tools
            # Force tool calls until the agent has actually READ a file.
            # The localization contract requires citing inspected code, so
            # keep tool_choice forced until a read happens; then drop to
            # "auto" so the model is free to commit. Bounded by
            # first_turn_only for the old single-turn behaviour.
            if self.first_turn_tool_choice and not has_read:
                if not (self._force_first_turn_only and turn > 0):
                    call_kwargs["tool_choice"] = self.first_turn_tool_choice

        trace.add(
            "llm_call",
            turn + 1,
            message_count=len(history.get_messages()),
            tool_count=len(call_kwargs.get("tools") or []),
            tool_choice=call_kwargs.get("tool_choice", "auto"),
        )
        response = self.llm._call_raw(history.get_messages(), **call_kwargs)
        choice = response.choices[0]
        assistant_msg = choice.message

        # Append assistant message to conversation (may evict oldest
        # non-system messages when a context-token budget is set).
        history.add_message(_message_to_dict(assistant_msg))

        # Check for tool calls
        tool_calls = getattr(assistant_msg, "tool_calls", None)
        trace.add(
            "llm_response",
            turn + 1,
            content_chars=len(getattr(assistant_msg, "content", None) or ""),
            tool_call_count=len(tool_calls or []),
        )
        if not tool_calls:
            # Terminal: LLM stopped calling tools. If it answered in prose
            # without the contract (it explored but didn't format), force one
            # schema turn so a genuine localization isn't lost to formatting.
            answer = getattr(assistant_msg, "content", None) or ""
            if (
                self._review_final_answer
                and all_tool_calls
                and not answer_reviewed
                and turn < max_turns - 1
            ):
                answer_reviewed = True
                trace.add(
                    "final_answer_review",
                    turn + 1,
                    status="requested",
                    draft_chars=len(answer),
                )
                history.add_message(
                    {
                        "role": "user",
                        "content": _GROUNDED_ANSWER_REVIEW_PROMPT,
                    }
                )
                continue
            answer_source = "assistant"
            if (
                self._force_contract
                and all_tool_calls
                and not _has_localization_contract(answer)
            ):
                trace.add(
                    "final_answer_forced",
                    turn + 1,
                    reason="missing_contract",
                    read_paths=list(dict.fromkeys(read_paths)),
                )
                answer = (
                    self._force_schema_answer(
                        history, usage_tracker, turn + 2, read_paths
                    )
                    or answer
                )
                answer_source = "forced_schema"
            return _finish_result(
                answer=answer,
                total_turns=turn + 1,
                answer_source=answer_source,
            )

        # Execute each tool call
        for tc in tool_calls:
            record = self._execute_tool_call(
                tc,
                allowed_tool_ids=active_tool_ids,
            )
            all_tool_calls.append(record)
            tool_content = _serialize_result(
                record.result if record.error is None else record.error
            )
            trace.add(
                "tool_call",
                turn + 1,
                **_trace_tool_call_data(record, tool_content),
            )
            # A successful read satisfies the "read before answering"
            # contract; once it happens, stop forcing tool calls.
            successful_read = (
                record.skill_id == "read"
                and record.error is None
                and not tool_content.lstrip().startswith("Error")
            )
            if record.skill_id == "read":
                trace.add(
                    "read",
                    turn + 1,
                    tool_call_id=record.tool_call_id,
                    path=str((record.arguments or {}).get("file_path") or ""),
                    offset=(record.arguments or {}).get("offset"),
                    limit=(record.arguments or {}).get("limit"),
                    status="ok" if successful_read else "error",
                )
            trace.add_context(
                record.skill_id,
                _context_state(record, tool_content, successful_read),
                turn + 1,
                **_context_ledger_data(record, tool_content),
            )
            if successful_read:
                has_read = True
                _rp = (record.arguments or {}).get("file_path")
                if _rp:
                    read_paths.append(str(_rp))
                read_outputs.append(
                    {"path": str(_rp or "(unknown)"), "content": tool_content}
                )

            # Append tool response message
            history.add_message(
                {
                    "role": "tool",
                    "tool_call_id": tc.id,
                    "content": tool_content,
                }
            )

        # eager_compact: once the eager phase has read a candidate and judged
        # the direction, COLLAPSE the bulky preload dump + the whole
        # exploration trace to a small distilled seed
        # ([system, clean_query + judged direction]) and let the
        # "correct-path" continuation finish from there. We do NOT carry the
        # preload candidate snippets or the exploration history into every
        # later turn (re-sending them is the dominant prompt-token cost).
        # Once only — this is the explore→commit boundary, not a rolling trim.
        if self._compact_after_read and has_read and not compacted:
            compacted = True
            if self._compact_history(
                history,
                read_paths,
                self._compact_keep_reads,
                read_outputs=read_outputs,
            ):
                trace.add(
                    "context_compacted",
                    turn + 1,
                    read_paths=list(dict.fromkeys(read_paths)),
                    keep_reads=self._compact_keep_reads,
                )
                trace.add_context(
                    "runtime.compaction",
                    "summarized",
                    turn + 1,
                    summary=(
                        "Compacted conversation after reads: "
                        + (", ".join(dict.fromkeys(read_paths)) or "(none)")
                    ),
                    metadata={
                        "read_paths": list(dict.fromkeys(read_paths)),
                        "keep_reads": self._compact_keep_reads,
                    },
                )
                logger.debug("eager_compact: collapsed to distilled direction seed")

    # Max turns exhausted. Prefer the last textual assistant message.
    trace.add(
        "max_turns_exhausted",
        max_turns,
        tool_calls=len(all_tool_calls),
        has_read=has_read,
    )
    last_content = ""
    for msg in reversed(history.get_messages()):
        if msg.get("role") == "assistant" and msg.get("content"):
            last_content = msg["content"]
            break

    # Budget exhausted. A capped agent usually left mid-exploration chatter
    # ("Let me check…"), so force a schema-conforming final answer unless it
    # already emitted the contract.
    answer_source = "max_turns"
    if (
        self._force_contract
        and all_tool_calls
        and not _has_localization_contract(last_content)
    ):
        trace.add(
            "final_answer_forced",
            max_turns,
            reason="max_turns_exhausted",
            read_paths=list(dict.fromkeys(read_paths)),
        )
        last_content = (
            self._force_schema_answer(
                history, usage_tracker, max_turns + 1, read_paths
            )
            or last_content
        )
        answer_source = "forced_schema"
    elif self._final_answer:
        trace.add(
            "final_answer_forced",
            max_turns,
            reason="max_turns_exhausted",
        )
        last_content = self._force_final_answer(
            history, usage_tracker, max_turns + 1
        )
        if not last_content.strip():
            raise RuntimeError(
                f"agent exhausted max_turns={max_turns} without producing "
                "a final answer"
            )
        answer_source = "forced_answer"

    return _finish_result(
        answer=last_content,
        total_turns=max_turns,
        answer_source=answer_source,
    )

CodeNibAgentOptions dataclass

CodeNibAgentOptions(
    repo_path: str | None = None,
    contexts: dict[str, Any] | None = None,
    manifest: "RepoManifest" | str | Path | None = None,
    languages: Sequence[str] = ("python",),
    primary_language: str | None = None,
    repo_size: int | None = None,
    index_cache_dir: str | None = None,
    embedding_model: str = DEFAULT_EMBEDDING_MODEL,
    embedding_dimension: int = DEFAULT_EMBEDDING_DIMENSION,
    default_top_k: int = 10,
    default_level: str = "l2",
    rebuild_indexes: bool = False,
    skills_dir: str | None = None,
    native_index_authorization: "NativeIndexAuthorization" | None = None,
    native_index_authorization_resolver: (
        Callable[["IndexEntry"], "NativeIndexAuthorization" | None] | None
    ) = None,
    allowed_skills: list[str] | None = None,
    excluded_skills: list[str] | None = None,
    compile_table: CompileTableInput | None = None,
    skill_params: dict[str, dict[str, Any]] | None = None,
    llm: LiteLLMChat | None = None,
    model: str | None = None,
    temperature: float = 0.0,
    max_tokens: int = 512,
    system_prompt: str | None = None,
    max_turns: int = 10,
    max_context_tokens: int | None = None,
    enable_lsp_route_context: bool = False,
    lsp_route_seed_limit: int = 8,
    lsp_route_seed_policy: str = "all",
    lsp_route_query_fallback: bool = False,
    lsp_route_top_k: int = 12,
    lsp_route_include_neighbors: bool = True,
    retry: RetryConfig | None = None,
    sandbox: "SandboxSession" | None = None,
    session_extras: dict[str, Any] = dict(),
)

Configuration for a single query() invocation.

Exactly one of repo_path, contexts, or manifest must be set — these are the three mutually-exclusive ways to tell query() where its indexes come from:

• repo_path → query() calls :func:~codenib.compiler.build_skill_contexts itself and caches indexes under index_cache_dir (or <repo>/.codenib_cache). The "build once at first query" path.
• contexts → caller pre-built the contexts dict (advanced; see :func:~codenib.compiler.build_skill_contexts / :func:~codenib.compiler.load_contexts_from_manifest for what shape to pass).
• manifest → caller compiled indexes ahead of time via :func:~codenib.agent.compile_repo (or :class:~codenib.compiler.IndexCompiler directly) and passes either a loaded :class:~codenib.compiler.RepoManifest or a path string to repo_manifest.json. The AoT (ahead-of-time) path: query() loads the artifacts named in the manifest and threads the manifest itself into AgentRunner so :class:~codenib.agent.resource_guard.ResourceGuard can run freshness checks. No inline build happens.

Skill selection forms a three-layer funnel::

registry  ⊇  allowed_skills  ⊇  compile_table[scenario]

compile_table narrows allowed_skills per query but never broadens it (see :func:codenib.agent.compile.agent_compile).

compile_table also operates at the index-build stage (for repo_path mode only): when set, only indexes for skills it ever names are compiled. Formally::

index_skills = allowed_skills  ∩  union(compile_table.values())

A vector index isn't built if every scenario in the table maps to bm25-only, even when embedding_search is in allowed_skills — CAR couldn't route to it at runtime anyway. (In manifest mode this rule is moot — the manifest dictates what exists.)

Native vector parsers are fail-closed. Pass either an already-bound native_index_authorization or a native_index_authorization_resolver that receives the exact manifest IndexEntry about to be loaded. The two options are mutually exclusive; neither can be derived from manifest fields by the agent runtime.

AgentRunTrace dataclass

AgentRunTrace(
    events: list[AgentTraceEvent] = list(),
    context: ContextLedger = ContextLedger(),
    start_monotonic: float = monotonic(),
)

Durable event log for one AgentRunner.run() invocation.

AgentTraceEvent dataclass

AgentTraceEvent(
    kind: str,
    turn: int,
    data: dict[str, Any] = dict(),
    timestamp_ms: float | None = None,
)

A replay-oriented event emitted by an agent run.

The trace is intentionally descriptive, not prescriptive: events explain what happened during runtime without encoding benchmark scoring or promotion policy.

ContextLedger

ContextLedger(entries: Sequence[ContextLedgerEntry] | None = None)

Append-only collection of context ledger entries for one agent run.

Source code in codenib/agent/runtime/context.py

def __init__(self, entries: Sequence[ContextLedgerEntry] | None = None) -> None:
    self._entries: List[ContextLedgerEntry] = list(entries or [])

ContextLedgerEntry dataclass

ContextLedgerEntry(
    source: str,
    state: str,
    turn: int,
    summary: str = "",
    path: str | None = None,
    tool_call_id: str | None = None,
    entry_id: str | None = None,
    provenance: dict[str, Any] = dict(),
    freshness: str | None = None,
    token_estimate: int | None = None,
    cost_estimate: float | None = None,
    consumed_by: list[str] = list(),
    consumed_turn: int | None = None,
    metadata: dict[str, Any] = dict(),
)

Bounded summary of context produced, retained, or consumed by a run.

Methods:

Name	Description
mark_consumed	Record that this context entry was consumed by a later runtime step.
mark_expired	Record that this context entry left the active working set.

mark_consumed

mark_consumed(*, turn: int, consumer: str) -> None

Record that this context entry was consumed by a later runtime step.

Source code in codenib/agent/runtime/context.py

def mark_consumed(self, *, turn: int, consumer: str) -> None:
    """Record that this context entry was consumed by a later runtime step."""

    if turn < 0:
        raise ValueError("turn must be non-negative")
    self.state = "consumed"
    self.consumed_turn = turn
    if consumer and consumer not in self.consumed_by:
        self.consumed_by.append(consumer)

mark_expired

mark_expired(*, turn: int, reason: str | None = None) -> None

Record that this context entry left the active working set.

Source code in codenib/agent/runtime/context.py

def mark_expired(self, *, turn: int, reason: str | None = None) -> None:
    """Record that this context entry left the active working set."""

    if turn < 0:
        raise ValueError("turn must be non-negative")
    self.state = "expired"
    self.consumed_turn = turn
    if reason:
        self.metadata["expiration_reason"] = reason

RepositoryContextExplorer

RepositoryContextExplorer(
    context: Any,
    *,
    policy: str = "auto",
    budget: BudgetInput = "balanced",
    level: str = "l2",
    _owns_context: bool = False
)

Plan and execute repository search over manifest-backed CodeNib views.

auto plans against manifest-advertised capabilities and loads only the selected query path. Explicit policies provide stable ablation arms. The class never materializes indexes; callers build views separately.

Methods:

Name	Description
from_manifest	Bind to a manifest without eagerly loading retrieval views.
from_repository	Resolve the manifest bound to one checkout and create an explorer.
close	Release resources loaded by a manifest-constructed explorer.
explore	Return bounded, source-validated evidence for one context request.
prepare	Plan a query and load only the views selected for that plan.

Source code in codenib/agent/runtime/explorer.py

def __init__(
    self,
    context: Any,
    *,
    policy: str = "auto",
    budget: BudgetInput = "balanced",
    level: str = "l2",
    _owns_context: bool = False,
) -> None:
    manifest = getattr(context, "manifest", None)
    if manifest is None:
        raise RepositoryExplorerCapabilityError("context has no manifest")
    raw_root = str(getattr(manifest, "repo_path", "") or "").strip()
    if not raw_root:
        raise RepositoryExplorerCapabilityError("manifest has no repository path")
    repo_root = Path(raw_root).expanduser().resolve()
    if not repo_root.is_dir():
        raise RepositoryExplorerCapabilityError(
            f"manifest repository is not available: {repo_root}"
        )
    normalized_level = str(level or "l2").strip().lower()
    if normalized_level not in {"l0", "l2"}:
        raise ValueError("repository explorer level must be 'l0' or 'l2'")

    self.context = context
    self.repo_root = repo_root
    self.policy = normalize_repository_explorer_policy(policy)
    self.planner = RetrievalPlanner()
    self.budget = self.planner.normalize_budget(budget)
    self.level = normalized_level
    self._owns_context = _owns_context
    self._source_lines: dict[str, tuple[str, ...]] = {}
    self._lock = RLock()
    self._validate_advertised_policy()

from_manifest classmethod

from_manifest(
    manifest_path: str | Path,
    *,
    policy: str = "auto",
    budget: BudgetInput = "balanced",
    level: str = "l2"
) -> RepositoryContextExplorer

Bind to a manifest without eagerly loading retrieval views.

Source code in codenib/agent/runtime/explorer.py

@classmethod
def from_manifest(
    cls,
    manifest_path: str | Path,
    *,
    policy: str = "auto",
    budget: BudgetInput = "balanced",
    level: str = "l2",
) -> RepositoryContextExplorer:
    """Bind to a manifest without eagerly loading retrieval views."""

    from ...mcp.context import ServerContext

    context = ServerContext.load(manifest_path, views=())
    return cls(
        context,
        policy=policy,
        budget=budget,
        level=level,
        _owns_context=True,
    )

from_repository classmethod

from_repository(
    repo_path: str | Path,
    *,
    manifest_path: str | Path | None = None,
    policy: str = "auto",
    budget: BudgetInput = "balanced",
    level: str = "l2"
) -> RepositoryContextExplorer

Resolve the manifest bound to one checkout and create an explorer.

Source code in codenib/agent/runtime/explorer.py

@classmethod
def from_repository(
    cls,
    repo_path: str | Path,
    *,
    manifest_path: str | Path | None = None,
    policy: str = "auto",
    budget: BudgetInput = "balanced",
    level: str = "l2",
) -> RepositoryContextExplorer:
    """Resolve the manifest bound to one checkout and create an explorer."""

    from ...clients._manifest import select_checkout_manifest

    selected = select_checkout_manifest(
        repo_path,
        integration_name="repository explorer",
        manifest_path=manifest_path,
    )
    return cls.from_manifest(
        selected,
        policy=policy,
        budget=budget,
        level=level,
    )

close

close() -> None

Release resources loaded by a manifest-constructed explorer.

Source code in codenib/agent/runtime/explorer.py

def close(self) -> None:
    """Release resources loaded by a manifest-constructed explorer."""

    if self._owns_context:
        close = getattr(self.context, "close", None)
        if callable(close):
            close()

explore

explore(
    query: str,
    *,
    top_k: int = 10,
    include_content: bool = False,
    budget: BudgetInput | None = None
) -> RepositoryExploreResult

Return bounded, source-validated evidence for one context request.

Source code in codenib/agent/runtime/explorer.py

def explore(
    self,
    query: str,
    *,
    top_k: int = 10,
    include_content: bool = False,
    budget: BudgetInput | None = None,
) -> RepositoryExploreResult:
    """Return bounded, source-validated evidence for one context request."""

    _validate_top_k(top_k)
    text = str(query or "").strip()
    if top_k == 0 or not text:
        return RepositoryExploreResult(evidence=(), trace=None)

    with self._lock:
        prepared = self.prepare(text, budget=budget)
        active_budget = prepared.budget
        signals = prepared.signals
        plan = prepared.plan
        capabilities = prepared.capabilities
        vector = getattr(self.context, "vector", None)
        reuse_query = getattr(vector, "reuse_query_embedding", None)
        scope = reuse_query() if callable(reuse_query) else nullcontext()
        with scope:
            candidates, stage_counts = self._execute_plan(
                text,
                plan,
                top_k=top_k,
            )
        evidence = self._source_evidence(
            candidates,
            top_k=top_k,
            include_content=include_content,
        )
        trace = RepositoryExploreTrace(
            policy=self.policy,
            plan=plan,
            signals=signals,
            budget=active_budget,
            capabilities=capabilities,
            advertised_views=tuple(sorted(self._advertised_views())),
            loaded_views=tuple(sorted(self._loaded_views())),
            stage_candidate_counts=tuple(stage_counts),
            retrieved_candidates=len(candidates),
            returned_evidence=len(evidence),
        )
        return RepositoryExploreResult(evidence=tuple(evidence), trace=trace)

prepare

prepare(
    query: str, *, budget: BudgetInput | None = None
) -> RepositoryExplorePreparation

Plan a query and load only the views selected for that plan.

Source code in codenib/agent/runtime/explorer.py

def prepare(
    self,
    query: str,
    *,
    budget: BudgetInput | None = None,
) -> RepositoryExplorePreparation:
    """Plan a query and load only the views selected for that plan."""

    text = str(query or "").strip()
    if not text:
        raise ValueError("repository explorer query must not be empty")
    with self._lock:
        active_budget = self.planner.normalize_budget(budget or self.budget)
        plan, capabilities = self._select_and_load_plan(text, active_budget)
        return RepositoryExplorePreparation(
            query=text,
            plan=plan,
            signals=self.planner.classify_query(text),
            budget=active_budget,
            capabilities=capabilities,
        )

RepositoryEvidence dataclass

RepositoryEvidence(
    path: str,
    start_line: int,
    end_line: int,
    score: float,
    node_name: str = "",
    node_type: str = "",
    node_id: str | None = None,
    content: str | None = None,
)

One ranked source span in CodeNib's 0-based coordinate system.

RepositoryExplorePreparation dataclass

RepositoryExplorePreparation(
    query: str,
    plan: RetrievalPathPlan,
    signals: QuerySignals,
    budget: RetrievalBudget,
    capabilities: RetrievalCapabilities,
)

A deterministic query plan with all selected views loaded.

RepositoryExplorerCapabilityError

Bases: RuntimeError

The selected exploration policy cannot run over the manifest.

RepositoryExploreResult dataclass

RepositoryExploreResult(
    evidence: tuple[RepositoryEvidence, ...], trace: RepositoryExploreTrace | None
)

Ranked repository evidence and the plan that produced it.

RepositoryExploreTrace dataclass

RepositoryExploreTrace(
    policy: str,
    plan: RetrievalPathPlan,
    signals: QuerySignals,
    budget: RetrievalBudget,
    capabilities: RetrievalCapabilities,
    advertised_views: tuple[str, ...],
    loaded_views: tuple[str, ...],
    stage_candidate_counts: tuple[tuple[str, int], ...],
    retrieved_candidates: int,
    returned_evidence: int,
)

Planner and execution provenance for one exploration request.

extract_keywords_from_statement

extract_keywords_from_statement(
    problem_statement: str, llm: LiteLLMChat
) -> KeywordExtraction

Extract keywords from a problem statement.

Parameters:

Name	Type	Description	Default
problem_statement	str	The problem statement to extract keywords from.	required
llm	LiteLLMChat	A configured LiteLLMChat instance.	required

Returns:

Name	Type	Description
KeywordExtraction	KeywordExtraction	Structured output with extracted keywords

Source code in codenib/agent/extract_agent.py

def extract_keywords_from_statement(
    problem_statement: str,
    llm: LiteLLMChat,
) -> KeywordExtraction:
    """
    Extract keywords from a problem statement.

    Args:
        problem_statement: The problem statement to extract keywords from.
        llm: A configured LiteLLMChat instance.

    Returns:
        KeywordExtraction: Structured output with extracted keywords
    """
    extractor = KeywordExtractor(llm=llm)
    return extractor.extract_keywords(problem_statement)

agent_working_directory

agent_working_directory(cwd: str | Path | None) -> Iterator[None]

Temporarily run agent default tools relative to cwd.

The built-in read/grep/glob tools intentionally accept repo-relative paths. Harnesses that run against many repos should scope the process cwd around each runner invocation and always restore it afterwards.

Source code in codenib/agent/harness.py

@contextmanager
def agent_working_directory(cwd: Optional[Union[str, Path]]) -> Iterator[None]:
    """Temporarily run agent default tools relative to *cwd*.

    The built-in read/grep/glob tools intentionally accept repo-relative paths.
    Harnesses that run against many repos should scope the process cwd around
    each runner invocation and always restore it afterwards.
    """
    if cwd is None:
        yield
        return

    previous = os.getcwd()
    os.chdir(cwd)
    try:
        yield
    finally:
        os.chdir(previous)

run_agent_in_directory

run_agent_in_directory(runner: Any, prompt: str, cwd: str | Path | None) -> Any

Run an AgentRunner-like object with repo-relative default tools.

Source code in codenib/agent/harness.py

def run_agent_in_directory(
    runner: Any,
    prompt: str,
    cwd: Optional[Union[str, Path]],
) -> Any:
    """Run an ``AgentRunner``-like object with repo-relative default tools."""
    with agent_working_directory(cwd):
        return runner.run(prompt)

count_message_tokens

count_message_tokens(
    messages: list[dict[str, Any]], *, model: str | None = None
) -> int

Best-effort token count for a list of chat messages.

Uses litellm.token_counter (which understands the model's real tokenizer and per-message role overhead) when a model is supplied and litellm is importable. Falls back to a chars / 4 heuristic otherwise, so this never raises and stays usable in offline tests.

Source code in codenib/agent/history.py

def count_message_tokens(
    messages: List[Dict[str, Any]],
    *,
    model: Optional[str] = None,
) -> int:
    """Best-effort token count for a list of chat messages.

    Uses ``litellm.token_counter`` (which understands the model's real
    tokenizer and per-message role overhead) when a ``model`` is supplied
    and litellm is importable. Falls back to a ``chars / 4`` heuristic
    otherwise, so this never raises and stays usable in offline tests.
    """
    if not messages:
        return 0
    if model:
        try:
            import litellm

            return int(litellm.token_counter(model=model, messages=messages))
        except Exception as exc:  # offline / unknown model / litellm bug
            logger.debug("token_counter unavailable (%s); using char heuristic", exc)

    chars = sum(len(_message_text(m)) for m in messages)
    # +4 per message approximates the role / formatting overhead litellm adds.
    return chars // _CHARS_PER_TOKEN + 4 * len(messages)

lsp_result_metadata

lsp_result_metadata(result: Any) -> dict[str, Any] | None

Return provider metadata from a list-compatible LSP result.

Source code in codenib/agent/lsp_provider.py

def lsp_result_metadata(result: Any) -> Optional[dict[str, Any]]:
    """Return provider metadata from a list-compatible LSP result."""

    if hasattr(result, "provider_metadata_dict"):
        raw = result.provider_metadata_dict()
    else:
        raw = getattr(result, "lsp_provider_metadata", None)
        if hasattr(raw, "to_dict"):
            raw = raw.to_dict()
    return dict(raw) if isinstance(raw, Mapping) else None

rerank_nodes_with_query

rerank_nodes_with_query(
    query: str,
    nodes: list[NodeInfo],
    llm: LiteLLMChat,
    top_k: int | None = None,
    window_size: int | None = None,
    window_step: int | None = None,
    include_content: bool = False,
) -> list[QueriedNode]

Convenience function to rerank nodes with a query.

Parameters:

Name	Type	Description	Default
query	str	The query to rank nodes against.	required
nodes	list[NodeInfo]	List of nodes with content to rank.	required
llm	LiteLLMChat	A configured LiteLLMChat instance.	required
top_k	int | None	Maximum number of results to return (None for all).	None
window_size	int | None	Optional sliding window size for reranking.	None
window_step	int | None	Optional stride between sliding windows.	None
include_content	bool	Whether to include node content in the result objects.	False

Returns:

Type	Description
list[QueriedNode]	List[QueriedNode]: Ranked nodes with relevance scores

Source code in codenib/agent/rerank_agent.py

def rerank_nodes_with_query(
    query: str,
    nodes: List[NodeInfo],
    llm: LiteLLMChat,
    top_k: Optional[int] = None,
    window_size: Optional[int] = None,
    window_step: Optional[int] = None,
    include_content: bool = False,
) -> List[QueriedNode]:
    """
    Convenience function to rerank nodes with a query.

    Args:
        query: The query to rank nodes against.
        nodes: List of nodes with content to rank.
        llm: A configured LiteLLMChat instance.
        top_k: Maximum number of results to return (None for all).
        window_size: Optional sliding window size for reranking.
        window_step: Optional stride between sliding windows.
        include_content: Whether to include node content in the result objects.

    Returns:
        List[QueriedNode]: Ranked nodes with relevance scores
    """
    agent = RerankAgent(llm=llm)
    return agent.rerank_nodes(
        query,
        nodes,
        top_k=top_k,
        window_size=window_size,
        window_step=window_step,
        include_content=include_content,
    )

build_lsp_route_context

build_lsp_route_context(
    executor: Any,
    query: str,
    *,
    explicit_seeds: Any = None,
    seed_limit: int = 8,
    seed_policy: str | None = "all",
    query_fallback: bool = False,
    top_k: int = 12,
    include_neighbors: bool = True
) -> LSPRouteContext

Run an lsp_route executor and render startup context.

Source code in codenib/agent/route_context.py

def build_lsp_route_context(
    executor: Any,
    query: str,
    *,
    explicit_seeds: Any = None,
    seed_limit: int = 8,
    seed_policy: str | None = "all",
    query_fallback: bool = False,
    top_k: int = 12,
    include_neighbors: bool = True,
) -> LSPRouteContext:
    """Run an ``lsp_route`` executor and render startup context."""

    seeds = extract_lsp_symbol_seeds(
        query,
        explicit=explicit_seeds,
        limit=seed_limit,
    )
    seeds = filter_lsp_symbol_seeds(seeds, seed_policy=seed_policy)
    if not seeds and not query_fallback:
        return LSPRouteContext(seeds=(), nodes=(), text="")

    route_args = canonical_lsp_route_args(
        symbols=seeds,
        query=query,
        top_k=top_k,
        include_neighbors=include_neighbors,
    )
    nodes = tuple(
        executor(
            **route_args,
        )
        or ()
    )
    return LSPRouteContext(
        seeds=tuple(seeds),
        nodes=nodes,
        text=render_lsp_route_context(seeds, nodes),
        arguments=route_args,
        route_fingerprint=fingerprint_lsp_route_nodes(nodes),
    )

canonical_lsp_route_args

canonical_lsp_route_args(
    *,
    symbols: Iterable[Any],
    query: Any = None,
    top_k: Any = 12,
    include_neighbors: Any = True
) -> dict[str, Any]

Return the canonical argument shape for a static LSP route call.

Source code in codenib/agent/route_context.py

def canonical_lsp_route_args(
    *,
    symbols: Iterable[Any],
    query: Any = None,
    top_k: Any = 12,
    include_neighbors: Any = True,
) -> dict[str, Any]:
    """Return the canonical argument shape for a static LSP route call."""

    return {
        "symbols": [str(symbol) for symbol in symbols or []],
        "query": str(query) if query is not None else None,
        "top_k": int(top_k or 12),
        "include_neighbors": bool(include_neighbors),
    }

extract_lsp_symbol_seeds

extract_lsp_symbol_seeds(
    *texts: Any, explicit: Any = None, limit: int = 8
) -> list[str]

Extract explicit symbol-like seeds from task text.

The extractor is intentionally conservative. It keeps user-supplied explicit seeds first, then adds backtick-delimited code names and code-like tokens from the task text. It does not inspect repository contents or benchmark labels.

Source code in codenib/agent/route_context.py

def extract_lsp_symbol_seeds(
    *texts: Any,
    explicit: Any = None,
    limit: int = 8,
) -> List[str]:
    """Extract explicit symbol-like seeds from task text.

    The extractor is intentionally conservative. It keeps user-supplied explicit
    seeds first, then adds backtick-delimited code names and code-like tokens
    from the task text. It does not inspect repository contents or benchmark
    labels.
    """

    seeds: List[str] = []
    _add_seed_values(seeds, explicit)

    text = "\n".join(_flatten_text_values(texts))
    for match in _BACKTICK.findall(text):
        _add_seed_values(seeds, match)
    for token in _TOKEN.findall(text):
        if token.upper() in _IDENT_STOP:
            continue
        if _CODEISH.search(token) or (token.isupper() and len(token) >= 3):
            _add_seed_values(seeds, token)

    seen: set[str] = set()
    out: List[str] = []
    max_items = max(1, int(limit or 8))
    for seed in seeds:
        key = seed.lower()
        if key in seen:
            continue
        seen.add(key)
        out.append(seed)
        if len(out) >= max_items:
            break
    return out

filter_lsp_symbol_seeds

filter_lsp_symbol_seeds(
    seeds: Iterable[str], *, seed_policy: str | None = "all"
) -> list[str]

Apply a route startup seed policy while preserving seed order.

Source code in codenib/agent/route_context.py

def filter_lsp_symbol_seeds(
    seeds: Iterable[str],
    *,
    seed_policy: str | None = "all",
) -> List[str]:
    """Apply a route startup seed policy while preserving seed order."""

    policy = normalize_lsp_route_seed_policy(seed_policy)
    out = list(seeds)
    if policy == "all":
        return out
    return [seed for seed in out if is_specific_lsp_symbol_seed(seed)]

fingerprint_lsp_route_nodes

fingerprint_lsp_route_nodes(nodes: Sequence[Any]) -> str

Return a stable fingerprint for an ordered route-node result.

Source code in codenib/agent/route_context.py

def fingerprint_lsp_route_nodes(nodes: Sequence[Any]) -> str:
    """Return a stable fingerprint for an ordered route-node result."""

    payload = summarize_lsp_route_nodes(nodes, max_nodes=len(nodes))
    encoded = json.dumps(
        payload,
        sort_keys=True,
        separators=(",", ":"),
        ensure_ascii=True,
    ).encode("utf-8")
    return hashlib.sha256(encoded).hexdigest()

is_specific_lsp_symbol_seed

is_specific_lsp_symbol_seed(seed: str) -> bool

Return whether seed is specific enough for gated startup routing.

Source code in codenib/agent/route_context.py

def is_specific_lsp_symbol_seed(seed: str) -> bool:
    """Return whether *seed* is specific enough for gated startup routing."""

    text = (seed or "").strip().strip("`'\"")
    if not text:
        return False
    if not _is_symbol_seed_shape(text):
        return False
    check = text[:-2] if text.endswith("()") else text
    if "/" in check or "::" in check or "#" in check:
        return True
    if "." in check:
        if _looks_like_domain_fragment(check):
            return False
        return any(_is_specific_name_part(part) for part in check.split("."))
    if "_" in check:
        return True
    if _CAMEL_SEED.search(text):
        return True
    if text.isupper() and len(text) >= 3:
        return True
    if (
        any(ch.isalpha() for ch in text)
        and any(ch.isdigit() for ch in text)
        and not text.islower()
    ):
        return True
    return False

normalize_lsp_route_seed_policy

normalize_lsp_route_seed_policy(seed_policy: str | None = 'all') -> str

Normalize and validate the startup route seed policy.

Source code in codenib/agent/route_context.py

def normalize_lsp_route_seed_policy(seed_policy: str | None = "all") -> str:
    """Normalize and validate the startup route seed policy."""

    policy = (seed_policy or "all").strip().lower()
    if policy not in _SEED_POLICIES:
        raise ValueError(
            f"Unsupported lsp_route seed_policy {seed_policy!r}; "
            f"supported values are {sorted(_SEED_POLICIES)}"
        )
    return policy

render_lsp_route_context

render_lsp_route_context(
    seeds: Sequence[str], nodes: Sequence[Any], *, max_nodes: int | None = None
) -> str

Render route nodes as compact, unverified prompt context.

Source code in codenib/agent/route_context.py

def render_lsp_route_context(
    seeds: Sequence[str],
    nodes: Sequence[Any],
    *,
    max_nodes: int | None = None,
) -> str:
    """Render route nodes as compact, unverified prompt context."""

    selected = list(nodes)
    if max_nodes is not None:
        selected = selected[: max(0, int(max_nodes))]
    if not selected:
        return ""

    seed_line = "Seeds: " + ", ".join(seeds) if seeds else "Seed source: query text"
    lines = [
        "# Static LSP route hints (unverified)",
        "These graph anchors come from explicit symbol-like names in the task. "
        "If no symbol seed was available, query text was used to find candidate "
        "graph anchors. Use them to choose the first one or two files to inspect; "
        "they are not a checklist. Do not cite them until you read the file.",
        seed_line,
        "",
    ]
    for index, node in enumerate(selected, 1):
        data = _node_dict(node)
        location = _location(data)
        symbol = str(data.get("node_name") or data.get("node_id") or "(unknown)")
        node_type = str(data.get("type") or "symbol")
        relation = str(data.get("content") or "").strip()
        suffix = f" - {relation}" if relation else ""
        lines.append(f"{index}. {location} {symbol} [{node_type}]{suffix}")
    return "\n".join(lines)

compile_repo

compile_repo(
    repo_path: str,
    *,
    index_types: Sequence[str] = ("bm25",),
    languages: Sequence[str] = ("python",),
    cache_dir: str | None = None,
    embedding_model: str = DEFAULT_EMBEDDING_MODEL,
    embedding_dimension: int = DEFAULT_EMBEDDING_DIMENSION,
    source_selection: RepositorySourceSelection | None = None
) -> "RepoManifest"

Compile indexes for repo_path ahead of time and return the manifest.

Thin convenience wrapper over :class:~codenib.compiler.IndexCompiler that registers the default index builders for the requested languages and writes <cache_dir>/repo_manifest.json.

Pair with :func:query to run the agent against the result without re-indexing on every call::

manifest = compile_repo(
    "/path/to/repo",
    index_types=("bm25", "vector"),
    languages=("python",),
)
result = query(
    "where is auth wired up?",
    options=CodeNibAgentOptions(
        manifest=manifest,
        allowed_skills=["bm25_search", "embedding_search"],
    ),
)

For advanced cases — custom builder registries, partial rebuilds — use :class:~codenib.compiler.IndexCompiler directly.

Source code in codenib/agent/runner.py

def compile_repo(
    repo_path: str,
    *,
    index_types: Sequence[str] = ("bm25",),
    languages: Sequence[str] = ("python",),
    cache_dir: Optional[str] = None,
    embedding_model: str = DEFAULT_EMBEDDING_MODEL,
    embedding_dimension: int = DEFAULT_EMBEDDING_DIMENSION,
    source_selection: RepositorySourceSelection | None = None,
) -> "RepoManifest":
    """Compile indexes for *repo_path* ahead of time and return the manifest.

    Thin convenience wrapper over :class:`~codenib.compiler.IndexCompiler`
    that registers the default index builders for the requested
    ``languages`` and writes ``<cache_dir>/repo_manifest.json``.

    Pair with :func:`query` to run the agent against the result without
    re-indexing on every call::

        manifest = compile_repo(
            "/path/to/repo",
            index_types=("bm25", "vector"),
            languages=("python",),
        )
        result = query(
            "where is auth wired up?",
            options=CodeNibAgentOptions(
                manifest=manifest,
                allowed_skills=["bm25_search", "embedding_search"],
            ),
        )

    For advanced cases — custom builder registries, partial rebuilds —
    use :class:`~codenib.compiler.IndexCompiler` directly.
    """
    from ..compiler.index_builders import (
        IndexBuilderRegistry,
        register_default_builders,
    )
    from ..compiler.index_compiler import IndexCompiler, IndexCompilerConfig
    from ..compiler.manifest_source import resolve_compiler_source_selection

    if cache_dir is None:
        cache_dir = str(repo_index_dir(repo_path))
    selected = resolve_compiler_source_selection(cache_dir, source_selection)

    builder_registry = IndexBuilderRegistry()
    register_default_builders(
        builder_registry,
        languages=list(languages),
        embedding_model=embedding_model,
        embedding_dimension=embedding_dimension,
        source_selection=selected,
    )

    cfg = IndexCompilerConfig(
        cache_dir_name=Path(cache_dir).name,
        index_types=list(index_types),
        languages=list(languages),
        source_selection=selected,
    )
    compiler = IndexCompiler(builder_registry, cfg)
    return compiler.compile_repo(
        repo_path,
        index_types=list(index_types),
        cache_dir=cache_dir,
    )

has_localization_contract

has_localization_contract(answer: str) -> bool

Return true when an answer carries the localization output contract.

Source code in codenib/agent/runner.py

def has_localization_contract(answer: str) -> bool:
    """Return true when an answer carries the localization output contract."""
    return all(
        pattern.search(answer or "")
        for pattern in (
            _HAS_FILES_CONTRACT,
            _HAS_SYMBOLS_CONTRACT,
            _HAS_LOCATIONS_CONTRACT,
        )
    )

query

query(prompt: str, *, options: CodeNibAgentOptions | None = None) -> AgentResult

Run one agent turn over a repo and return the result.

See :class:CodeNibAgentOptions for the full options surface, including the three mutually-exclusive index-source modes (repo_path, contexts, manifest).

Raises:

Type	Description
ValueError	if not exactly one of options.repo_path, options.contexts, or options.manifest is set; or if a manifest is supplied but is missing an index required by options.allowed_skills.

Source code in codenib/agent/runner.py

def query(
    prompt: str,
    *,
    options: Optional[CodeNibAgentOptions] = None,
) -> AgentResult:
    """Run one agent turn over a repo and return the result.

    See :class:`CodeNibAgentOptions` for the full options surface,
    including the three mutually-exclusive index-source modes
    (``repo_path``, ``contexts``, ``manifest``).

    Raises:
        ValueError: if not exactly one of ``options.repo_path``,
            ``options.contexts``, or ``options.manifest`` is set; or if a
            manifest is supplied but is missing an index required by
            ``options.allowed_skills``.
    """
    opts = options or CodeNibAgentOptions()

    _check_exactly_one_mode(opts)
    if opts.contexts is not None and (
        opts.native_index_authorization is not None
        or opts.native_index_authorization_resolver is not None
    ):
        raise ValueError(
            "native-index authorization options are not accepted in contexts "
            "mode because its pre-built objects are not loaded by query()"
        )
    if (
        opts.native_index_authorization is not None
        and opts.native_index_authorization_resolver is not None
    ):
        raise ValueError(
            "provide either native_index_authorization or "
            "native_index_authorization_resolver, not both"
        )
    if opts.sandbox is not None and opts.manifest is None:
        raise ValueError(
            "query() sandbox execution requires a trusted, revision-matched "
            "manifest; repo_path can run repository-selected toolchains on the "
            "host and contexts have no source provenance"
        )

    if opts.llm is None and opts.model is None:
        model_for_llm: Optional[str] = _DEFAULT_MODEL
    else:
        model_for_llm = opts.model

    # --- 1. Resolve the compile_table (path → dict if needed) ---
    table = _load_compile_table_if_path(opts.compile_table)

    # --- 1b. Resolve the manifest (path → loaded RepoManifest if needed).
    # We do this early so the resolved manifest can be threaded into both
    # ``load_contexts_from_manifest`` (loading) and ``AgentRunner`` (for
    # ``ResourceGuard`` freshness checks).
    manifest = _load_manifest_if_path(opts.manifest)
    if opts.sandbox is not None and manifest is not None:
        manifest_revision = str(manifest.commit or "")
        sandbox_revision = str(opts.sandbox.metadata.source_revision or "")
        if not manifest_revision:
            raise ValueError(
                "sandbox execution requires a revision-bearing RepoManifest"
            )
        if manifest_revision != sandbox_revision:
            raise ValueError(
                "sandbox source revision does not match RepoManifest.commit: "
                f"{sandbox_revision!r} != {manifest_revision!r}"
            )
        manifest_fingerprint = str(manifest.source_fingerprint or "")
        sandbox_fingerprint = str(opts.sandbox.metadata.source_fingerprint or "")
        if not manifest_fingerprint or not sandbox_fingerprint:
            raise ValueError(
                "sandbox execution requires matching source fingerprints in the "
                "manifest and sandbox metadata"
            )
        if not (
            is_secure_source_fingerprint_v2(manifest_fingerprint)
            and is_secure_source_fingerprint_v2(sandbox_fingerprint)
        ):
            raise ValueError(
                "sandbox execution requires matching secure source fingerprint "
                "v2 identities"
            )
        if manifest_fingerprint != sandbox_fingerprint:
            raise ValueError(
                "sandbox source fingerprint does not match RepoManifest: "
                f"{sandbox_fingerprint!r} != {manifest_fingerprint!r}"
            )

    # --- 1c. Surface allowed_skills ↔ compile_table mismatches before any
    # work happens. Two failure modes the caller almost certainly didn't
    # intend; both fail at runtime ("Skill 'X' not available") deep in
    # the agent loop, which is a miserable debugging path.
    _warn_on_skill_set_mismatch(
        set(opts.allowed_skills) if opts.allowed_skills else None,
        table,
    )

    # --- 2. Reset the singleton registry so this query's contexts win.
    # SkillLoader.load_all() skips already-registered skills, so without a
    # reset we'd run against whatever the previous query loaded.
    SkillRegistry.reset()
    registry = SkillRegistry()
    skills_dir = opts.skills_dir or str(_DEFAULT_SKILLS_DIR)
    loader = SkillLoader()

    # --- 3. Resolve contexts according to the selected mode.
    # ``build_skill_contexts`` (repo_path mode) reads each skill's
    # index_requirements straight off ``config.yaml`` on disk (see #154), so
    # it no longer needs a metadata-only pre-pass of SkillLoader. The manifest
    # mode still reads requirements from the registry, so it keeps its
    # pre-pass + reset. The ``contexts`` mode skips both — the caller already
    # did the equivalent.
    if opts.contexts is not None:
        contexts = opts.contexts
    elif manifest is not None:
        # AoT mode: load indexes directly from the manifest's recorded
        # paths. No build step.
        loader.load_all(skills_dir, contexts={}, registry=registry)
        skill_ids = (
            list(opts.allowed_skills)
            if opts.allowed_skills
            else _discover_skill_ids(skills_dir)
        )
        from ..compiler import load_contexts_from_manifest

        contexts = load_contexts_from_manifest(
            manifest,
            skill_ids=skill_ids,
            skill_registry=registry,
            default_top_k=opts.default_top_k,
            default_level=opts.default_level,
            native_index_authorization=opts.native_index_authorization,
            native_index_authorization_resolver=(
                opts.native_index_authorization_resolver
            ),
        )
        SkillRegistry.reset()
        registry = SkillRegistry()
    else:
        # repo_path mode: inline build via build_skill_contexts. It resolves
        # index_requirements from config.yaml on disk, so no metadata pre-pass
        # (and no reset to undo it) is needed before the real load in step 4.
        contexts = _build_contexts(opts, table=table)

    # --- 4. Load (or reload) the skill packages with the real contexts ---
    loaded = loader.load_all(skills_dir, contexts=contexts, registry=registry)
    logger.debug("query(): loaded %d skills from %s", len(loaded), skills_dir)

    # Apply caller-supplied per-skill default overrides (Layer 1+2 of
    # ``resolve_params``). We mutate the metadata in place because the
    # registry holds references; this only affects the current process,
    # and ``SkillRegistry.reset()`` on the next call clears the slate.
    if opts.skill_params:
        for skill_id, overrides in opts.skill_params.items():
            meta = registry.get(skill_id)
            if meta is None:
                logger.warning(
                    "skill_params: skill %r not in registry, ignoring %r",
                    skill_id,
                    sorted(overrides),
                )
                continue
            merged = dict(meta.defaults or {})
            merged.update(overrides)
            meta.defaults = merged

    # --- 5. Build the SessionContext for CAR + parameter scaling ---
    from ..compiler.params import SessionContext

    session_ctx = SessionContext(
        repo_path=opts.repo_path,
        repo_size=opts.repo_size,
        primary_language=opts.primary_language,
        extras=dict(opts.session_extras),
    )

    # --- 6. Construct the runner and execute ---
    # Pass the *resolved* manifest (RepoManifest | None) so ResourceGuard
    # sees the loaded object — never a raw path string.
    runner = AgentRunner(
        llm=opts.llm,
        registry=registry,
        model=model_for_llm,
        temperature=opts.temperature,
        max_tokens=opts.max_tokens,
        system_prompt=opts.system_prompt,
        max_turns=opts.max_turns,
        max_context_tokens=opts.max_context_tokens,
        enable_lsp_route_context=opts.enable_lsp_route_context,
        lsp_route_seed_limit=opts.lsp_route_seed_limit,
        lsp_route_seed_policy=opts.lsp_route_seed_policy,
        lsp_route_query_fallback=opts.lsp_route_query_fallback,
        lsp_route_top_k=opts.lsp_route_top_k,
        lsp_route_include_neighbors=opts.lsp_route_include_neighbors,
        retry=opts.retry,
        allow_skills=set(opts.allowed_skills) if opts.allowed_skills else None,
        exclude_skills=set(opts.excluded_skills) if opts.excluded_skills else None,
        manifest=manifest,
        session_ctx=session_ctx,
        compile_table=table,
        sandbox=opts.sandbox,
    )
    return runner.run(prompt)

normalize_repository_explorer_policy

normalize_repository_explorer_policy(value: str) -> str

Return the canonical repository-explorer policy name.

Source code in codenib/agent/runtime/explorer.py

def normalize_repository_explorer_policy(value: str) -> str:
    """Return the canonical repository-explorer policy name."""

    normalized = str(value or "auto").strip().lower().replace("-", "_")
    aliases = {"sparse": "bm25", "semantic": "dense", "rrf": "hybrid"}
    normalized = aliases.get(normalized, normalized)
    if normalized not in REPOSITORY_EXPLORER_POLICIES:
        supported = ", ".join(sorted(REPOSITORY_EXPLORER_POLICIES))
        raise ValueError(
            f"unknown repository explorer policy; choose from: {supported}"
        )
    return normalized

repository_explorer_build_views

repository_explorer_build_views(policy: str) -> tuple[str, ...]

Return the deterministic materialization set for a benchmark policy.

Source code in codenib/agent/runtime/explorer.py

def repository_explorer_build_views(policy: str) -> tuple[str, ...]:
    """Return the deterministic materialization set for a benchmark policy."""

    normalized = normalize_repository_explorer_policy(policy)
    if normalized == "auto":
        return _NATIVE_VIEWS
    return _POLICY_VIEWS[normalized]

registry_to_tools

registry_to_tools(
    registry: SkillRegistry | None = None,
    *,
    allow: set[str] | None = None,
    exclude: set[str] | None = None
) -> list[dict[str, Any]]

Convert all skills in the registry to OpenAI tool schemas.

Parameters:

Name	Type	Description	Default
registry	SkillRegistry | None	Registry to read from; defaults to the singleton.	None
allow	set[str] | None	If provided, only skills in this set are included (allowlist, applied first). If None, all registered skills are eligible.	None
exclude	set[str] | None	Skill IDs to skip (denylist, applied after allow).	None

Returns:

Type	Description
list[dict[str, Any]]	List of tool dicts ready for litellm.completion(tools=...).

Source code in codenib/agent/tool_schema.py

def registry_to_tools(
    registry: Optional[SkillRegistry] = None,
    *,
    allow: Optional[Set[str]] = None,
    exclude: Optional[Set[str]] = None,
) -> List[Dict[str, Any]]:
    """Convert all skills in the registry to OpenAI tool schemas.

    Args:
        registry: Registry to read from; defaults to the singleton.
        allow: If provided, only skills in this set are included (allowlist,
            applied first). If ``None``, all registered skills are eligible.
        exclude: Skill IDs to skip (denylist, applied after ``allow``).

    Returns:
        List of tool dicts ready for ``litellm.completion(tools=...)``.
    """
    reg = registry or SkillRegistry()
    exclude = exclude or set()
    tools: List[Dict[str, Any]] = []

    for skill_id, meta in reg.list_skills().items():
        if allow is not None and skill_id not in allow:
            continue
        if skill_id in exclude:
            continue
        tools.append(skill_to_tool_schema(meta))

    return tools

skill_to_tool_schema

skill_to_tool_schema(meta: SkillMetadata) -> dict[str, Any]

Convert a single SkillMetadata to an OpenAI function tool dict.

Source code in codenib/agent/tool_schema.py

def skill_to_tool_schema(meta: SkillMetadata) -> Dict[str, Any]:
    """Convert a single ``SkillMetadata`` to an OpenAI function tool dict."""
    return _function_schema(
        meta.skill_id, meta.skill_doc or meta.description or "", meta.inputs
    )