From cba39b862022a9fb0c24a2715ed0632de04250fa Mon Sep 17 00:00:00 2001 From: Dmitry Voropaev Date: Tue, 7 Jul 2026 15:44:34 +0300 Subject: [PATCH] feat(extract): deterministic business-process path extractor + Tier-1 processes gate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Step 2 toward the grounded business-process extractor (DESIGN §14 item 2): materialize NAMED REAL PATHS deterministically — an extracted entrypoint (api_route / event_handler handler) -> a shortest chain of resolved first-party call edges -> a terminal function containing a SINK call. The LLM labeler is the next slice; the path itself needs no model. This is the first SECOND-ORDER extractor: ExtractContext gains `prior_artifacts` (the pipeline threads each extractor's outputs to later ones via dataclasses.replace; registration order is load-bearing — ProcessPathExtractor runs last in default_extractors()). The pipeline also materializes the analyzed repo's `.kb/` directory (new git.repo.iter_files_under_at + an extra_files channel kept OUT of the Python file map, so root discovery / parsing / diff reuse never see non-Python paths). PathEngine (the first shipped increment of the DESIGN §11 seam) BFS-slices the call_edge graph: visited-on-enqueue (cycle-safe, O(V+E), shortest path, deterministic tie-breaks via sorted adjacency), paths terminate at the first sink-bearing node, 0-hop supported (a handler with a direct sink grounds a single-span artifact, the direct-recursion precedent), depth/path caps retro-flag `limitations` — never silent. Sinks are typically third-party calls that the precision-first call_edge extractor skips, so sink detection is a separate textual scan against a registry, reusing calls.py's exact attribution machinery (_caller_scan_root lifted to module level) so edges and sinks always agree. Registry = built-in defaults (collision-averse) merged with an optional `.kb/sinks.yaml` override (strictly validated, loud on malformed input); the effective registry's digest is folded into framework_versions so a registry edit can never serve a stale payload (the identity-v2 discipline). Found paths are exact (confidence 1.0); incompleteness is an explicit payload fact, priced into the future labeled artifact. One `process_path` per (entrypoint, sink, terminal), grounded on EVERY span along the path (roles entrypoint/step/terminal — multi-file provenance, the thesis apex: the gate proves ONE artifact spanning THREE files). pyyaml moves to runtime deps; summarize/embed_text gain process_path branches. New Tier-1 HARD gate tier1_processes_test (hand-labeled oracle): 2-hop cross-file chain via the override-only sink, 0-hop built-in sink, event-handler entrypoint, reachable cycle no-hang, no-sink route yields nothing, override chain proven end-to-end, re-index determinism, depth caps. Headline HARD gates: thirteen -> fourteen. 102 eval tests pass. --- CHANGELOG.md | 22 ++ DESIGN.md | 14 +- README.md | 20 +- pyproject.toml | 4 +- src/kb/daemon/pipeline.py | 19 +- src/kb/daemon/watch.py | 9 +- src/kb/embed/text.py | 12 + src/kb/eval/tier1_processes_test.py | 231 +++++++++++++++ src/kb/extract/base.py | 5 + src/kb/extract/deterministic/calls.py | 23 +- src/kb/extract/deterministic/paths.py | 397 ++++++++++++++++++++++++++ src/kb/git/repo.py | 33 +++ src/kb/mcp/records.py | 6 + uv.lock | 4 +- 14 files changed, 767 insertions(+), 32 deletions(-) create mode 100644 src/kb/eval/tier1_processes_test.py create mode 100644 src/kb/extract/deterministic/paths.py diff --git a/CHANGELOG.md b/CHANGELOG.md index fb462b6..95f1cbc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,28 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- **Deterministic business-process path extractor** (`kb.extract.deterministic.paths`): the first + SECOND-ORDER extractor — `ExtractContext` gained `prior_artifacts` (the pipeline feeds each + extractor's outputs to later ones; registration order is load-bearing) and the pipeline now + materializes the analyzed repo's `.kb/` directory alongside the Python tree. `PathEngine` (the + first shipped increment of the DESIGN §11 seam) BFS-slices the `call_edge` graph from extracted + entrypoints (`api_route` / `event_handler` handlers) to functions containing SINK calls — matched + textually against a built-in registry (db/http/email/subprocess/file/queue) merged with an + optional per-repo `.kb/sinks.yaml` override (strictly validated; the effective registry's digest + is folded into artifact identity). One `process_path` per (entrypoint, sink, terminal): shortest + chain, cycle-safe, 0-hop supported, depth/path caps retro-flagged in `limitations` (never + silent), grounded on EVERY span along the path (roles `entrypoint`/`step`/`terminal` — + multi-file provenance). Found paths are exact (`confidence` 1.0); incompleteness is an explicit + payload fact. `pyyaml` moved to runtime deps; `summarize`/`embed_text` gained `process_path` + branches. The LLM labeler is the next slice. +- **Tier-1 process-paths HARD gate** (`kb.eval.tier1_processes_test`): hand-labeled oracle (2-hop + cross-file chain, 0-hop direct sink, event-handler entrypoint), the flagship artifact grounded + across **three files**, cycle no-hang, `.kb/sinks.yaml` override proven end-to-end, no-sink route + yields nothing, depth caps honored, re-index determinism. Headline HARD gates: thirteen → + **fourteen**. + +### Added + - **Call-graph edge extractor** (`kb.extract.deterministic.calls`): the sixth deterministic extractor — one `call_edge` artifact per RESOLVED caller→callee pair (`call:{caller}->{callee}`, call-site lines aggregated in the payload, the import-edge precedent). Three deterministic diff --git a/DESIGN.md b/DESIGN.md index 523dea6..063e6bf 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -338,6 +338,13 @@ rejected. **Verbalized LLM confidence is never used as the score.** > still validate against code spans, so provenance stays code-grounded). The gate is deterministic, > so `semantic_grounding_test` enforces it in CI (stub LLM, no API key), including an adversarial > fabricated claim that must be dropped — on the artifact, module, and package paths. +> +> *Implemented (deterministic process slice):* `process_path` artifacts satisfy the floor **by +> construction** — every sink claim IS a registry match on the materialized path and every endpoint +> IS an extracted entrypoint. Their `confidence` is 1.0: a found path is machine-checked exact (the +> `call_edge` precedent); the unknown-unknowns of the bounded-recall call graph are priced into the +> FUTURE LLM-labeled process artifact (confidence < 1.0 there), with incompleteness carried as +> explicit payload facts here, never a fudged scalar. --- @@ -363,7 +370,7 @@ freshness(current|stale@sha)`, with a deterministic tie-break for reproducible e | Module | Responsibility | Key tech | |--------|----------------|----------| | `kb.structural` | Parse Python without executing it; enumerate symbols/imports/call-sites with per-SHA byte/line ranges; compute content-addressed span identity; incremental reparse. Hidden behind a `StructuralIndex`/`PathEngine` interface so a SCIP backend can replace tree-sitter later. | tree-sitter + tree-sitter-python (canonical bindings) | -| `kb.extract.deterministic` | No-LLM extractors → exact artifacts (confidence=1.0): import graph; FastAPI API contract (static, cross-file grounded); domain entities (pydantic/dataclass/SQLAlchemy, static, cross-file links to referenced entities, hand-labeled gate); library public-API surface (static tree-sitter, cross-file `__init__` re-export resolution, independent griffe-oracle gate); event handlers (pydantic validators / FastAPI `on_event` / SQLAlchemy `listens_for`, static, cross-file target grounding, hand-labeled gate); call-graph edges (per-edge artifacts, three resolution tiers, caller+callee span grounding, hand-labeled gate). | grimp, tree-sitter queries; griffe (dev-only oracle) | +| `kb.extract.deterministic` | No-LLM extractors → exact artifacts (confidence=1.0): import graph; FastAPI API contract (static, cross-file grounded); domain entities (pydantic/dataclass/SQLAlchemy, static, cross-file links to referenced entities, hand-labeled gate); library public-API surface (static tree-sitter, cross-file `__init__` re-export resolution, independent griffe-oracle gate); event handlers (pydantic validators / FastAPI `on_event` / SQLAlchemy `listens_for`, static, cross-file target grounding, hand-labeled gate); call-graph edges (per-edge artifacts, three resolution tiers, caller+callee span grounding, hand-labeled gate); process paths (`paths.PathEngine` — the first shipped increment of the PathEngine seam — BFS to sink-registry matches, multi-file grounded, registry digest identity-bearing). | grimp, tree-sitter queries; griffe (dev-only oracle) | | `kb.introspect` | Eval-only runtime oracle: runs a FastAPI app in a network-blocked sandbox and emits `app.openapi()` for the Tier-1 API gate. Never on the index path. | subprocess sandbox, fastapi | | `kb.embed` | Replaceable embedding adapters + snapshot population for `search_knowledge`. Torch isolated behind the `embed` extra and a lazy import. | sentence-transformers (default), OpenAI (optional), pgvector | | `kb.rag` | Frozen pgvector RAG-over-source baseline — the "other arm" of the knowledge-vs-RAG A/B (no provenance/grounding). | deterministic line-window chunker, pgvector | @@ -453,8 +460,9 @@ Review fact-checked these against current (2026) sources. Caveats are first-clas `event.listen` deferred). 2. The **one** grounded business-process extractor (named real path + labeler + validator + deterministic sub-property gate). *Deterministic foundation shipped:* the `call_edge` extractor - + Tier-1 calls gate; `PathEngine` slicing, the sink registry, the labeler and the span-binding - validator are next. + + Tier-1 calls gate, and now the `process_path` builder (PathEngine BFS + sink registry + + `.kb/sinks.yaml` override, multi-file grounded paths, Tier-1 processes gate). The LLM labeler + and the span-binding validator are next. 3. Recursive invalidation (`artifact_depends_on`), multi-branch dedup, freshness precompute. 4. Embeddings + `search_knowledge` *(shipped v0.2)*; then `pg_search`/BM25 + RRF if vector ranking is insufficient. 5. ADR mining from git/PR history. diff --git a/README.md b/README.md index 88fcc72..8204c98 100644 --- a/README.md +++ b/README.md @@ -84,21 +84,21 @@ flowchart LR ## Status -**v0.5 — spine (artifact-identity rule v2), six deterministic extractors (cross-file grounding, incl. the library public-API surface, event handlers, and call-graph edges), LLM-grounded descriptions up to per-package architecture overviews, live incremental re-index (`kb watch`), MCP serving, the thirteen knowledge gates, and published Docker images.** Everything here grounds what it claims, and nothing it cannot: +**v0.5 — spine (artifact-identity rule v2), six deterministic extractors (cross-file grounding, incl. the library public-API surface, event handlers, and call-graph edges) plus the grounded process-path builder over them, LLM-grounded descriptions up to per-package architecture overviews, live incremental re-index (`kb watch`), MCP serving, the fourteen knowledge gates, and published Docker images.** Everything here grounds what it claims, and nothing it cannot: - **Provenance spine** — content-addressed `span_id` (LOCKED); tree-sitter spans with a normalized S-expression fingerprint and per-SHA location; a single-Postgres, Alembic-managed store with content-addressed idempotent writes; the ≥ 1 `derived_from` anti-hallucination invariant enforced in-app *and* by a deferred DB trigger; pygit2 git ingest (no checkout) with a diff-based invalidation seed. -- **Deterministic extractors** — the **import / dependency graph** (grimp resolves the edge, tree-sitter grounds it on the exact import statement, with an honest `approximate` fallback for re-exports / relative / unmappable imports — never a silent loss); the **FastAPI API-contract** extractor, which grounds a single route **across files** (handler in `routes.py` + `response_model` class in `schemas.py`); the **domain-entity** extractor (pydantic / dataclass / SQLAlchemy classes and their fields, grounded on the class definition **and cross-file on the entities they reference** — purely static, with documented detection limits); the **library public-API-surface** extractor (what a package exposes from its `__init__.py` — `__all__`-authoritative, with `__init__` re-exports resolved **cross-file** to the defining function/class — validated against an independent **griffe** static oracle); the **event-handler** extractor (pydantic `@field_validator`/`@model_validator`, FastAPI `@app.on_event`, SQLAlchemy `@event.listens_for` — one artifact per handler, grounded on the handler span and **cross-file** on the class it listens to; the call-form `event.listen(...)` and dynamic names are documented gaps); and the **call-graph edge** extractor (deterministic caller→callee edges — same-module, imported **cross-file**, and `self.` method calls — one `call_edge` per resolved pair with call-site lines aggregated; `obj.method(...)`, `getattr`, inherited self-calls and decorator/default-arg expressions are documented gaps; the deterministic foundation under the future business-process extractor). +- **Deterministic extractors** — the **import / dependency graph** (grimp resolves the edge, tree-sitter grounds it on the exact import statement, with an honest `approximate` fallback for re-exports / relative / unmappable imports — never a silent loss); the **FastAPI API-contract** extractor, which grounds a single route **across files** (handler in `routes.py` + `response_model` class in `schemas.py`); the **domain-entity** extractor (pydantic / dataclass / SQLAlchemy classes and their fields, grounded on the class definition **and cross-file on the entities they reference** — purely static, with documented detection limits); the **library public-API-surface** extractor (what a package exposes from its `__init__.py` — `__all__`-authoritative, with `__init__` re-exports resolved **cross-file** to the defining function/class — validated against an independent **griffe** static oracle); the **event-handler** extractor (pydantic `@field_validator`/`@model_validator`, FastAPI `@app.on_event`, SQLAlchemy `@event.listens_for` — one artifact per handler, grounded on the handler span and **cross-file** on the class it listens to; the call-form `event.listen(...)` and dynamic names are documented gaps); and the **call-graph edge** extractor (deterministic caller→callee edges — same-module, imported **cross-file**, and `self.` method calls — one `call_edge` per resolved pair with call-site lines aggregated; `obj.method(...)`, `getattr`, inherited self-calls and decorator/default-arg expressions are documented gaps; the deterministic foundation under the future business-process extractor); and the **process-path builder** (second-order: BFS over the call edges from extracted entrypoints — route/event handlers — to sink-registry matches, one `process_path` grounded on EVERY span of the chain, multi-file provenance; built-in sink registry + per-repo `.kb/sinks.yaml` override). - **`kb introspect`** — a sandboxed, network-blocked `app.openapi()` oracle, eval-only and never on the index path, that the API gate scores the static contract against. - **Read-only MCP server** — `find_provenance`, `get_knowledge`, and `search_knowledge`, each returning provenance-carrying units (method + confidence + freshness). - **pgvector embeddings + semantic search** — a replaceable embedding provider (sentence-transformers by default, OpenAI optional) populated by a separate `kb embed` pass; torch stays out of the index path. - **A frozen RAG-over-source baseline** and the **Tier-3 knowledge-vs-RAG recall gate** — the honest A/B that backs the "knowledge > RAG" thesis. - **LLM-grounded descriptions** — an optional, key-gated `kb describe` pass has an LLM write NL summaries for routes, entities, modules (per file), and **packages** (a per-package architecture overview that synthesizes the import graph + public surface + member-module summaries, grounded on the package's own and its direct submodules' spans); every claim is validated against the target's own spans by a deterministic sub-property gate, so ungrounded claims are *dropped* (the anti-hallucination invariant, with a model in the loop). Stored as `extraction_method = "llm_grounded"`, grounded on code spans. - **Incremental re-index** — `kb index --parent ` reuses unchanged files' spans from the parent snapshot and parses only the diff; extraction stays full, so the result is identical to a full re-index (a HARD gate proves it). Auto-detects the parent; falls back to full when none is indexed. -- **Thirteen HARD CI eval gates** (see [Development](#development)). +- **Fourteen HARD CI eval gates** (see [Development](#development)). - **A nightly LLM-judged A/B** (optional, key-gated, **non-gating**) — an answerer LLM answers each question from knowbase's grounded context vs a RAG-over-source context, and a judge LLM scores **answer accuracy** (against hand-written gold) + **hallucination**. Tracked metrics on top of recall; it never blocks CI. -**Not done yet** (and deliberately not faked): ADR mining from git history, grounded business-process extraction, and languages beyond Python. See the [Roadmap](#roadmap). +**Not done yet** (and deliberately not faked): ADR mining from git history, the LLM labeling of business-process paths, and languages beyond Python. See the [Roadmap](#roadmap). ## Quickstart @@ -120,7 +120,7 @@ The base `--extra dev` install stays torch-free; the `embed` extra pulls sentenc ### Run the gates ```bash -uv run pytest src/kb/eval -q # the thirteen HARD gates (spins an ephemeral local Postgres) +uv run pytest src/kb/eval -q # the fourteen HARD gates (spins an ephemeral local Postgres) ``` ### Index a commit @@ -234,6 +234,7 @@ A Python package `kb` (uv, src-layout). Modules and their responsibilities: | `kb.extract.deterministic.entities` | Static domain-entity extractor — pydantic / dataclass / SQLAlchemy classes + their fields, grounded on the class definition **and, across files, on the entities they reference** (field types + `relationship()`); detection signals and limits recorded in the payload. | | `kb.extract.deterministic.library_surface` | Static library public-API-surface extractor — one `public_symbol` per name a package exposes from its `__init__.py` (`__all__`-authoritative), `__init__` re-exports resolved **cross-file** to the defining function/class; validated by an independent griffe static oracle. Never imports user code. | | `kb.extract.deterministic.calls` | Static call-graph edge extractor — one `call_edge` per resolved caller→callee pair (same-module / imported **cross-file** / `self.` method), call-site lines aggregated; only resolved first-party edges are emitted; dynamic/attribute/inherited calls are documented gaps. Never imports user code. | +| `kb.extract.deterministic.paths` | Second-order process-path builder — `process_path` per (entrypoint, sink, terminal): shortest chain of resolved call edges from an extracted route/event handler to a sink-registry match, grounded on every span along the path (multi-file); built-in registry + `.kb/sinks.yaml` override (registry digest is identity-bearing); caps flagged, never silent. | | `kb.extract.deterministic.events` | Static event-handler extractor — one `event_handler` per handler carrying pydantic validator / FastAPI `on_event` / SQLAlchemy `listens_for` decorator registrations (stacked decorators in `payload.registrations`), grounded on the handler span **and, across files, on the class it listens to**; call-form `event.listen(...)` and dynamic names are documented gaps. | | `kb.introspect` | Sandboxed, network-blocked `app.openapi()` oracle — eval-only ground truth for the API gate, never on the index path. | | `kb.mcp` | Read-only MCP server and its provenance-carrying records: `find_provenance`, `get_knowledge`, `search_knowledge`. | @@ -241,7 +242,7 @@ A Python package `kb` (uv, src-layout). Modules and their responsibilities: | `kb.rag` | The frozen pgvector RAG-over-source baseline — the "other arm" of the knowledge-vs-RAG A/B (no provenance, no grounding). | | `kb.extract.semantic` | LLM-grounded extraction (`kb describe`): NL descriptions of routes/entities/modules **and per-package architecture overviews** (rich synthesis from the import graph + public surface + member summaries, grounded on code spans) with a deterministic sub-property gate (`grounding.validate_claims`) that drops any claim not backed by the target's spans. Separate key-gated pass; never on `index`. | | `kb.daemon.cli` | The `kb` CLI: `index` (full or `--incremental`/`--parent`), `watch` (poll a local branch ref, per-commit incremental catch-up, `branch_ref` cursor), `migrate`, `embed`, `describe`, `serve` (MCP), and `introspect` — all functional. | -| `kb.eval` | Thirteen HARD CI gates (identity reproducibility, adversarial grounding, Tier-1 import oracle, Tier-1 API oracle, Tier-1 entities oracle, Tier-1 library-surface oracle, Tier-1 events oracle, Tier-1 call-graph oracle, Tier-3 knowledge-vs-RAG recall, Tier-4 one-hop invalidation, invariants, semantic grounding floor, incremental re-index equivalence) plus the supporting MCP / embed / store suite. | +| `kb.eval` | Fourteen HARD CI gates (identity reproducibility, adversarial grounding, Tier-1 import oracle, Tier-1 API oracle, Tier-1 entities oracle, Tier-1 library-surface oracle, Tier-1 events oracle, Tier-1 call-graph oracle, Tier-1 process-paths oracle, Tier-3 knowledge-vs-RAG recall, Tier-4 one-hop invalidation, invariants, semantic grounding floor, incremental re-index equivalence) plus the supporting MCP / embed / store suite. | Core tables: `commit_ref`, `branch_ref`, `code_span`, `span_occurrence`, `artifact` (now with `embedding vector(384)` + `embedding_model_id`), `artifact_derived_from`, `snapshot_entry`, and `rag_chunk` (the baseline arm). @@ -251,10 +252,10 @@ Core tables: `commit_ref`, `branch_ref`, `code_span`, `span_occurrence`, `artifa uv sync --extra dev # venv + install uv run ruff check src/kb # lint uv run mypy # strict type-check -uv run pytest src/kb/eval -q # the thirteen HARD eval gates +uv run pytest src/kb/eval -q # the fourteen HARD eval gates ``` -CI (GitHub Actions, workflow **"CI"**, `.github/workflows/ci.yml`) runs ruff, `mypy --strict`, and the eval gates against a `pgvector/pgvector:pg17` service (with the embedding model cached). The **thirteen HARD gates** that block a merge: +CI (GitHub Actions, workflow **"CI"**, `.github/workflows/ci.yml`) runs ruff, `mypy --strict`, and the eval gates against a `pgvector/pgvector:pg17` service (with the embedding model cached). The **fourteen HARD gates** that block a merge: 1. **Identity reproducibility** — formatting / comment / docstring / location changes must NOT change `span_id`; a rename MUST. Pure identity core, no database. 2. **Adversarial grounding** — an ungrounded artifact is rejected by *both* layers (the app's `GroundingError` and the DB's deferred `artifact_grounded_check` trigger); a genuinely grounded artifact commits cleanly. @@ -269,6 +270,7 @@ CI (GitHub Actions, workflow **"CI"**, `.github/workflows/ci.yml`) runs ruff, `m 11. **Invariants** — zero orphans (every snapshot artifact is grounded), and re-indexing the same SHA yields the identical set of artifact ids. 12. **Semantic grounding floor** — the LLM-grounded describer's claims are validated against the artifact's own spans by a deterministic sub-property gate; an adversarial fabricated claim is *dropped*, never stored (run on a stub LLM, so it gates without an API key). 13. **Incremental re-index equivalence** — an incremental re-index (reuse unchanged files' spans from the parent, parse only the diff) yields the *identical* `{logical_key: artifact_id}` snapshot as a full re-index of the same tree, and the parse is provably skipped for unchanged files (counter assertions); a missing/unindexed parent falls back to full. +14. **Tier-1 process-paths oracle** — materialized business-process paths match a hand-labeled oracle (2-hop cross-file chain, 0-hop direct-sink handler, event-handler entrypoint); the flagship artifact is grounded across **three files**; a reachable call cycle cannot hang the BFS; the fixture's own `.kb/sinks.yaml` proves the override chain; a no-sink route yields no artifact; depth caps are honored. The identity rules in `kb.ids` (and `kb.structural`) are **LOCKED**: changing one is a breaking change, gated behind a `NORMALIZATION_VERSION` / `extractor_version` bump so existing digests are invalidated rather than silently colliding. @@ -298,7 +300,7 @@ Next milestones: - [x] **LLM-grounded semantic layer** — model-backed artifacts that still carry ≥ 1 span (`extraction_method = "llm_grounded"`): `kb describe` writes span-validated descriptions for routes, entities, modules, and per-package architecture overviews. *(shipped)* - [x] **Incremental re-index on git push** — `kb index --incremental`/`--parent` reuses unchanged files' spans from the parent snapshot (extraction stays full; equivalence is gated), and `kb watch` polls a local branch ref, indexing each new first-parent commit incrementally with a `branch_ref` cursor (`--max-catchup` guard, `--once` for cron/CI). *(shipped)* - [ ] **ADR mining** from git / PR history. -- [ ] **Grounded business-process extraction.** +- [~] **Grounded business-process extraction** — *deterministic path extraction shipped*: `process_path` artifacts (entrypoint → call chain → sink-registry match, multi-file grounding). LLM labeling + span-binding validator are the remaining piece. - [ ] **More languages** beyond Python. ## Contributing diff --git a/pyproject.toml b/pyproject.toml index 73b6f60..483b4e5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -19,12 +19,12 @@ dependencies = [ "sqlparse>=0.5.5", "fastmcp>=3.3,<4", "pgvector>=0.4,<0.5", + "pyyaml>=6", ] [project.optional-dependencies] dev = [ "pytest>=8", - "pyyaml>=6", "ruff>=0.6", "mypy>=1.10", # fastapi powers the eval oracle (kb introspect) and the Tier-1 API gate only; the serving @@ -96,7 +96,7 @@ ignore_errors = true [[tool.mypy.overrides]] module = [ "pygit2.*", "grimp.*", "sqlparse.*", "fastapi.*", "starlette.*", "fastmcp.*", "mcp.*", - "sentence_transformers.*", "pgvector.*", "openai.*", "anthropic.*", + "sentence_transformers.*", "pgvector.*", "openai.*", "anthropic.*", "yaml.*", ] ignore_missing_imports = true diff --git a/src/kb/daemon/pipeline.py b/src/kb/daemon/pipeline.py index 3b8efdd..11a507e 100644 --- a/src/kb/daemon/pipeline.py +++ b/src/kb/daemon/pipeline.py @@ -11,13 +11,13 @@ import tempfile from collections.abc import Sequence -from dataclasses import dataclass +from dataclasses import dataclass, replace from pathlib import Path import pygit2 from sqlalchemy import Connection, Engine -from kb.extract.base import ExtractContext, Extractor +from kb.extract.base import ExtractContext, ExtractedArtifact, Extractor from kb.git import repo as gitrepo from kb.git.diff import changed_paths from kb.store.queries import is_sha_indexed, reusable_spans @@ -67,6 +67,7 @@ def index_commit( repo = gitrepo.open_repo(repo_path) sha = gitrepo.commit_sha(repo, rev) files = dict(gitrepo.iter_python_files_at(repo, sha)) + extra_files = dict(gitrepo.iter_files_under_at(repo, sha, ".kb/")) root = ( first_party_root if first_party_root is not None @@ -134,7 +135,7 @@ def index_commit( artifacts = 0 if extractors: artifacts = _run_extractors( - conn, extractors, files, sha, root, spans_by_module, path_by_module + conn, extractors, files, extra_files, sha, root, spans_by_module, path_by_module ) return IndexResult( @@ -219,6 +220,7 @@ def _run_extractors( conn: Connection, extractors: Sequence[Extractor], files: dict[str, bytes], + extra_files: dict[str, bytes], sha: str, first_party_root: str, spans_by_module: dict[str, list[ParsedSpan]], @@ -226,20 +228,25 @@ def _run_extractors( ) -> int: written = 0 with tempfile.TemporaryDirectory(prefix="kb-snapshot-") as tmp: - for path, content in files.items(): + for path, content in {**files, **extra_files}.items(): target = Path(tmp) / path target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(content) - ctx = ExtractContext( + base_ctx = ExtractContext( sha=sha, materialized_root=tmp, first_party_root=first_party_root, spans_by_module=spans_by_module, path_by_module=path_by_module, ) + prior: list[ExtractedArtifact] = [] for extractor in extractors: - for artifact in extractor.extract(ctx): + # second-order extractors see EARLIER extractors' outputs (order is load-bearing) + ctx = replace(base_ctx, prior_artifacts=tuple(prior)) + produced = extractor.extract(ctx) + for artifact in produced: artifact_id = write_grounded_artifact(conn, artifact) write_snapshot_entry(conn, sha, artifact.logical_key, artifact_id) written += 1 + prior.extend(produced) return written diff --git a/src/kb/daemon/watch.py b/src/kb/daemon/watch.py index e9c23fd..36d0e43 100644 --- a/src/kb/daemon/watch.py +++ b/src/kb/daemon/watch.py @@ -32,13 +32,19 @@ from kb.extract.deterministic.fastapi_contract import FastAPIExtractor from kb.extract.deterministic.imports import ImportExtractor from kb.extract.deterministic.library_surface import LibrarySurfaceExtractor +from kb.extract.deterministic.paths import ProcessPathExtractor from kb.git import repo as gitrepo from kb.store.queries import branch_head, is_sha_indexed from kb.store.writer import upsert_branch_ref def default_extractors() -> list[Extractor]: - """The standard extractor set shared by ``kb index`` and ``kb watch``.""" + """The standard extractor set shared by ``kb index`` and ``kb watch``. + + Order is load-bearing: the pipeline feeds each extractor's outputs to LATER ones via + ``ExtractContext.prior_artifacts``; ProcessPathExtractor is second-order (consumes + api_route + event_handler + call_edge) and must stay LAST. + """ return [ ImportExtractor(), FastAPIExtractor(), @@ -46,6 +52,7 @@ def default_extractors() -> list[Extractor]: EventExtractor(), LibrarySurfaceExtractor(), CallGraphExtractor(), + ProcessPathExtractor(), ] diff --git a/src/kb/embed/text.py b/src/kb/embed/text.py index 92efd0e..0b9fbc2 100644 --- a/src/kb/embed/text.py +++ b/src/kb/embed/text.py @@ -57,6 +57,18 @@ def embed_text(kind: str, payload: dict[str, Any]) -> str: "fields " + " ".join(str(f) for r in regs for f in r.get("fields", [])), ] return " ".join(p for p in parts if p.strip()) + if kind == "process_path": + sink = payload.get("sink", {}) + parts = [ + head, + f"process {payload.get('entrypoint', '')}", + f"route {payload.get('entrypoint_reference', '')}", + "steps " + " ".join(str(s) for s in payload.get("steps", [])), + f"sink {sink.get('name', '')} " + + " ".join(str(m.get("text", "")) for m in sink.get("matches", [])), + f"terminal {payload.get('terminal', '')}", + ] + return " ".join(p for p in parts if p.strip()) if kind == "call_edge": parts = [ head, diff --git a/src/kb/eval/tier1_processes_test.py b/src/kb/eval/tier1_processes_test.py new file mode 100644 index 0000000..47b1429 --- /dev/null +++ b/src/kb/eval/tier1_processes_test.py @@ -0,0 +1,231 @@ +"""HARD GATE — Tier 1: deterministic business-process paths vs a hand-labeled oracle (§9, §14). + +The DESIGN §9 hard floor holds by construction: every sink claim IS a registry match on the +materialized path, every endpoint IS an extracted entrypoint. The fixture ships its own +``.kb/sinks.yaml`` override, proving the whole git-blob → pipeline materialization → registry-merge +chain; a reachable call cycle proves the BFS cannot hang; the flagship path grounds ONE artifact +across THREE files — the thesis apex. +""" + +from __future__ import annotations + +from pathlib import Path + +from sqlalchemy import Engine, func, select + +from kb.daemon.pipeline import index_commit +from kb.eval._fixtures import make_git_repo +from kb.extract.deterministic.calls import CallGraphExtractor +from kb.extract.deterministic.events import EventExtractor +from kb.extract.deterministic.fastapi_contract import FastAPIExtractor +from kb.extract.deterministic.paths import _BUILTIN_SINKS, ProcessPathExtractor +from kb.store import models as m +from kb.store.queries import provenance_for_artifact + +FILES = { + ".kb/sinks.yaml": ( + "version: 1\nextend: true\nsinks:\n" + ' - name: ledger_write\n patterns: ["*.record_ledger"]\n' + ), + "src/bp/__init__.py": "", + "src/bp/api.py": ( + "from fastapi import APIRouter\n" + "import requests\n" + "from bp.svc import place, lookup\n\n" + "router = APIRouter()\n\n\n" + "@router.post('/orders')\n" + "def create_order():\n return place()\n\n\n" + "@router.get('/orders')\n" + "def list_orders():\n return lookup()\n\n\n" + "@router.post('/ping')\n" + "def ping():\n return requests.post('http://x')\n" + ), + "src/bp/svc.py": ( + "from bp.repo import save_order\n\n\n" + "def place():\n spin()\n return save_order()\n\n\n" + "def lookup():\n return 1\n\n\n" + "def spin():\n return unspin()\n\n\n" + "def unspin():\n return spin()\n" + ), + "src/bp/repo.py": ( + "from bp.ledger import ledger\n\n\n" + "def save_order():\n return ledger.record_ledger('order')\n" + ), + "src/bp/ledger.py": ( + "class Ledger:\n def record_ledger(self, kind):\n return kind\n\n\n" + "ledger = Ledger()\n" + ), + "src/bp/boot.py": ( + "from fastapi import FastAPI\n" + "from bp.repo import save_order\n\n" + "app = FastAPI()\n\n\n" + "@app.on_event('startup')\n" + "def warmup():\n return save_order()\n" + ), +} + +# Hand-labeled oracle: (entrypoint, kind, reference, steps, sink_name, terminal, depth). +EXPECTED_PATHS = { + ("bp.api.create_order", "api_route", "POST /orders", + ("bp.api.create_order", "bp.svc.place", "bp.repo.save_order"), + "ledger_write", "bp.repo.save_order", 2), + ("bp.api.ping", "api_route", "POST /ping", + ("bp.api.ping",), "http_call", "bp.api.ping", 0), + ("bp.boot.warmup", "event_handler", "fastapi_on_event:startup", + ("bp.boot.warmup", "bp.repo.save_order"), + "ledger_write", "bp.repo.save_order", 1), +} +NO_PATH_ENTRYPOINTS = {"bp.api.list_orders"} # its call chain reaches no sink + + +def _extractors(**kw: int) -> list: + # ORDER is load-bearing: ProcessPathExtractor is second-order and must run last. + return [FastAPIExtractor(), EventExtractor(), CallGraphExtractor(), + ProcessPathExtractor(**kw)] + + +def _index(engine: Engine, tmp_path: Path, **kw: int) -> str: + sha = make_git_repo(tmp_path, [FILES])[0] + index_commit(engine, str(tmp_path), sha, extractors=_extractors(**kw), + first_party_root="src") + return sha + + +def _payloads(engine: Engine, sha: str) -> list[dict]: + join = m.snapshot_entry.join( + m.artifact, m.artifact.c.artifact_id == m.snapshot_entry.c.artifact_id + ) + with engine.connect() as conn: + return list( + conn.execute( + select(m.artifact.c.payload) + .select_from(join) + .where(m.snapshot_entry.c.sha == sha, m.artifact.c.kind == "process_path") + ).scalars() + ) + + +def test_process_paths_match_oracle(engine: Engine, tmp_path: Path) -> None: + sha = _index(engine, tmp_path) # completing at all proves the spin<->unspin cycle can't hang + extracted = { + (p["entrypoint"], p["entrypoint_kind"], p["entrypoint_reference"], + tuple(p["steps"]), p["sink"]["name"], p["terminal"], p["depth"]) + for p in _payloads(engine, sha) + } + assert extracted == EXPECTED_PATHS + assert all(p["entrypoint"] not in NO_PATH_ENTRYPOINTS for p in _payloads(engine, sha)) + + +def test_multi_file_provenance_thesis(engine: Engine, tmp_path: Path) -> None: + """The apex assertion: ONE artifact grounded across THREE files.""" + sha = _index(engine, tmp_path) + key = "process:bp.api.create_order->ledger_write@bp.repo.save_order" + with engine.connect() as conn: + prov = {(p.file_path, p.role) for p in provenance_for_artifact(conn, sha, key)} + assert prov == { + ("src/bp/api.py", "entrypoint"), + ("src/bp/svc.py", "step"), + ("src/bp/repo.py", "terminal"), + } + + +def test_flagship_payload_shape(engine: Engine, tmp_path: Path) -> None: + sha = _index(engine, tmp_path) + flagship = next( + p for p in _payloads(engine, sha) if p["entrypoint"] == "bp.api.create_order" + ) + assert flagship["edges"] == [ + {"caller": "bp.api.create_order", "callee": "bp.svc.place", "resolution": "imported"}, + {"caller": "bp.svc.place", "callee": "bp.repo.save_order", "resolution": "imported"}, + ] + assert flagship["sink"]["matches"] == [ + {"text": "ledger.record_ledger", "pattern": "*.record_ledger", "line": 5} + ] + assert flagship["span_mapping"] == "exact" + assert flagship["limitations"] == [] + join = m.snapshot_entry.join( + m.artifact, m.artifact.c.artifact_id == m.snapshot_entry.c.artifact_id + ) + with engine.connect() as conn: + fw = conn.execute( + select(m.artifact.c.framework_versions) + .select_from(join) + .where( + m.snapshot_entry.c.sha == sha, + m.snapshot_entry.c.logical_key + == "process:bp.api.create_order->ledger_write@bp.repo.save_order", + ) + ).scalar() + assert fw and "sink_registry" in fw # the effective registry is identity-bearing + + +def test_zero_hop_path(engine: Engine, tmp_path: Path) -> None: + sha = _index(engine, tmp_path) + with engine.connect() as conn: + prov = provenance_for_artifact(conn, sha, "process:bp.api.ping->http_call@bp.api.ping") + assert len(prov) == 1 # entrypoint == terminal span: one grounding row ... + assert prov[0].role == "entrypoint" # ... keeping the entrypoint role + + +def test_override_registry_is_live(engine: Engine, tmp_path: Path) -> None: + assert all(r.name != "ledger_write" for r in _BUILTIN_SINKS) # override-only sink ... + sha = _index(engine, tmp_path) + sinks = {p["sink"]["name"] for p in _payloads(engine, sha)} + assert "ledger_write" in sinks # ... proves .kb/sinks.yaml -> materialization -> merge + + +def test_every_process_path_is_grounded(engine: Engine, tmp_path: Path) -> None: + sha = _index(engine, tmp_path) + join = m.snapshot_entry.outerjoin( + m.artifact_derived_from, + m.artifact_derived_from.c.artifact_id == m.snapshot_entry.c.artifact_id, + ).join(m.artifact, m.artifact.c.artifact_id == m.snapshot_entry.c.artifact_id) + with engine.connect() as conn: + ungrounded = conn.execute( + select(func.count()) + .select_from(join) + .where( + m.snapshot_entry.c.sha == sha, + m.artifact.c.kind == "process_path", + m.artifact_derived_from.c.artifact_id.is_(None), + ) + ).scalar() + assert ungrounded == 0 + + +def test_determinism_reindex(engine: Engine, tmp_path: Path) -> None: + sha = _index(engine, tmp_path) + + def ids() -> set[bytes]: + with engine.connect() as conn: + join = m.snapshot_entry.join( + m.artifact, m.artifact.c.artifact_id == m.snapshot_entry.c.artifact_id + ) + return { + bytes(r) + for r in conn.execute( + select(m.snapshot_entry.c.artifact_id) + .select_from(join) + .where( + m.snapshot_entry.c.sha == sha, m.artifact.c.kind == "process_path" + ) + ).scalars() + } + + first = ids() + index_commit(engine, str(tmp_path), sha, extractors=_extractors(), first_party_root="src") + assert ids() == first + assert len(first) == len(EXPECTED_PATHS) + + +def test_depth_cap(engine: Engine, tmp_path: Path) -> None: + # a marker file makes the sha unique: the shared session DB already holds this fixture + # indexed at the DEFAULT depth under the identical-content sha + files = {**FILES, "src/bp/depth_marker.py": "DEPTH_CAP_FIXTURE = 1\n"} + sha = make_git_repo(tmp_path, [files])[0] + index_commit(engine, str(tmp_path), sha, extractors=_extractors(max_depth=1), + first_party_root="src") + entrypoints = {p["entrypoint"] for p in _payloads(engine, sha)} + # the 2-hop flagship exceeds max_depth=1 and is not emitted; 0/1-hop paths survive + assert "bp.api.create_order" not in entrypoints + assert {"bp.api.ping", "bp.boot.warmup"} <= entrypoints diff --git a/src/kb/extract/base.py b/src/kb/extract/base.py index 554d0aa..909f1bb 100644 --- a/src/kb/extract/base.py +++ b/src/kb/extract/base.py @@ -77,3 +77,8 @@ class ExtractContext: first_party_root: str spans_by_module: Mapping[str, Sequence[ParsedSpan]] path_by_module: Mapping[str, str] + # Artifacts produced by extractors that ran EARLIER in this same snapshot pass, in emission + # order — empty for first-order extractors. The pipeline appends each extractor's outputs + # before invoking the next, so registration ORDER is load-bearing for second-order extractors + # (see kb.daemon.watch.default_extractors). + prior_artifacts: Sequence[ExtractedArtifact] = () diff --git a/src/kb/extract/deterministic/calls.py b/src/kb/extract/deterministic/calls.py index 8dae0e0..efc1331 100644 --- a/src/kb/extract/deterministic/calls.py +++ b/src/kb/extract/deterministic/calls.py @@ -211,15 +211,7 @@ def _parse_file(self, ctx: ExtractContext, module: str) -> Node | None: return self._parser.parse(source).root_node def _caller_scan_root(self, span: ParsedSpan) -> Node | None: - """The subtree whose calls belong to this caller: a def's BODY (its own decorators, - parameters and annotations are excluded by construction), or the whole file for a module.""" - root = self._parser.parse(textwrap.dedent(span.raw_text).encode("utf-8")).root_node - if span.span_kind == "module": - return root - deco = _first_child_of_type(root, "decorated_definition") - host = deco if deco is not None else root - fn = _first_child_of_type(host, "function_definition") - return fn.child_by_field_name("body") if fn is not None else None + return _caller_scan_root(self._parser, span) # --- artifact assembly ----------------------------------------------------- @@ -253,6 +245,19 @@ def _build_artifact(self, key: str, edge: _Edge) -> ExtractedArtifact: # --- module-level helpers (kept local; mirror the other deterministic extractors) ---- +def _caller_scan_root(parser: Parser, span: ParsedSpan) -> Node | None: + """The subtree whose calls belong to one caller: a def's BODY (its own decorators, parameters + and annotations are excluded by construction), or the whole file for a module. Shared with the + process-path sink scan so call attribution stays bit-identical across both.""" + root = parser.parse(textwrap.dedent(span.raw_text).encode("utf-8")).root_node + if span.span_kind == "module": + return root + deco = _first_child_of_type(root, "decorated_definition") + host = deco if deco is not None else root + fn = _first_child_of_type(host, "function_definition") + return fn.child_by_field_name("body") if fn is not None else None + + def _iter_calls(node: Node) -> Iterator[Node]: """``call`` nodes lexically owned by ONE caller: stops at nested def/class subtrees (each nested def is its own caller span; descending would double-attribute). Recurses INTO call diff --git a/src/kb/extract/deterministic/paths.py b/src/kb/extract/deterministic/paths.py new file mode 100644 index 0000000..cc06348 --- /dev/null +++ b/src/kb/extract/deterministic/paths.py @@ -0,0 +1,397 @@ +"""Deterministic business-process path extractor (DESIGN.md §4, §9, §14 item 2 — step 2). + +Materializes **named real paths**: an ENTRYPOINT (an already-extracted ``api_route`` / +``event_handler`` handler) → a shortest chain of resolved first-party ``call_edge``s → a TERMINAL +function containing a SINK call (a registry match). One ``process_path`` artifact per +``(entrypoint, sink_name, terminal)``, grounded on EVERY def span along the path (multi-file +provenance). This is a SECOND-ORDER extractor: it consumes ``ctx.prior_artifacts`` and must be +registered AFTER the fastapi/events/calls extractors (see ``kb.daemon.watch.default_extractors``). + +The DESIGN §9 hard floor holds BY CONSTRUCTION: every sink claim IS a registry match on the +materialized path, and every endpoint IS an extracted entrypoint. ``confidence`` is 1.0 — a found +path is machine-checked exact (the ``call_edge`` precedent); incompleteness (paths missed by the +bounded-recall call graph or textual sink matching) is a payload/doc fact, priced into the FUTURE +LLM-labeled artifact, not this one. The effective sink registry's digest is folded into +``framework_versions`` so a registry edit can never silently serve a stale payload (identity-v2 +discipline). + +Sinks are typically third-party calls, which the precision-first ``call_edge`` extractor skips — +so sink detection is a separate textual scan of call sites against the registry, reusing the exact +call-attribution machinery of ``kb.extract.deterministic.calls`` (body-only scan roots, no-descend +traversal), so edges and sinks always agree about which function contains a call. + +Registry: built-in defaults below, plus an optional ``.kb/sinks.yaml`` at the ANALYZED repo's root +(travels with the commit; materialized by the pipeline). Pattern grammar is deliberately minimal: +exact dotted ``A.B``, prefix ``P.*``, suffix ``*.NAME`` (attribute-name match — works even when the +receiver is dynamic). Documented gaps: alias-renamed receivers (``import requests as rq``), +mid-segment globs, argument inspection, class-instantiation edges (``__init__`` side effects are +invisible), sink-bearing intermediates shadow deeper terminals (paths terminate at the first sink). +A malformed override raises ``ValueError`` loudly — never silently ignored. +""" + +from __future__ import annotations + +import hashlib +import json +from collections import deque +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +import tree_sitter_python as tsp +import yaml +from tree_sitter import Language, Parser + +from kb.extract.base import DerivedEdge, ExtractContext, ExtractedArtifact +from kb.extract.deterministic.calls import ( + _caller_scan_root, + _dotted_object_text, + _iter_calls, + _text, +) +from kb.structural.interface import ParsedSpan + +EXTRACTOR_ID = "process_paths" +EXTRACTOR_VERSION = "1" +SINKS_FILE = ".kb/sinks.yaml" +DEFAULT_MAX_DEPTH = 8 +DEFAULT_MAX_PATHS_PER_ENTRYPOINT = 32 + +_LANGUAGE = Language(tsp.language()) + + +@dataclass(frozen=True) +class SinkRule: + name: str + patterns: tuple[str, ...] # exact "A.B" | prefix "P.*" | suffix "*.NAME" + + +# Curation rule: no collision-prone names (*.add, *.write, *.flush ...) — precision-first; +# repos opt into project-specific sinks via the .kb/sinks.yaml override. +_BUILTIN_SINKS: tuple[SinkRule, ...] = ( + SinkRule("db_write", ("*.commit", "*.execute", "*.executemany", "*.bulk_save_objects")), + SinkRule("http_call", ("requests.*", "httpx.*", "urllib.request.*", "aiohttp.*")), + SinkRule("email_send", ("smtplib.*", "*.send_message", "*.sendmail")), + SinkRule("subprocess_exec", ("subprocess.*", "os.system", "os.popen")), + SinkRule("file_write", ("*.write_text", "*.write_bytes", "shutil.*", "os.remove")), + SinkRule("queue_publish", ("*.publish", "*.apply_async", "*.send_task", "*.basic_publish")), +) + + +@dataclass(frozen=True) +class _SinkMatch: + text: str # matched dotted callee text; "?." when the receiver is dynamic + pattern: str + line: int # absolute 1-based + + +@dataclass(frozen=True) +class _Entrypoint: + fq: str + kind: str # "api_route" | "event_handler" + reference: str + references: tuple[tuple[str, str], ...] # all (kind, reference) pairs for this fq + + +@dataclass(frozen=True) +class _FoundPath: + entrypoint: _Entrypoint + steps: tuple[str, ...] + resolutions: tuple[str, ...] + sink_name: str + matches: tuple[_SinkMatch, ...] + limitations: tuple[str, ...] + + +class PathEngine: + """Deterministic shortest-path slicer over first-party call edges (the DESIGN §11 seam).""" + + def __init__( + self, + adjacency: Mapping[str, Sequence[tuple[str, str]]], + sinks_by_fq: Mapping[str, Mapping[str, Sequence[_SinkMatch]]], + *, + max_depth: int, + max_paths_per_entrypoint: int, + ) -> None: + self._adjacency = adjacency + self._sinks_by_fq = sinks_by_fq + self._max_depth = max_depth + self._max_paths = max_paths_per_entrypoint + + def paths_from(self, entrypoint: _Entrypoint) -> list[_FoundPath]: + """BFS with visited-on-enqueue: cycle-safe, O(V+E), first arrival = a shortest path; + sorted adjacency + FIFO order make ties deterministic (lexicographically earliest).""" + found: list[_FoundPath] = [] + limitations: set[str] = set() + visited = {entrypoint.fq} + queue: deque[tuple[str, tuple[str, ...], tuple[str, ...]]] = deque( + [(entrypoint.fq, (entrypoint.fq,), ())] + ) + while queue: + fq, steps, resolutions = queue.popleft() + sink_names = self._sinks_by_fq.get(fq) + if sink_names: + for sink_name in sorted(sink_names): + if len(found) >= self._max_paths: + limitations.add("paths_capped") + break + found.append( + _FoundPath( + entrypoint=entrypoint, + steps=steps, + resolutions=resolutions, + sink_name=sink_name, + matches=tuple( + sorted(sink_names[sink_name], key=lambda m: m.line) + ), + limitations=(), + ) + ) + continue # a path TERMINATES at its first sink-bearing node + if len(steps) - 1 >= self._max_depth: + limitations.add("depth_capped") + continue + for callee, resolution in self._adjacency.get(fq, ()): + if callee not in visited: + visited.add(callee) + queue.append((callee, (*steps, callee), (*resolutions, resolution))) + if limitations: # retro-tag every emitted path — no silent caps + found = [ + _FoundPath( + p.entrypoint, p.steps, p.resolutions, p.sink_name, p.matches, + tuple(sorted(limitations)), + ) + for p in found + ] + return found + + +class ProcessPathExtractor: + extractor_id = EXTRACTOR_ID + extractor_version = EXTRACTOR_VERSION + + def __init__( + self, + *, + max_depth: int = DEFAULT_MAX_DEPTH, + max_paths_per_entrypoint: int = DEFAULT_MAX_PATHS_PER_ENTRYPOINT, + ) -> None: + self._parser = Parser(_LANGUAGE) + self._max_depth = max_depth + self._max_paths = max_paths_per_entrypoint + + def extract(self, ctx: ExtractContext) -> list[ExtractedArtifact]: + rules = _load_registry(ctx.materialized_root) + digest = _registry_digest(rules) + entrypoints = _entrypoints(ctx.prior_artifacts) + if not entrypoints: + return [] + adjacency = _adjacency(ctx.prior_artifacts) + span_index = _def_span_index(ctx) + sinks_by_fq = self._scan_sinks(ctx, rules) + engine = PathEngine( + adjacency, sinks_by_fq, + max_depth=self._max_depth, max_paths_per_entrypoint=self._max_paths, + ) + artifacts: list[ExtractedArtifact] = [] + for entrypoint in entrypoints: + for path in engine.paths_from(entrypoint): + art = self._build_artifact(path, span_index, digest) + if art is not None: + artifacts.append(art) + return artifacts + + def _scan_sinks( + self, ctx: ExtractContext, rules: Sequence[SinkRule] + ) -> dict[str, dict[str, list[_SinkMatch]]]: + out: dict[str, dict[str, list[_SinkMatch]]] = {} + for spans in ctx.spans_by_module.values(): + for span in spans: + if span.span_kind not in ("function", "method"): + continue + root = _caller_scan_root(self._parser, span) + if root is None: + continue + for call in _iter_calls(root): + fn = call.child_by_field_name("function") + if fn is None: + continue + dotted: str | None = None + attr: str | None = None + if fn.type == "identifier": + dotted = _text(fn) + elif fn.type == "attribute": + attr = _text(fn.child_by_field_name("attribute")) + obj_node = fn.child_by_field_name("object") + obj = _dotted_object_text(obj_node) if obj_node is not None else None + dotted = f"{obj}.{attr}" if obj and attr else None + else: + continue # call-of-call etc. -> documented gap + for sink_name, pattern, text in _match_call(rules, dotted, attr): + line = span.start_line + call.start_point[0] + out.setdefault(span.fq_symbol_path, {}).setdefault( + sink_name, [] + ).append(_SinkMatch(text=text, pattern=pattern, line=line)) + return out + + def _build_artifact( + self, path: _FoundPath, span_index: dict[str, ParsedSpan], digest: str + ) -> ExtractedArtifact | None: + spans = [span_index.get(fq) for fq in path.steps] + if any(s is None for s in spans): # cannot happen within one snapshot; guard DB drift + return None + grounding: dict[bytes, DerivedEdge] = {} + for i, span in enumerate(spans): + assert span is not None + role = "entrypoint" if i == 0 else ("terminal" if i == len(spans) - 1 else "step") + grounding.setdefault(span.span_id, DerivedEdge(span.span_id, role)) + ep = path.entrypoint + payload: dict[str, Any] = { + "entrypoint": ep.fq, + "entrypoint_kind": ep.kind, + "entrypoint_reference": ep.reference, + "entrypoint_references": [list(r) for r in ep.references], + "steps": list(path.steps), + "edges": [ + {"caller": a, "callee": b, "resolution": r} + for (a, b), r in zip( + zip(path.steps, path.steps[1:], strict=False), + path.resolutions, strict=True, + ) + ], + "terminal": path.steps[-1], + "sink": { + "name": path.sink_name, + "matches": [ + {"text": m.text, "pattern": m.pattern, "line": m.line} + for m in path.matches + ], + }, + "depth": len(path.steps) - 1, + "span_mapping": "exact", + "completeness": ( + "paths found are exact; missing paths possible " + "(call-edge recall bounds, textual sink matching)" + ), + "limitations": list(path.limitations), + } + return ExtractedArtifact( + kind="process_path", + logical_key=f"process:{ep.fq}->{path.sink_name}@{path.steps[-1]}", + payload=payload, + derived_from=list(grounding.values()), + extractor_id=self.extractor_id, + extractor_version=self.extractor_version, + framework_versions={"sink_registry": digest}, + ) + + +# --- registry ---------------------------------------------------------------- + + +def _load_registry(materialized_root: str) -> tuple[SinkRule, ...]: + path = Path(materialized_root) / SINKS_FILE + if not path.exists(): + return _BUILTIN_SINKS + data = yaml.safe_load(path.read_text()) + if not isinstance(data, dict) or data.get("version") != 1: + raise ValueError(f"{SINKS_FILE}: expected a mapping with version: 1") + extend = data.get("extend", True) + if not isinstance(extend, bool): + raise ValueError(f"{SINKS_FILE}: 'extend' must be a boolean") + overrides: list[SinkRule] = [] + for item in data.get("sinks", []) or []: + if not isinstance(item, dict): + raise ValueError(f"{SINKS_FILE}: each sink must be a mapping, got {item!r}") + name, patterns = item.get("name"), item.get("patterns") + if not isinstance(name, str) or not name.replace("_", "a").isalnum(): + raise ValueError(f"{SINKS_FILE}: bad sink name {name!r}") + if not isinstance(patterns, list) or not patterns: + raise ValueError(f"{SINKS_FILE}: sink {name!r} needs a non-empty patterns list") + for pat in patterns: + if not isinstance(pat, str) or pat.count("*") > 1 or pat in ("*", ""): + raise ValueError(f"{SINKS_FILE}: bad pattern {pat!r} in sink {name!r}") + overrides.append(SinkRule(name, tuple(patterns))) + by_name = {r.name: r for r in _BUILTIN_SINKS} if extend else {} + for rule in overrides: + by_name[rule.name] = rule # same-name override REPLACES the built-in (predictable) + return tuple(sorted(by_name.values(), key=lambda r: r.name)) + + +def _registry_digest(rules: Sequence[SinkRule]) -> str: + canonical = json.dumps( + [{"name": r.name, "patterns": list(r.patterns)} for r in rules], + separators=(",", ":"), ensure_ascii=False, + ) + return hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:12] + + +def _match_call( + rules: Sequence[SinkRule], dotted: str | None, attr: str | None +) -> list[tuple[str, str, str]]: + hits: list[tuple[str, str, str]] = [] + for rule in rules: + for pattern in rule.patterns: + if pattern.startswith("*."): + if attr is not None and attr == pattern[2:]: + hits.append((rule.name, pattern, dotted or f"?.{attr}")) + elif pattern.endswith(".*"): + prefix = pattern[:-2] + if dotted is not None and (dotted == prefix or dotted.startswith(prefix + ".")): + hits.append((rule.name, pattern, dotted)) + elif dotted is not None and dotted == pattern: + hits.append((rule.name, pattern, dotted)) + return hits + + +# --- prior-artifact readers ---------------------------------------------------- + + +def _entrypoints(prior: Sequence[ExtractedArtifact]) -> list[_Entrypoint]: + refs_by_fq: dict[str, list[tuple[str, str]]] = {} + for art in prior: + if art.kind == "api_route": + fq = str(art.payload["handler"]) + ref = f"{art.payload.get('method', '?')} {art.payload.get('path', '?')}" + refs_by_fq.setdefault(fq, []).append(("api_route", ref)) + elif art.kind == "event_handler": + fq = str(art.payload["handler"]) + regs = art.payload.get("registrations", []) + ref = ";".join( + f"{r.get('family', '?')}:{r.get('event')}" if r.get("event") + else str(r.get("family", "?")) + for r in regs + ) + refs_by_fq.setdefault(fq, []).append(("event_handler", ref)) + out: list[_Entrypoint] = [] + for fq in sorted(refs_by_fq): + pairs = tuple(sorted(set(refs_by_fq[fq]))) + out.append(_Entrypoint(fq=fq, kind=pairs[0][0], reference=pairs[0][1], references=pairs)) + return out + + +def _adjacency(prior: Sequence[ExtractedArtifact]) -> dict[str, list[tuple[str, str]]]: + adjacency: dict[str, list[tuple[str, str]]] = {} + for art in prior: + if art.kind != "call_edge": + continue + # class-instantiation edges dead-end (the __init__ body is invisible) -> excluded + if art.payload.get("callee_kind") not in ("function", "method"): + continue + adjacency.setdefault(str(art.payload["caller"]), []).append( + (str(art.payload["callee"]), str(art.payload.get("resolution", "?"))) + ) + for callees in adjacency.values(): + callees.sort() # deterministic BFS tie-break + return adjacency + + +def _def_span_index(ctx: ExtractContext) -> dict[str, ParsedSpan]: + index: dict[str, ParsedSpan] = {} + for spans in ctx.spans_by_module.values(): + for span in spans: + if span.span_kind in ("function", "method"): + index[span.fq_symbol_path] = span # last-wins, the calls.py precedent + return index diff --git a/src/kb/git/repo.py b/src/kb/git/repo.py index e884b6a..7dc8286 100644 --- a/src/kb/git/repo.py +++ b/src/kb/git/repo.py @@ -73,6 +73,39 @@ def iter_python_files_at(repo: pygit2.Repository, rev: str) -> Iterator[tuple[st yield from _walk_tree(repo, commit.tree, "") +def iter_files_under_at( + repo: pygit2.Repository, rev: str, prefix: str +) -> Iterator[tuple[str, bytes]]: + """Yield ``(posix_path, bytes)`` for every blob under ``prefix`` in the tree at ``rev``. + + Materialization aid for extractor config that travels with the commit (e.g. ``.kb/``); the + ``*.py``-only ``iter_python_files_at`` stays untouched so root discovery, parsing and diff + reuse never see non-Python paths. + """ + commit = resolve_commit(repo, rev) + key = prefix.rstrip("/") + try: + entry = commit.tree[key] + except KeyError: + return + if entry.type_str != "tree": + return + subtree = cast("pygit2.Tree", repo[entry.id]) + yield from _walk_any_tree(repo, subtree, f"{key}/") + + +def _walk_any_tree( + repo: pygit2.Repository, tree: pygit2.Tree, prefix: str +) -> Iterator[tuple[str, bytes]]: + for entry in tree: + name = entry.name or "" + path = f"{prefix}{name}" + if entry.type_str == "tree": + yield from _walk_any_tree(repo, cast("pygit2.Tree", repo[entry.id]), f"{path}/") + elif entry.type_str == "blob": + yield path, bytes(cast("pygit2.Blob", repo[entry.id]).data) + + def _walk_tree( repo: pygit2.Repository, tree: pygit2.Tree, prefix: str ) -> Iterator[tuple[str, bytes]]: diff --git a/src/kb/mcp/records.py b/src/kb/mcp/records.py index da57ae8..92f3b69 100644 --- a/src/kb/mcp/records.py +++ b/src/kb/mcp/records.py @@ -90,6 +90,12 @@ def summarize(kind: str, payload: dict[str, Any]) -> str: suffix = f" +{len(regs) - 1} more" if len(regs) > 1 else "" family = first.get("family", "?") return f"{payload.get('handler', '?')} on {target}:{trigger} ({family}{suffix})" + if kind == "process_path": + sink = payload.get("sink", {}) + return ( + f"{payload.get('entrypoint_reference', '?')} -> {sink.get('name', '?')} " + f"@ {payload.get('terminal', '?')} ({payload.get('depth', '?')} hops)" + ) if kind == "call_edge": n = len(payload.get("lines", [])) return ( diff --git a/uv.lock b/uv.lock index 115d786..c02235e 100644 --- a/uv.lock +++ b/uv.lock @@ -1052,6 +1052,7 @@ dependencies = [ { name = "psycopg", extra = ["binary"] }, { name = "pydantic" }, { name = "pygit2" }, + { name = "pyyaml" }, { name = "sqlalchemy" }, { name = "sqlparse" }, { name = "tree-sitter" }, @@ -1066,7 +1067,6 @@ dev = [ { name = "mypy" }, { name = "pytest" }, { name = "pytest-asyncio" }, - { name = "pyyaml" }, { name = "ruff" }, ] embed = [ @@ -1093,7 +1093,7 @@ requires-dist = [ { name = "pygit2", specifier = ">=1.14" }, { name = "pytest", marker = "extra == 'dev'", specifier = ">=8" }, { name = "pytest-asyncio", marker = "extra == 'dev'", specifier = ">=0.23" }, - { name = "pyyaml", marker = "extra == 'dev'", specifier = ">=6" }, + { name = "pyyaml", specifier = ">=6" }, { name = "ruff", marker = "extra == 'dev'", specifier = ">=0.6" }, { name = "sentence-transformers", marker = "extra == 'embed'", specifier = ">=3,<6" }, { name = "sqlalchemy", specifier = ">=2.0,<3" },