"""LookupService — unified dispatch for the 'lookup' MCP tool (spec §6).
Routes a single ``LookupInput.target`` string (empty / package /
package.module / package.module.symbol) to the right backing service:
``PackageLookup`` for package metadata, ``TreeService`` for file
structure, ``ReferenceService`` for the call graph.
Service deps are mandatory (post-I9 refactor). Production composition
root (:mod:`pydocs_mcp.storage.factories`) always wires the real
:class:`TreeService` / :class:`ReferenceService`; the
:class:`NullTreeService` / :class:`NullReferenceService` stand-ins in
:mod:`null_services` exist as ready substitutes and are wired today
only by tests. When they are wired, the Null impls raise
``ServiceUnavailableError`` with a YAML-anchored pointer (e.g.
``reference_graph.capture.enabled``) so end users hitting the failure
mode can fix it without reading release notes.
Internal structure:
- :class:`LookupTarget` — frozen value object encapsulating target-string
parsing. ``LookupService.lookup`` becomes a thin dispatcher over the
parsed target shape (I1).
- :data:`_REF_GETTERS` — dispatch table mapping ``show`` strings to
``ref_svc`` calls; replaces a 6-level nested if/elif (I8).
- :data:`_MODULE_ID_VARIANTS` — module-level constant (S4) — single
source of truth for the dotted-prefix probe suffixes.
"""
from __future__ import annotations
from collections.abc import Awaitable, Callable, Sequence
from dataclasses import dataclass, replace
from dataclasses import field as dataclasses_field
from pathlib import Path
from typing import TYPE_CHECKING, Any
from pydocs_mcp.application.cross_repo_navigator import NullCrossRepoNavigator
from pydocs_mcp.application.formatting import (
format_context,
format_impact,
format_package_doc,
format_packages_list,
format_references,
)
from pydocs_mcp.application.listing_targets import (
impact_listing_targets,
reference_listing_targets,
)
from pydocs_mcp.application.mcp_errors import (
InvalidArgumentError,
NotFoundError,
)
from pydocs_mcp.application.mcp_inputs import LookupInput
from pydocs_mcp.application.module_references import (
INHERITS_NEEDS_CLASS,
PACKAGE_NEEDS_MODULE,
log_module_target,
module_impact_rows,
module_importer_rows,
module_internal_qnames,
module_seed_ids,
record_seed_cap,
reject_module_show,
)
from pydocs_mcp.application.package_lookup import PackageLookup
from pydocs_mcp.application.reference_service import CrossReferenceRow
from pydocs_mcp.application.symbol_views import render_outline, render_symbol_card
from pydocs_mcp.application.target_resolution import NullTargetResolver, with_target_fallback
from pydocs_mcp.models import PROJECT_PACKAGE_NAME
from pydocs_mcp.pointer_table import PointerTableConfig
from pydocs_mcp.retrieval.config import (
_DEFAULT_CONTEXT_MAX_DEPTH,
_DEFAULT_CONTEXT_RENDER,
_DEFAULT_CONTEXT_TOKEN_BUDGET,
_DEFAULT_IMPACT_MAX_DEPTH,
_DEFAULT_MAX_MODULE_SEEDS,
_DEFAULT_OUTLINE_RECOVERY_POINTER_COUNT,
_DEFAULT_OUTLINE_TOKEN_BUDGET,
_DEFAULT_SKELETON_BODY_RATIO,
_DEFAULT_SYMBOL_CARD_CHILD_CAP,
)
if TYPE_CHECKING:
# Structural typing targets for the two collaborator fields and the
# ``_REF_GETTERS`` dispatch table: TreeService / NullTreeService and
# ReferenceService / NullReferenceService all conform to the
# navigation Protocols in ``application.protocols``.
from pydocs_mcp.application.protocols import (
CrossNavigator,
ReferenceNavigator,
TargetResolver,
TreeNavigator,
)
from pydocs_mcp.application.reference_service import ContextNode, ImpactNode
from pydocs_mcp.application.target_resolution import TargetRewrite
from pydocs_mcp.extraction.model import DocumentNode
from pydocs_mcp.storage.node_reference import NodeReference
# One rendered lookup body: ``(markdown, items, meta_extras)`` — the envelope
# body-producer triple (contract §2.1; ``application.envelope.BodyResult``).
LookupBody = tuple[str, tuple[dict[str, Any], ...], dict[str, Any]]
# ── Module-level constants (S4, S20 — single source of truth) ────────────
# Module-id variants we probe for each dotted-prefix candidate. The
# bare variant matches Python modules (``pkg.foo`` for ``pkg/foo.py``);
# the suffixed variants match doc/notebook trees whose ids preserve
# the extension to avoid PK collision with sibling .py files (sub-PR
# #5 F20). ``lookup("pkg.foo")`` is the natural user query; without
# this fallback markdown/notebook content would be unreachable unless
# the user knew to type the ``.md`` suffix. Order matters — bare
# variant probed first so a real Python module wins over a sibling
# doc file with the same stem (matches user intent).
_MODULE_ID_VARIANTS: tuple[str, ...] = ("", ".md", ".ipynb")
# ── LookupTarget value object (I1) ───────────────────────────────────────
# Callback shape for ``LookupTarget.parse``. Returns
# ``(module_id, n_consumed_input_parts) | None``. ``module_id`` is the
# full module identifier (possibly carrying a synthetic ``.md`` /
# ``.ipynb`` suffix); ``n_consumed_input_parts`` is the count of the
# user's input dotted-parts that the match covered — NOT
# ``len(module_id.split("."))``, because the synthetic suffix isn't in
# the user's input and slicing by segment count would discard a
# trailing symbol the user actually typed.
LongestModuleFn = Callable[
[str, tuple[str, ...]],
Awaitable["tuple[str, int] | None"],
]
def _split_heading_anchor(target: str) -> tuple[tuple[str, ...], str]:
"""``"docs.guide.md#install"`` → ``(("docs", "guide", "md"), "install")``.
The heading anchor that the markdown, notebook and text-section chunkers
store in a ``module#slug`` qualified name is split off BEFORE the dotted
module walk (ADR 0023 decision (e)). The walk probes dotted prefixes
longest-first, so an anchor glued to the last segment makes every probe
miss: ``settings.toml#tool-widget`` never matches the stored module id
``settings.toml``, and the target resolved to nothing even though the
search row that advertised it came straight out of the index.
"""
head, _hash, anchor = target.partition("#")
return tuple(head.split(".")), anchor
@dataclass(frozen=True, slots=True)
class LookupTarget:
"""Parsed shape of a ``lookup`` target string.
Branches the dispatcher reads:
- ``package is None and module is None`` → empty target →
list all indexed packages.
- ``package is not None and module is None and len(symbol_path) == 0``
→ single-segment target → package overview. Distinguish from
"no module match" via the original input length (caller has it).
- ``module is not None and len(symbol_path) == 0`` → module-only
target → render the module tree.
- ``module is not None and len(symbol_path) > 0`` → symbol within
module → dispatch on ``show`` mode.
``consumed`` is the count of input dotted-parts the module match
covered, kept separate from ``len(module.split("."))`` for the
suffix-probe rationale (see :data:`_MODULE_ID_VARIANTS`).
"""
package: str | None
module: str | None
consumed: int
symbol_path: tuple[str, ...]
@classmethod
async def parse(
cls,
target: str,
*,
longest_module: LongestModuleFn,
) -> LookupTarget:
"""Parse a dotted target string into a frozen :class:`LookupTarget`.
The ``longest_module`` callback resolves the longest indexed
module prefix; it returns ``(module_id, n_consumed) | None`` and
is invoked ONLY when the target has more than one dotted segment
— single-segment targets are package-overview requests that
don't need a module probe (avoids a backend round-trip for the
common "what's in package X" case).
"""
if not target:
return cls(package=None, module=None, consumed=0, symbol_path=())
parts, anchor = _split_heading_anchor(target)
package = parts[0]
# An anchor always names a node INSIDE a module, so it must reach the
# module probe even when the head is a single segment.
if len(parts) == 1 and not anchor:
return cls(
package=package,
module=None,
consumed=1,
symbol_path=(),
)
match_pkg = package
match = await longest_module(package, parts)
if match is None:
# Contract §3 project-code addressing (ADR 0004 fix i):
# project source is stored under the reserved ``__project__``
# package with PREFIXLESS module ids, so a bare
# project-qualified target ("mypkg.mod.thing") never matches
# when parts[0] is treated as the package. Re-probe the full
# dotted string under ``__project__`` before giving up — an
# indexed dependency of the same name wins (probed first).
match_pkg = PROJECT_PACKAGE_NAME
match = await longest_module(PROJECT_PACKAGE_NAME, parts)
if match is None:
# Multi-segment target with no module match — the dispatcher
# raises ``NotFoundError`` using the original target string.
# We collapse to "package-only" shape and leave symbol_path
# empty so callers don't accidentally treat unresolved parts
# as a symbol path.
return cls(
package=package,
module=None,
consumed=1,
symbol_path=(),
)
module, consumed = match
# The anchor is the symbol path's last step: it makes the parsed shape
# a SYMBOL lookup (the dispatcher's branch 4), where the full target
# string — anchor included — matches the stored ``qualified_name``.
return cls(
package=match_pkg,
module=module,
consumed=consumed,
symbol_path=parts[consumed:] + ((anchor,) if anchor else ()),
)
def _pinned_target(rewrite: TargetRewrite) -> LookupTarget:
"""The fallback retry's parse result, pinned to ``__project__`` (spec 2026-09-10 P2).
Never re-parse ``rewrite.canonical``: ``LookupTarget.parse`` probes a
dependency named ``parts[0]`` first, and this repo can index itself as a
dependency (benchmarks/pyproject.toml requires pydocs-mcp), so a re-parse
would render the dependency's node instead of the proven project one.
"""
return LookupTarget(
package=PROJECT_PACKAGE_NAME,
module=rewrite.module,
consumed=len(rewrite.module.split(".")),
symbol_path=rewrite.symbol_path,
)
# ── Reference-graph dispatch table (I8) ──────────────────────────────────
# Lambdas wrap each ``ref_svc`` method so the dispatch table has a
# uniform ``(svc, package, node_id) → awaitable`` shape. ``inherits``
# answers BOTH senses (the target's bases via from-side edges, its
# subclasses via edges into the target — see
# ``ReferenceService.inherits``); ``package`` is informational for
# ``callers`` / ``callees`` (storage is cross-package per spec §6.2)
# but is part of the 2-arg call signature pinned by Decision C1.
_REF_GETTERS: dict[
str,
Callable[
[ReferenceNavigator, str, str],
Awaitable[tuple[NodeReference, ...]],
],
] = {
"callers": lambda svc, p, n: svc.callers(p, n),
"callees": lambda svc, p, n: svc.callees(p, n),
# Name-keyed locally, unioned with overlay INHERITS edges by the
# service (spec 2026-07-11 §3.4a) — hence a first-class method now.
"inherits": lambda svc, p, n: svc.inherits(p, n),
# GOVERNS edges pointing AT the target (decisions-as-graph-nodes, spec §D18):
# "which decisions govern this symbol?" rendered through format_references.
"governed_by": lambda svc, p, n: svc.governed_by(p, n),
}
# Show modes that render a SYMBOL VIEW of a tree/node (``application.symbol_views``)
# rather than a graph answer: ``default`` is the symbol card, ``tree`` the outline.
_TREE_SHOWS: frozenset[str] = frozenset({"default", "tree"})
# The ``show`` behind get_symbol(depth="summary") — the card (ADR 0023 (a)).
# ``tree`` keeps its own rendering, so the two depths are complementary.
_CARD_SHOW: str = "default"
# Extras key carrying a resolved get_references target's file extension (e.g.
# ".py", ".toml") up to ToolRouter, which maps it through the analyzer registry
# into the honest meta.resolution (ADR 0021 Decision 6). ToolRouter STRIPS it
# before the wire meta: get_symbol shares ``_lookup_body`` and must not surface
# it. Only the get_references-only branches (impact + reference-graph) set it.
TARGET_EXTENSION_EXTRA: str = "target_extension"
def _target_extension(source_path: str | None) -> str | None:
"""Lowercase file suffix of a resolved node's ``source_path``, or None.
Analyzer-registry keys are lowercase dotted suffixes (``".py"``); a node with
no ``source_path`` (or a suffixless path) yields None → "unavailable".
"""
if not source_path:
return None
return Path(source_path).suffix.lower() or None
# ── items[] builders (contract §3.3/§3.4, Task 6) ────────────────────────
def _outline_item(node: DocumentNode) -> dict[str, Any]:
"""One §3.3 ``get_symbol`` row — CONTRACT names (``path``/``start_line``/
``end_line``), not the pageindex keys (``source_path``/``start_index``/
``end_index``) the text body carries."""
# str(NodeKind.X) == the wire value (StrEnum) — no ``.value`` reach so
# duck-typed tree doubles carrying plain-str kinds keep working.
return {
"node_id": node.node_id,
"kind": str(node.kind),
"qualified_name": node.qualified_name,
"path": node.source_path or None,
"start_line": node.start_line,
"end_line": node.end_line,
}
def _symbol_view(
node: DocumentNode,
show: str,
*,
card_child_cap: int,
outline_token_budget: int,
outline_recovery_pointer_count: int,
pointers: PointerTableConfig,
) -> tuple[str, tuple[dict[str, Any], ...]]:
"""Render one resolved node at the depth ``show`` asks for (contract §3.3).
``default`` → the symbol card; ``tree`` → the budgeted outline. Either way
the items[] rows carry exactly the node set the text names — for the
outline that is the one deliberate items-pruning in the surface (ADR 0023) —
and either way the follow-up calls come from the pointer table's row for
that view, never from a literal here.
"""
if show == _CARD_SHOW:
card = render_symbol_card(node, child_cap=card_child_cap, pointers=pointers)
return card.text, tuple(_outline_item(member) for member in card.nodes)
outline = render_outline(
node,
token_budget=outline_token_budget,
recovery_pointer_count=outline_recovery_pointer_count,
pointers=pointers,
)
return outline.text, tuple(_outline_item(shown) for shown in outline.nodes)
def _context_item(node: DocumentNode) -> dict[str, Any]:
"""One §3.4 ``get_context`` row for a target's resolved focus node."""
return {
"qualified_name": node.qualified_name,
"kind": str(node.kind),
"path": node.source_path or None,
"start_line": node.start_line,
"end_line": node.end_line,
}
# ── §3.5 reference items (get_references, Task 8) ────────────────────────
# The degraded path/span triple — §3.5 "null on miss" (unresolved endpoint,
# endpoint outside the indexed trees, synthetic decision:<key> nodes,
# cross-repo rows whose trees live in another project).
_NULL_SPAN: tuple[str | None, int | None, int | None] = (None, None, None)
def _reference_item(
row: NodeReference | CrossReferenceRow,
show: str,
span: tuple[str | None, int | None, int | None],
) -> dict[str, Any]:
"""One §3.5 ``get_references`` row. ``to_qualified_name`` degrades to the
captured ``to_name`` for unresolved edges so the intent stays visible
(mirrors the ⚠ text rendering)."""
path, start_line, end_line = span
return {
"from_qualified_name": row.from_node_id,
"to_qualified_name": row.to_node_id if row.to_node_id is not None else row.to_name,
"kind": str(row.kind),
"direction": show,
"path": path,
"start_line": start_line,
"end_line": end_line,
}
# ── LookupService ────────────────────────────────────────────────────────
[docs]
@dataclass(frozen=True, slots=True)
class LookupService:
"""Routes lookup targets to the right backing service.
Post-I9: ``tree_svc`` and ``ref_svc`` are mandatory parameters —
no more ``if X is None:`` guards in the dispatcher. The Null
impls in :mod:`pydocs_mcp.application.null_services` exist as
ready stand-ins that preserve the user-visible
``ServiceUnavailableError`` contract; today only test code wires
them. Production composition root
(:mod:`pydocs_mcp.storage.factories`) always wires the real
:class:`TreeService` / :class:`ReferenceService`. A future
deployment that opts out of tree indexing or reference capture
(via YAML config) could swap the Null impls in at the composition
root without touching this class.
"""
package_lookup: PackageLookup
tree_svc: TreeNavigator
ref_svc: ReferenceNavigator
# Reverse-traversal depth for ``show="impact"``. NOT an MCP param — the
# composition root threads ``reference_graph.impact.max_depth`` here; the
# default keeps direct/test construction working (single source of truth:
# ``retrieval.config._DEFAULT_IMPACT_MAX_DEPTH``).
impact_max_depth: int = _DEFAULT_IMPACT_MAX_DEPTH
# Fan-out cap for a MODULE target's seed set (root + direct members). Same
# posture as ``impact_max_depth``: the composition root threads
# ``reference_graph.impact.max_module_seeds``; the constant keeps direct /
# test construction working.
module_seed_cap: int = _DEFAULT_MAX_MODULE_SEEDS
# Forward-closure depth + token budget for ``show="context"`` — same
# posture (YAML tunables via the composition root, not MCP params).
context_max_depth: int = _DEFAULT_CONTEXT_MAX_DEPTH
context_token_budget: int = _DEFAULT_CONTEXT_TOKEN_BUDGET
# Render strategy + skeleton body budget for ``show="context"`` (spec §D6).
# Skeleton is the shipped default; ``format_context`` reads both.
context_render: str = _DEFAULT_CONTEXT_RENDER
context_body_ratio: float = _DEFAULT_SKELETON_BODY_RATIO
# How many immediate children the symbol card names (ADR 0023). Same posture
# as the knobs above: the composition root threads ``symbol_card.child_cap``.
card_child_cap: int = _DEFAULT_SYMBOL_CARD_CHILD_CAP
# The outline's token budget and how many elided subtrees a cut points at
# (ADR 0023). Same posture again: ``symbol_outline.*`` from the composition
# root, constants here so direct/test construction stays config-free.
outline_token_budget: int = _DEFAULT_OUTLINE_TOKEN_BUDGET
outline_recovery_pointer_count: int = _DEFAULT_OUTLINE_RECOVERY_POINTER_COUNT
# Which follow-up calls each rendered view offers (``output.pointers``).
# Same posture: the composition root threads the deployment's table, and the
# shipped one keeps direct / test construction rendering what a server does.
pointers: PointerTableConfig = dataclasses_field(default_factory=PointerTableConfig)
# Workspace federation (spec 2026-07-11 §3.4b): the impact walk and
# governed_by decision hydration delegate here. The Null impl returns
# the local walk unchanged, so single-project behavior is byte-identical.
cross_navigator: CrossNavigator = dataclasses_field(default_factory=NullCrossRepoNavigator)
# Miss-path target fallbacks (spec 2026-09-10 §2.5). The composition root
# (storage.factories) wires ProjectTargetResolver, or the Null object when
# every target_resolution rule flag is off; this Null default exists for
# direct/test construction only.
target_resolver: TargetResolver = dataclasses_field(default_factory=NullTargetResolver)
[docs]
def on_branch(self, branch: str | None) -> LookupService:
"""This service answering from ``branch`` (spec §6.4, #313): every index
reader it composes is bound to it, so no private method threads the
branch. ``None`` returns the service itself: the served default for the
branch-keyed reads (trees, members, edges), unpinned chunk rows (#312)
for the rest — exact only on a one-branch bundle, the only place the
router passes ``None`` (``read_branch_of``)."""
if branch is None:
return self
return replace(
self,
package_lookup=self.package_lookup.on_branch(branch),
tree_svc=self.tree_svc.on_branch(branch),
ref_svc=self.ref_svc.on_branch(branch),
target_resolver=self.target_resolver.on_branch(branch),
)
[docs]
async def lookup(self, payload: LookupInput, *, branch: str | None = None) -> str:
"""Text-only façade over :meth:`lookup_with_items` — one dispatch run,
first element. Kept for the deprecated ``lookup`` alias and direct
callers that never consume structured rows."""
body, _items, _extras = await self.lookup_with_items(payload, branch=branch)
return body
[docs]
async def lookup_with_items(
self, payload: LookupInput, *, branch: str | None = None
) -> LookupBody:
"""Dispatch + render one lookup, returning the envelope body triple.
Tree-rendering branches (``show`` in ``_TREE_SHOWS``, whatever the
target shape) emit one contract-§3.3 row per rendered outline node;
reference-graph branches (callers/callees/inherits/governed_by) emit
one §3.5 row per rendered edge — for a MODULE target too, which answers
its import graph (see :meth:`_module_target`).
``impact``/``context`` carry empty items[] — they render
ranked NODES, not graph edges, so the §3.5 edge rows don't apply.
The exact path runs first; only its ``NotFoundError`` consults
``target_resolver`` for one pinned retry (spec 2026-09-10 §2.5), so a
target that resolves today never reaches the resolver. ``branch``: see
:meth:`on_branch`.
"""
reader = self.on_branch(branch)
return await with_target_fallback(
payload.target,
entry="lookup",
resolver=reader.target_resolver,
run_exact=lambda: reader.lookup_exact(payload),
run_rewrite=lambda rewrite: reader.lookup_rewritten(payload, rewrite),
)
[docs]
async def lookup_exact(self, payload: LookupInput) -> LookupBody:
"""The exact-only path, no fallback — multi-project pass 1 runs this so
an exact hit in any project beats a rewrite in another (AC13)."""
parsed = await LookupTarget.parse(payload.target, longest_module=self._longest_module)
return await self._dispatch_parsed(payload, parsed)
[docs]
async def lookup_rewritten(self, payload: LookupInput, rewrite: TargetRewrite) -> LookupBody:
"""Dispatch ``rewrite`` pinned to ``__project__``. The canonical name
becomes the target so ``_symbol_lookup``'s full-string node match hits."""
canonical_payload = payload.model_copy(update={"target": rewrite.canonical})
return await self._dispatch_parsed(canonical_payload, _pinned_target(rewrite))
async def _dispatch_parsed(self, payload: LookupInput, parsed: LookupTarget) -> LookupBody:
"""Branch dispatch over a parsed target (the pre-fallback body, unedited)."""
target_str = payload.target
# 1. Empty target → list all indexed packages.
if parsed.package is None:
packages = await self.package_lookup.list_packages()
return format_packages_list(packages), (), {}
# 2. Single-segment target → package overview. Distinguish via
# the original input: if the user typed a multi-segment target
# and we collapsed to "package-only" shape, the module probe
# didn't match and we raise NotFoundError. A target carrying a
# heading anchor is never a package overview — it named a node
# inside a module, so an unresolvable one is a miss, not a request
# for the package card of whatever its head happens to spell.
original_parts, anchor = _split_heading_anchor(target_str)
if parsed.module is None:
if len(original_parts) == 1 and not anchor:
return await self._package_overview(parsed.package, payload.show, payload.limit)
# Multi-segment target but no module match → NotFoundError
# using the user's original string (preserves the pre-refactor
# message shape).
raise NotFoundError(f"no module matching '{target_str}' found under '{parsed.package}'")
# 3. Module-only target → the module tree, or its import graph.
if not parsed.symbol_path:
return await self._module_target(
parsed.package, parsed.module, payload.show, payload.limit
)
# 4. Symbol lookup — ``limit`` flows down into the reference-graph
# branches so YAML-tuned ``reference_graph.output.default_limit``
# caps the rendered row count.
return await self._symbol_lookup(
parsed.package,
parsed.module,
target_str,
payload.show,
payload.limit,
)
async def _package_overview(self, package: str, show: str, limit: int) -> LookupBody:
"""Package overview, with the single-segment project-code fallback.
Contract §3 addressing: a single-segment target can name a project
top-level module / package dir (stored under ``__project__`` with a
prefixless module id) rather than an indexed dependency. The indexed
dependency wins when both exist; the pre-fix NotFoundError message
is preserved for genuine misses.
A graph direction cannot be answered by a package DOC, so it routes to
the package's same-named top module instead (``requests`` →
``requests`` the module) and says so when there is none.
"""
doc = await self.package_lookup.get_package_doc(package)
if doc is not None and show in _TREE_SHOWS:
return format_package_doc(doc, pointers=self.pointers), (), {}
owner = package if doc is not None else PROJECT_PACKAGE_NAME
fallback = await self._longest_indexed_module(owner, [package])
if fallback is not None:
return await self._module_target(owner, fallback[0], show, limit)
if doc is not None:
raise InvalidArgumentError(PACKAGE_NEEDS_MODULE.format(show=show, package=package))
raise NotFoundError(f"package '{package}' not indexed")
async def _module_target(self, package: str, module: str, show: str, limit: int) -> LookupBody:
"""Dispatch one module target: the outline, or its import graph (§1).
``callers`` and ``impact`` fan out over the module's members and have
their own merges; ``callees`` / ``governed_by`` read the module ROOT,
whose ``qualified_name`` is the module id, so ``_symbol_lookup``
resolves it unchanged.
"""
if show in _TREE_SHOWS:
return await self._module_lookup(package, module, show)
reject_module_show(module, show)
if show == "callers":
return await self._module_callers(package, module, limit)
if show == "impact":
return await self._module_impact(package, module, limit)
return await self._symbol_lookup(package, module, module, show, limit)
async def _module_callers(self, package: str, module: str, limit: int) -> LookupBody:
"""Importers of the module: edges into it + IMPORTS into its members."""
root = await self._module_root(package, module)
seeds = module_seed_ids(root, self.module_seed_cap)
record_seed_cap(module, seeds)
rows = await module_importer_rows(self.ref_svc, package, seeds.ids)
log_module_target(module, "callers", len(seeds.ids), len(rows))
extras = {TARGET_EXTENSION_EXTRA: _target_extension(root.source_path)}
return await self._render_reference_rows(package, module, "callers", rows, limit, extras)
async def _module_impact(self, package: str, module: str, limit: int) -> LookupBody:
"""Blast radius of the module AND its members, its own internals removed."""
root = await self._module_root(package, module)
seeds = module_seed_ids(root, self.module_seed_cap)
record_seed_cap(module, seeds)
rows = await module_impact_rows(
self.cross_navigator,
self.ref_svc,
package,
module,
seeds.ids,
module_internal_qnames(root),
max_depth=self.impact_max_depth,
limit=limit,
)
log_module_target(module, "impact", len(seeds.ids), len(rows))
extras = {TARGET_EXTENSION_EXTRA: _target_extension(root.source_path)}
return await self._render_impact(package, module, rows, limit, extras)
def _symbol_view_of(
self, node: DocumentNode, show: str
) -> tuple[str, tuple[dict[str, Any], ...]]:
"""``_symbol_view`` with this deployment's YAML output bounds applied.
One place threads the three knobs, so the module path and the symbol
path cannot drift into rendering the same node at different bounds.
"""
return _symbol_view(
node,
show,
card_child_cap=self.card_child_cap,
outline_token_budget=self.outline_token_budget,
outline_recovery_pointer_count=self.outline_recovery_pointer_count,
pointers=self.pointers,
)
async def _module_root(self, package: str, module: str) -> DocumentNode:
"""The module's stored tree root, or ``NotFoundError``."""
tree = await self.tree_svc.get_tree(package, module)
if tree is None:
raise NotFoundError(f"no tree stored for '{package}.{module}'")
return tree
async def _module_lookup(self, package: str, module: str, show: str) -> LookupBody:
"""The module root's symbol view — its card, or its outline."""
tree = await self._module_root(package, module)
# Honest-resolution channel (ADR 0021 Decision 6): module targets reach
# get_references through THIS path, so an extras-free return would map
# a .py module to "unavailable" (wire-verified regression). Thread the
# module file's own extension; non-reference consumers strip the key.
extras = {TARGET_EXTENSION_EXTRA: _target_extension(tree.source_path)}
text, items = self._symbol_view_of(tree, show)
return text, items, extras
async def _symbol_lookup(
self,
package: str,
module: str,
target: str,
show: str,
limit: int,
) -> LookupBody:
"""Resolve a symbol within a known module and render per ``show``.
Replaces the pre-refactor 6-level nested if/elif with
guard-clauses + a :data:`_REF_GETTERS` dispatch table (I8):
each branch returns early, and the reference-graph dispatch
is one table lookup + one ``await``.
"""
tree = await self.tree_svc.get_tree(package, module)
if tree is None:
raise NotFoundError(f"no tree for '{package}.{module}'")
node = tree.find_node_by_qualified_name(target)
if node is None:
raise NotFoundError(f"'{target}' not found in {module}")
# get_references honest-resolution channel (ADR 0021 Decision 6): the
# target's own file extension, threaded to ToolRouter on EVERY branch —
# get_references reaches the tree branch too (a module target renders
# the outline; wire-verified), so an extras-free tree return would map
# a .py module to "unavailable". Non-reference consumers strip the key.
ref_extras: dict[str, Any] = {TARGET_EXTENSION_EXTRA: _target_extension(node.source_path)}
# Card / outline → the node's symbol view (+ its §3.3 rows).
if show in _TREE_SHOWS:
text, items = self._symbol_view_of(node, show)
return text, items, ref_extras
# Ranked blast-radius — multi-hop REVERSE traversal, its own return
# shape (ranked ImpactNodes) + formatter, so it can't ride _REF_GETTERS.
# Applies to any node kind (unlike ``inherits``). ``limit`` slices the
# rendered rows AFTER the service ranks the full discovered set.
if show == "impact":
impacted = await self.cross_navigator.impact(
self.ref_svc,
package,
node.node_id,
max_depth=self.impact_max_depth,
limit=limit,
)
return await self._render_impact(package, target, impacted, limit, ref_extras)
# Smart-context — forward dependency-closure packed at graded fidelity
# under a token budget. Own return shape (ContextNodes) + formatter.
# ``limit`` caps the candidate node count; the token budget caps output.
# Composed from the two reusable helpers ``get_context`` calls directly
# for its proportional multi-target budget split — behavior here is
# unchanged (full ``context_token_budget`` for the single-target path).
if show == "context":
ctx = await self._context_closure(package, node.node_id, limit=limit)
card = self.render_context_card(target, ctx, token_budget=self.context_token_budget)
return card, (), {}
# Reference-graph dispatch (callers / callees / inherits).
getter = _REF_GETTERS.get(show)
if getter is None:
raise InvalidArgumentError(f"unknown show value: {show}")
# E1: the message names the CLIENT's vocabulary (``direction``), not the
# internal ``show``, so MCP and CLI users read the same sentence.
if show == "inherits" and node.kind != "class":
raise InvalidArgumentError(
INHERITS_NEEDS_CLASS.format(target=target, kind=str(node.kind))
)
# Null impls (deployments without the reference graph) raise
# ``ServiceUnavailableError`` with the YAML-anchored message
# from this same call site — the dispatcher stays branch-free.
rows = await getter(self.ref_svc, package, node.node_id)
return await self._render_reference_rows(package, target, show, rows, limit, ref_extras)
async def _render_reference_rows(
self,
package: str,
target: str,
show: str,
rows: tuple[NodeReference | CrossReferenceRow, ...],
limit: int,
extras: dict[str, Any],
) -> LookupBody:
"""Cap, hydrate decision titles, render + mirror one page of edges.
Cap before render — the service may return more than ``limit`` rows
(cross-package fan-in is unbounded), and ``format_references`` must
receive the same bound the user is shown. Shared by the symbol branch
and the module-importer branch so both pages render identically.
"""
if len(rows) > limit:
rows = rows[:limit]
titles = await self._decision_titles(show, rows)
rendered = format_references(
rows,
target=target,
show=show,
limit=limit,
decision_titles=titles,
pointers=self.pointers,
followup_targets=await self._followup_targets(
package, reference_listing_targets(rows, target=target)
),
)
return rendered, await self._reference_items(rows, show), extras
async def _render_impact(
self,
package: str,
target: str,
rows: tuple[ImpactNode, ...],
limit: int,
extras: dict[str, Any],
) -> LookupBody:
"""Render one blast radius with the follow-up targets its rows introduce.
Shared by the symbol path and the module path so both pages advertise
the same calls. items[] stay empty: a ring lists ranked NODES, not the
graph edges the §3.5 rows describe.
"""
targets = await self._followup_targets(package, impact_listing_targets(rows, target=target))
rendered = format_impact(
rows, target=target, limit=limit, pointers=self.pointers, followup_targets=targets
)
return rendered, (), extras
async def _followup_targets(self, package: str, candidates: Sequence[str]) -> tuple[str, ...]:
"""The candidates a listing's bundle may name — module ids dropped.
A listing offers ONE target set, so it has to hold names every call in
its row can answer, and the strictest is ``get_context``: it packs a
SYMBOL's dependency closure and rejects a module target outright
(:meth:`_context_target_from_parsed`). One module counterpart — an import
edge's endpoint, a module node in a blast radius — would fail the single
batch call that covers every other row. The overview module map records
the same trap; here it is closed by asking the index rather than by
guessing from the name.
The walk stops at the batch ceiling, because a bundle never names more
targets than that: a fifty-row page costs a handful of probes, not fifty.
"""
kept: list[str] = []
for name in candidates:
if len(kept) >= self.pointers.batch_max:
break
if not await self._is_indexed_module(package, name):
kept.append(name)
return tuple(kept)
async def _is_indexed_module(self, package: str, qname: str) -> bool:
"""Whether ``qname`` names an indexed module.
Probes the listing's own package first, then the package its leading
segment names — a resolved callee can live in another one, exactly as
:meth:`_defining_span` finds it.
"""
for pkg in dict.fromkeys((package, qname.split(".", 1)[0])):
if await self.tree_svc.exists(pkg, qname):
return True
return False
async def _decision_titles(self, show: str, rows) -> dict[tuple[str, str], str]:
"""Hydrate cross-repo governed_by rows' decision titles (spec §A1.2).
Only fires for cross rows on the governed_by direction; degraded
(missing record / project) rows simply stay key-only downstream.
"""
if show != "governed_by":
return {}
wanted = tuple(
(r.from_project, r.from_node_id.removeprefix("decision:"))
for r in rows
if isinstance(r, CrossReferenceRow) and r.from_node_id.startswith("decision:")
)
if not wanted:
return {}
return dict(await self.cross_navigator.decision_titles(wanted))
async def _reference_items(
self,
rows: tuple[NodeReference | CrossReferenceRow, ...],
show: str,
) -> tuple[dict[str, Any], ...]:
"""§3.5 rows mirroring the (already limit-sliced) reference rows 1:1.
Path/span come from the resolvable endpoint's DEFINING tree node —
callees attribute the to-node, every other direction the from-node
(per-call-site line numbers are not stored in the graph). Spans are
memoized per ``(package, qname)`` so fan-in from one caller costs one
tree probe.
"""
cache: dict[tuple[str, str], tuple[str | None, int | None, int | None]] = {}
items: list[dict[str, Any]] = []
for row in rows:
items.append(_reference_item(row, show, await self._row_span(row, show, cache)))
return tuple(items)
async def _row_span(
self,
row: NodeReference | CrossReferenceRow,
show: str,
cache: dict[tuple[str, str], tuple[str | None, int | None, int | None]],
) -> tuple[str | None, int | None, int | None]:
"""The one endpoint span for ``row`` (§3.5 direction rule), cached."""
if isinstance(row, CrossReferenceRow):
# The defining node lives in ANOTHER project's trees — probing this
# project's trees could hit a same-named module and lie.
return _NULL_SPAN
qname = row.to_node_id if show == "callees" else row.from_node_id
if qname is None:
return _NULL_SPAN
key = (row.from_package, qname)
if key not in cache:
cache[key] = await self._defining_span(row.from_package, qname)
return cache[key]
async def _defining_span(
self, package: str, qname: str
) -> tuple[str | None, int | None, int | None]:
"""Locate ``qname``'s defining tree node → ``(path, start, end)``.
Probes the row's owning package first, then the qname's leading segment
(a resolved callee may live in a different package than its caller —
dependency qnames carry their package as the first segment). Any miss
degrades to the null span, never an error.
"""
parts = qname.split(".")
for pkg in dict.fromkeys((package, parts[0])):
found = await self._longest_indexed_module(pkg, parts)
if found is None:
continue
tree = await self.tree_svc.get_tree(pkg, found[0])
node = tree.find_node_by_qualified_name(qname) if tree is not None else None
if node is not None:
return node.source_path or None, node.start_line, node.end_line
return _NULL_SPAN
[docs]
async def context_nodes(
self, target: str, *, branch: str | None = None
) -> tuple[str, tuple[ContextNode, ...], dict[str, Any]]:
"""Resolve ``target`` to its forward dependency closure.
Runs the same parse + node-resolution path as ``lookup(show="context")``
but stops BEFORE rendering, returning ``(display_target, nodes,
focus_row)`` so the caller can render one card per target under a
per-card token budget (``ToolRouter.get_context``'s proportional
multi-target split). ``target`` is echoed back verbatim as the card
heading target — matching the single-target ``show="context"`` path;
``focus_row`` is the contract-§3.4 items[] row for the resolved focus
node (Task 6). On a target fallback (spec 2026-09-10 §2.5) the
canonical name is the display target instead — only for calls that
fail today.
``NotFoundError`` propagates (bad package / module / symbol, plus any
closest-name candidates); ``ServiceUnavailableError`` from a
``NullReferenceService`` is never caught by the fallback. ``branch``:
see :meth:`on_branch`.
"""
reader = self.on_branch(branch)
return await with_target_fallback(
target,
entry="context",
resolver=reader.target_resolver,
run_exact=lambda: reader.context_nodes_exact(target),
run_rewrite=reader.context_nodes_rewritten,
)
[docs]
async def context_nodes_exact(
self, target: str
) -> tuple[str, tuple[ContextNode, ...], dict[str, Any]]:
"""Exact-only context resolution, no fallback (multi-project pass 1)."""
parsed = await LookupTarget.parse(target, longest_module=self._longest_module)
package, node = await self._context_target_from_parsed(target, parsed)
return await self._context_bundle(target, package, node)
[docs]
async def context_nodes_rewritten(
self, rewrite: TargetRewrite
) -> tuple[str, tuple[ContextNode, ...], dict[str, Any]]:
"""Context for ``rewrite`` pinned to ``__project__``; the canonical name
is the display target (card heading and focus row)."""
pinned = _pinned_target(rewrite)
package, node = await self._context_target_from_parsed(rewrite.canonical, pinned)
return await self._context_bundle(rewrite.canonical, package, node)
async def _context_bundle(
self, display_target: str, package: str, node: DocumentNode
) -> tuple[str, tuple[ContextNode, ...], dict[str, Any]]:
# ``limit`` mirrors the single-target ``lookup(show="context")`` path,
# which flows the ``LookupInput.limit`` default (single source of truth:
# ``reference_graph.output.default_limit`` via the YAML-backed slot).
limit = LookupInput(target=display_target).limit
nodes = await self._context_closure(package, node.node_id, limit=limit)
return display_target, nodes, _context_item(node)
[docs]
def render_context_card(
self,
target: str,
nodes: tuple[ContextNode, ...],
*,
token_budget: int,
) -> str:
"""Render one context card via ``format_context`` at ``token_budget``.
Threads the service's render strategy + skeleton body ratio so a card
rendered here is byte-identical to the single-target path at the same
budget. Pure (no I/O) — the ``get_context`` split renders every card
with the same helper at each target's proportional share.
"""
return format_context(
nodes,
target=target,
token_budget=token_budget,
render=self.context_render,
body_ratio=self.context_body_ratio,
pointers=self.pointers,
)
async def _context_closure(
self, package: str, node_id: str, *, limit: int
) -> tuple[ContextNode, ...]:
"""The ``ref_svc.context`` call, shared by the sync and split paths."""
return await self.ref_svc.context(
package,
node_id,
max_depth=self.context_max_depth,
limit=limit,
)
async def _context_target_from_parsed(
self, target: str, parsed: LookupTarget
) -> tuple[str, DocumentNode]:
"""Parsed ``target`` → ``(package, focus_node)`` for a symbol-level
context request. Raises ``NotFoundError`` with the same messages the
``lookup`` dispatcher surfaces (unresolved package / module / symbol)."""
if parsed.module is None or not parsed.symbol_path:
raise NotFoundError(f"no symbol matching '{target}' found for context closure")
tree = await self.tree_svc.get_tree(parsed.package, parsed.module)
if tree is None:
raise NotFoundError(f"no tree for '{parsed.package}.{parsed.module}'")
node = tree.find_node_by_qualified_name(target)
if node is None:
raise NotFoundError(f"'{target}' not found in {parsed.module}")
return parsed.package, node
async def _longest_module(
self,
package: str,
parts: tuple[str, ...],
) -> tuple[str, int] | None:
"""Adapter that exposes :meth:`_longest_indexed_module` to
:meth:`LookupTarget.parse` with the callback shape it expects.
Kept as a thin wrapper rather than passing the method directly
so the parse logic stays decoupled from how we walk dotted
prefixes — a future refactor can swap the walker without
touching ``LookupTarget``.
"""
return await self._longest_indexed_module(package, list(parts))
async def _longest_indexed_module(
self,
package: str,
parts: list[str],
) -> tuple[str, int] | None:
"""Walk longest-prefix-first; return ``(module_id, n)`` where n is
the count of input parts consumed by the match, or ``None``.
For each dotted-prefix candidate we probe three module-id shapes:
the bare name (``pkg.foo`` — Python module), and the two
doc/notebook suffixed forms (``pkg.foo.md`` / ``pkg.foo.ipynb``).
The bare form is tried first so a real Python module wins over
a sibling doc file with the same stem, matching the user's most
likely intent. The suffix is ours, never the user's — so
``consumed`` reflects the prefix length ``i`` regardless of
which variant matched (the caller uses ``consumed`` to slice
``parts`` for the symbol-path remainder).
Probes ``tree_svc.exists`` first (cheap row check, no JSON
parse); if the Null impl is wired ``exists`` returns ``False``
and we fall through to ``PackageLookup.find_module`` — the
no-tree-index deployment still resolves modules.
Example (spec S30) — target ``"fastapi.routing.APIRouter.include_router"``:
* Caller (:meth:`LookupTarget.parse`) splits to
``parts = ["fastapi", "routing", "APIRouter", "include_router"]``
and dispatches ``package = "fastapi"``.
* If only ``fastapi.routing`` is indexed (the class + method are NOT
separate modules in the tree), this walk tries ``i=4`` → ``i=3``
→ ``i=2`` and matches at ``i=2`` with ``module_id = "fastapi.routing"``.
* Returns ``("fastapi.routing", 2)`` — ``consumed == 2`` indexes
INTO ``parts`` so the caller's ``symbol_path`` becomes
``parts[2:] = ("APIRouter", "include_router")``. Note that
``consumed`` is the prefix LENGTH (including the package name
slot at ``parts[0]``), NOT ``len(module.split('.'))`` of just
the post-package suffix.
"""
for i in range(len(parts), 0, -1):
candidate = ".".join(parts[:i])
for suffix in _MODULE_ID_VARIANTS:
variant = candidate + suffix
if await self.tree_svc.exists(package, variant):
return (variant, i)
if await self.package_lookup.find_module(package, variant):
return (variant, i)
return None