Application services

The use-case services wired by the composition root (server.py / __main__.py / storage/factories.py). They depend only on the storage Protocols and take a uow_factory closure — see the architecture overview.

Application-layer use-case services (spec §5.1).

Thin orchestration objects composed from Protocol-only constructor arguments — PackageStore / ChunkStore / ModuleMemberStore on the storage side, CodeRetrieverPipeline on the retrieval side.

Write-side bootstrap (ProjectIndexer) composes with the strategy classes from pydocs_mcp.extraction (PipelineChunkExtractor / AstMemberExtractor / InspectMemberExtractor / StaticDependencyResolver). The query services ship alongside so the CLI (__main__) can wire a full indexing + query stack with one import. Rendering helpers in pydocs_mcp.application.formatting are the single source of truth for byte-level output but are imported directly by their consumers.

class pydocs_mcp.application.ApiSearch(member_pipeline)

Bases: object

Runs the module-member retrieval pipeline and wraps its state as a SearchResponse.

Mirrors DocsSearch but for the module-member pipeline: deliberately thin and substitutes an empty ModuleMemberList when the pipeline returns no result.

NOTE (spec S21): this class and DocsSearch are intentionally near-duplicate thin wrappers — see the longer note on DocsSearch for the rationale. Briefly: the duplication is grep-friendly and avoids over-parameterization; a third near-identical service is the trigger to factor out a shared base, not the second.

Parameters:

member_pipeline (CodeRetrieverPipeline)

async ranked(query)

Return the RANKED candidate members (pre composite collapse).

Mirrors DocsSearch.ranked() — multi-repo union needs per-item members (score + package / module / name metadata) to merge and dedup across databases.

Parameters:

query (SearchQuery)

Return type:

ModuleMemberList

class pydocs_mcp.application.DocsSearch(chunk_pipeline)

Bases: object

Runs the chunk retrieval pipeline and wraps its state as a SearchResponse.

Deliberately thin: all ranking/filtering logic lives in the pipeline stages. This class only threads query → pipeline → response.

Reads state.result first (composite output from token_budget_formatter — what chunk_search.yaml produces) and falls back to state.candidates (ranked top-K from chunk_search_ranked.yaml-style presets that omit the formatter). Avoids the silent-empty-results footgun if a deployment overlays the ranked preset onto the MCP server.

NOTE (spec S21): this class and ApiSearch are intentionally near-duplicate thin wrappers. Keeping them as two separate classes instead of one parameterized PipelineSearchService is deliberate — the duplication is grep-friendly (a developer looking for “where do chunk searches happen” finds exactly DocsSearch) and avoids over-parameterization. If a third near-identical service appears in the future, that’s the trigger to parameterize into a shared base class — not now.

Parameters:

chunk_pipeline (CodeRetrieverPipeline)

async ranked(query)

Return the RANKED candidate chunks (pre composite collapse).

Multi-repo union needs item-level candidates — each chunk’s per-item relevance score and package / qualified_name metadata — to merge and dedup across databases, which the single composite search output cannot provide. Reads state.candidates; falls back to state.result for a ranked preset that skips the formatter.

Parameters:

query (SearchQuery)

Return type:

ChunkList

class pydocs_mcp.application.IndexingService(uow_factory, node_scores_enabled=False)

Bases: object

Coordinates atomic write-side indexing through a UnitOfWork (spec §5.6).

Single dependency — uow_factory: Callable[[], UnitOfWork]. The service opens a UoW per public-method call, drives the write sequence inside it, and commits. All writes are atomic; partial indexing state never becomes visible (eng-review §14 bug #4).

Parameters:

• uow_factory (Callable[[], UnitOfWork])

• node_scores_enabled (bool)

async reindex_package(package, chunks, module_members, trees=(), references=(), reference_aliases=None, class_attribute_types=None, decisions=(), decision_structured=None, project_root=None)

Replace every row for package.name atomically (spec §13.3).

Canonical order: diff chunks by content_hash (delete removed + insert added, keep unchanged in place) → delete members → delete pkg → upsert pkg → trees (delete then save_many) → upsert members → delete references for package → write resolved references → cross-package re-resolution UPDATE → commit. The chunks-side diff-merge replaces the legacy delete + upsert pair so unchanged rows survive (and their vectors with them).

references is emitted by ReferenceCaptureStage; reference_aliases is its sibling alias map. class_attribute_types is the per-class self.X → <type> table built by capture_self_attribute_types — feeds the resolver’s Rule 0 for cross-method self.X.Y inference. The resolver runs inside this method using the cross-package qname universe loaded from uow.trees (so it sees the just-upserted trees).

decisions is the merged RawDecision tuple emitted by the capture_decisions sub-pipeline (project targets only; dependency packages pass ()). They are reconciled + persisted BEFORE the chunk diff so each decision chunk’s decision_id metadata can be stamped from the decision_key → id map before it lands (spec §D8-§D10). project_root is where the staleness scorer os.stats the affected files; when absent, staleness is left at 0.

Implementation: a thin orchestrator over _persist_decisions() (reconcile + upsert + delete + backlink map), _diff_merge_chunks() (chunk diff + stale-vector cleanup) and _persist_references() (sweep + resolve + save + cross-package re-resolution). Each helper is independently testable; the orchestrator reads as a sequence of named writes under one UoW (Task 7 I2).

Parameters:

• package (Package)

• chunks (tuple[Chunk, ...])

• module_members (tuple[ModuleMember, ...])

• trees (Sequence[DocumentNode])

• references (Sequence[NodeReference])

• reference_aliases (dict[str, dict[str, str]] | None)

• class_attribute_types (dict[str, dict[str, str]] | None)

• decisions (Sequence[RawDecision])

• decision_structured (Mapping[str, tuple[dict[str, object], str]] | None)

• project_root (Path | None)

Return type:

None

async remove_package(name)

Delete a package and every chunk / member / tree / ref it owns.

Capture the soon-to-be-stale chunk IDs BEFORE deleting from SQLite, then wipe their vectors from the (real or null) backend after. Without this, a package’s vectors outlive its SQLite rows on composite deployments and pollute future similarity searches with orphaned embeddings. Atomic via the surrounding UoW transaction. Spec S15 — uow.vectors is always present; SQLite-only deployments route through NullVectorStore (silent no-op).

Parameters:

name (str)

Return type:

None

async clear_all()

Wipe every row across all five entity stores + every vector.

Spec I3 — UnitOfWork.delete_all() drives the per-store sweep in one call. Atomic within the UoW transaction. The vectors clear (NullVectorStore on SQLite-only deployments, TurboQuant on composite deployments) is part of that single delete sequence.

Return type:

None

async recompute_node_scores()

Recompute the node_scores table over the FULL reference graph.

A single post-index pass — global PageRank / Louvain communities must see the fully-resolved cross-package graph, so this runs ONCE after ProjectIndexer.index_project() finishes (and its cross-package re-resolution), NOT per package. No-op unless node_scores_enabled; degrades gracefully (logs a warning) when the [graph] extra is absent, leaving the table empty so the rerank steps simply no-op.

Return type:

None

async find_stale_packages(*, current_model)

Return packages whose stored embedding_model differs from current_model.

Read-only companion to invalidate_stale_embeddings(), which additionally clears each stale package’s content_hash in the same transaction. See _stale_packages() for why embedding_model is None rows are skipped.

Parameters:

current_model (str)

Return type:

list[str]

async invalidate_stale_embeddings(*, current_model)

Clear content_hash on every package embedded with another model.

One UoW = one transaction: the stale-set read and the clearing upserts land atomically — no read/write gap where a concurrent index pass could observe half-cleared state. An empty content_hash never equals a freshly-extracted package’s real hash, so the skip check in ProjectIndexer (existing.content_hash == pkg.content_hash) falls through to a full re-extract + re-embed under the current model.

Returns the stale package names (for the caller’s log line).

Parameters:

current_model (str)

Return type:

list[str]

exception pydocs_mcp.application.InvalidArgumentError

Bases: MCPToolError

Semantic validation failure — input parsed but domain-invalid.

Pydantic ValidationError covers schema-level failures; this is for post-parse checks (e.g., show="inherits" on a non-class target).

class pydocs_mcp.application.LookupInput(*, target='', show='default', project='', limit=<factory>)

Bases: BaseModel

Internal routing input for the deprecated lookup CLI verb — the MCP surface exposes get_symbol / get_references instead.

Parameters:

• target (str)

• show (Literal['default', 'tree', 'callers', 'callees', 'inherits', 'impact', 'context', 'governed_by'])

• project (str)

• limit (int)

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydocs_mcp.application.LookupService(package_lookup, tree_svc, ref_svc, impact_max_depth=3, context_max_depth=2, context_token_budget=2048, context_render='skeleton', context_body_ratio=0.35, cross_navigator=<factory>)

Bases: object

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 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 (pydocs_mcp.storage.factories) always wires the real TreeService / 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.

Parameters:

• package_lookup (PackageLookup)

• tree_svc (TreeNavigator)

• ref_svc (ReferenceNavigator)

• impact_max_depth (int)

• context_max_depth (int)

• context_token_budget (int)

• context_render (str)

• context_body_ratio (float)

• cross_navigator (CrossNavigator)

async lookup(payload)

Text-only façade over lookup_with_items() — one dispatch run, first element. Kept for the deprecated lookup alias and direct callers that never consume structured rows.

Parameters:

payload (LookupInput)

Return type:

str

async lookup_with_items(payload)

Dispatch + render one lookup, returning the envelope body triple.

Tree-rendering branches (module target, show in _TREE_SHOWS) 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. impact/context carry empty items[] — they render ranked NODES, not graph edges, so the §3.5 edge rows don’t apply.

Parameters:

payload (LookupInput)

Return type:

tuple[str, tuple[dict[str, Any], …], dict[str, Any]]

async context_nodes(target)

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

NotFoundError propagates unchanged (bad package / module / symbol), as does ServiceUnavailableError from a NullReferenceService.

Parameters:

target (str)

Return type:

tuple[str, tuple[ContextNode, …], dict[str, Any]]

render_context_card(target, nodes, *, token_budget)

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.

Parameters:

• target (str)

• nodes (tuple[ContextNode, ...])

• token_budget (int)

Return type:

str

exception pydocs_mcp.application.MCPToolError

Bases: PydocsMCPError

Base — every handler-raised error inherits from this.

class pydocs_mcp.application.ModuleInspector(uow_factory)

Bases: object

Live-import a package/submodule and render its public API.

Depends only on uow_factory — opens a UoW per inspect call to check the package is indexed before crossing the import boundary.

Parameters:

uow_factory (Callable[[], UnitOfWork])

exception pydocs_mcp.application.NotFoundError

Bases: MCPToolError

Specified target doesn’t exist in the index.

Raised by lookup for unknown packages/modules/symbols. NOT raised by search — an empty search returns success with an empty-result string.

class pydocs_mcp.application.PackageLookup(uow_factory)

Bases: object

Composes the three domain stores (via UoW) into a read-only package view.

Post-#5a-2: depends only on uow_factory. Each public method opens a fresh UoW; reads run inside async with and exit without committing.

Note: get_package_doc no longer uses asyncio.gather for the two list reads — both go through the same held connection inside the UoW, and concurrent access would race _sqlite_transaction’s lock. Per spec §7.2, the two ~1ms SELECTs run sequentially.

Parameters:

uow_factory (Callable[[], UnitOfWork])

class pydocs_mcp.application.ProjectIndexer(indexing_service, dependency_resolver, chunk_extractor, member_extractor, uow_factory)

Bases: object

Coordinates project + dependency indexing, returning fresh stats.

Post-#5a-2: takes its own uow_factory for the hash-cache check (was a reach-through to indexing_service.package_store.get — eng plan-review #4). Composition root wires the same factory to both IndexingService and ProjectIndexer.

Parameters:

• indexing_service (IndexingService)

• dependency_resolver (DependencyResolver)

• chunk_extractor (ChunkExtractor)

• member_extractor (MemberExtractor)

• uow_factory (Callable[[], UnitOfWork])

class pydocs_mcp.application.ReferenceService(uow_factory, project_name='', cross_links=<factory>)

Bases: object

Reads the cross-node reference graph through a per-call UnitOfWork.

All three public methods open a fresh UoW, read via uow.references, and return tuples (not lists — frozen+hashable contract for the rendering layer downstream).

2-arg signature note (controller decision C1): callers and callees take (package, target_node_qname) to match the existing LookupService._symbol_lookup call site in tests/application/test_lookup_service.py:357,381. The package argument is informational — the underlying uow.references.find_* calls remain cross-package per spec §6.2 (no package filter). This fixes a spec/test inconsistency in §8.1 (spec was 1-arg). Task 20 amends the spec to match.

Parameters:

• uow_factory (Callable[[], UnitOfWork])

• project_name (str)

• cross_links (CrossLinkStore)

async callers(package, target_node_qname)

Return every ref whose to_node_id == target_node_qname.

Cross-package per spec §6.2 — the answer to “who calls X” should not be filtered by source package. The package argument is provided for API symmetry with LookupService._symbol_lookup (which already passes 2 args today) and for downstream rendering context. It is NOT used to filter results.

Cross-repo union (spec §3.4a): overlay edges_into this project’s target append AFTER the local rows (local-first ordering, §A1.8), deduped local-wins on (from_node_id, to_node_id, kind).

Parameters:

• package (str)

• target_node_qname (str)

Return type:

tuple[NodeReference | CrossReferenceRow, …]

async callees(package, from_node_qname)

Return every ref originating from from_node_qname.

Same rationale as callers — package is informational, the storage Protocol is cross-package.

Cross-repo substitution (spec §3.4a): a local UNRESOLVED row whose (to_name, kind) matches an overlay edges_from entry is returned as the resolved, project-qualified cross row instead; a cross edge matching an already-RESOLVED local row is suppressed (the §A1.8 read-side dedup — local wins).

Parameters:

• package (str)

• from_node_qname (str)

Return type:

tuple[NodeReference | CrossReferenceRow, …]

async find_by_name(name, *, kind=None)

Find every ref whose to_name == name (queryable for both resolved AND unresolved edges — that’s the whole point of keeping unresolved rows queryable).

Parameters:

• name (str)

• kind (ReferenceKind | None)

Return type:

tuple[NodeReference, …]

async governed_by(package, node_qname)

GOVERNS edges pointing AT node_qname — its governing decisions (§D18).

The inbound GOVERNS edges (from_node_id='decision:<key>', to_node_id == node_qname, kind='governs') render as reference rows so get_references(direction='governed_by') answers “which decisions govern this symbol?” through the same surface as callers / callees. Resolver-backed (matches on to_node_id, so an unresolved GOVERNS edge naming this qname is excluded). package is informational (governance is cross-package, like callers). Read-only.

Parameters:

• package (str)

• node_qname (str)

Return type:

tuple[NodeReference | CrossReferenceRow, …]

async inherits(package, node_qname)

Both inheritance senses of node_qname — bases first, then subclasses.

BASES: from-side INHERITS edges (from_node_id == node_qname) — the target’s own base list, kept even when the base is unresolved (to_name stores the bare source-text name, e.g. a from-imported base). SUBCLASSES: INHERITS edges INTO the target — resolved (to_node_id == node_qname) plus exact fully-dotted to_name == node_qname matches; never bare-suffix matches (precision bias — a bare Base could be anyone’s). The cross union appends resolved project-qualified subclasses from sibling bundles (spec §3.4a). package is informational, as everywhere on this service.

Parameters:

• package (str)

• node_qname (str)

Return type:

tuple[NodeReference | CrossReferenceRow, …]

async impact(package, qname, *, max_depth, limit)

Ranked blast-radius: who transitively calls qname (what breaks).

Walks the reference graph BACKWARD up to max_depth hops, then ranks by (hop asc, pagerank desc, in_degree desc, qname asc) and slices to limit. PageRank comes from the index-time node_scores table when enabled; otherwise every pagerank is 0.0 and the sort collapses to fan-in (in-degree) ranking — no [graph] extra required. package is informational (the walk is cross-package, like callers). Read-only: no commit.

Parameters:

• package (str)

• qname (str)

• max_depth (int)

• limit (int)

Return type:

tuple[ImpactNode, …]

async context(package, qname, *, max_depth, limit)

Smart-context: the seed’s dependency closure, ready for graded packing.

Walks FORWARD up to max_depth hops (what qname transitively calls), ranks the callees by (hop asc, pagerank desc, in_degree desc, qname), keeps the seed (hop 0) plus the top limit - 1 callees, and hydrates each with its source text from the chunk store (the only content persisted; the renderer derives the signature/outline tiers from it). The seed is ALWAYS first (the focus). PageRank comes from node_scores when enabled; otherwise ranking collapses to fan-in. Rendering (format_context) packs these under the token budget at graded fidelity. Read-only; package informational (cross-package walk).

Parameters:

• package (str)

• qname (str)

• max_depth (int)

• limit (int)

Return type:

tuple[ContextNode, …]

class pydocs_mcp.application.SearchInput(*, query, kind='any', package='', scope='all', project='', limit=<factory>)

Bases: BaseModel

Input for the search_codebase MCP tool.

Parameters:

• query (str)

• kind (Literal['docs', 'api', 'any', 'decision'])

• package (str)

• scope (Literal['project', 'deps', 'all'])

• project (str)

• limit (int)

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

exception pydocs_mcp.application.ServiceUnavailableError

Bases: MCPToolError

Backend raised an unexpected error (SQLite, pipeline) or a required optional service is missing (e.g. tree_svc is None but show="tree" was requested).

class pydocs_mcp.application.TreeService(uow_factory)

Bases: object

Fetches DocumentNode trees through a per-call UnitOfWork.

Parameters:

uow_factory (Callable[[], UnitOfWork])

async pydocs_mcp.application.run_index_pass(*, orchestrator, indexing_service, pipeline_hash, project, embedding_provider, embedding_model, embedding_dim, force, include_project_source, include_dependencies, workers, check_integrity, rebuild_fts, stamp_metadata, write_aggregates)

Run one end-to-end indexing pass; return the orchestrator’s stats.

Example:

bundle = build_project_indexer(config, db_path, use_inspect=True, inspect_depth=None)
stats = await run_index_pass(
    orchestrator=bundle.orchestrator,
    indexing_service=bundle.indexing_service,
    pipeline_hash=bundle.pipeline_hash,
    project=project,
    embedding_provider=config.embedding.provider,
    embedding_model=config.embedding.model_name,
    embedding_dim=config.embedding.dim,
    force=False,
    include_project_source=True,
    include_dependencies=True,
    workers=4,
    check_integrity=bundle.check_integrity,
    rebuild_fts=bundle.rebuild_fts,
    stamp_metadata=bundle.stamp_metadata,
    write_aggregates=bundle.write_aggregates,
)

write_aggregates runs LAST (after the stamp): a bounded index-time git log spawn feeds the §D17 overview activity block; a no-op closure when overview.git_activity.enabled is false.

Parameters:

• orchestrator (ProjectIndexer)

• indexing_service (IndexingService)

• pipeline_hash (str)

• project (Path)

• embedding_provider (str)

• embedding_model (str)

• embedding_dim (int)

• force (bool)

• include_project_source (bool)

• include_dependencies (bool)

• workers (int)

• check_integrity (Callable[[], Awaitable[list[str]]])

• rebuild_fts (Callable[[], Awaitable[None]])

• stamp_metadata (Callable[[IndexMetadata], None])

• write_aggregates (Callable[[Path], Awaitable[None]])

Return type:

IndexingStats