From c69cd16515c24bb03d9a313d7a167c7bc06b878d Mon Sep 17 00:00:00 2001 From: Andraxion Date: Wed, 29 Jul 2026 04:09:28 -0400 Subject: [PATCH] Bound indexed retrieval operations --- DEVELOPMENT_NOTES.md | 21 ++++- src/docforge/cli.py | 19 ++++- src/docforge/index.py | 156 +++++++++++++++++++++++++++++-------- src/docforge/mcp_server.py | 26 +++++-- tests/test_core.py | 20 +++++ tests/test_mcp_server.py | 12 ++- 6 files changed, 208 insertions(+), 46 deletions(-) diff --git a/DEVELOPMENT_NOTES.md b/DEVELOPMENT_NOTES.md index 0b50857..5b8da3c 100644 --- a/DEVELOPMENT_NOTES.md +++ b/DEVELOPMENT_NOTES.md @@ -59,9 +59,11 @@ of project size. The inherited dependency-cycle preparation scanned every edge once for every node. The graph validator now constructs dependency adjacency in one edge pass and sorts each adjacency list before -the existing deterministic depth-first cycle check. +an iterative deterministic depth-first cycle check. The iterative stack also removes recursion +depth as a failure mode on large valid graphs. The same edge pass now rejects missing sources as +well as missing targets. -A 2,000-node regression test counts complete edge-collection iteration passes and caps them at +A 10,000-node regression test counts complete edge-collection iteration passes and caps them at four. The focused correctness and bounded-pass tests pass, and the configured strict source type gate is clean. @@ -137,6 +139,21 @@ One audit identified a correctness risk beyond latency: a large mutating MCP ope successfully and then be replaced by `result_too_large`. This must be fixed in Milestone 1 so exactly-once operations never report a false failure after mutation. +#### Bounded indexed retrieval + +Search, metadata filtering, backlinks, dependency traversal, and impact traversal now query one +extra row beyond the requested bound and report `limit` plus `truncated`. Backlinks, dependency, +and impact APIs accept the same additive `limit` option through Python, CLI, and MCP surfaces. +Omitted limits are capped by the project `max_results` policy. + +Traversal no longer loads the complete edge table and repeatedly scans it. It performs +deterministically ordered frontier queries through the existing source primary key or target index. +Each request also has a deterministic edge-examination budget derived from its result limit. The +response includes `examined_edges` and `examined_edges_limit` counters so algorithmic work can be +asserted independently of machine timing. `truncated` is true when either another unique result +exists or the work budget prevents proving completeness. A focused core, CLI, MCP, Ruff, and +Pyright gate passes for this work-in-progress slice. + ### Initial design constraints - Full rebuild remains the recovery and equivalence oracle. diff --git a/src/docforge/cli.py b/src/docforge/cli.py index 75db8e9..e3c2023 100644 --- a/src/docforge/cli.py +++ b/src/docforge/cli.py @@ -53,6 +53,7 @@ def _parser() -> argparse.ArgumentParser: command.add_argument("--relation") else: command.add_argument("--depth", type=int, default=2) + command.add_argument("--limit", type=int) context = commands.add_parser("context") context.add_argument("profile") context.add_argument("--budget", type=int) @@ -149,11 +150,23 @@ def _run(arguments: argparse.Namespace) -> dict[str, object]: limit=arguments.limit, ) if arguments.command == "backlinks": - return index.backlinks(arguments.node_id, relation=arguments.relation) + return index.backlinks( + arguments.node_id, + relation=arguments.relation, + limit=arguments.limit, + ) if arguments.command == "dependencies": - return index.dependencies(arguments.node_id, depth=arguments.depth) + return index.dependencies( + arguments.node_id, + depth=arguments.depth, + limit=arguments.limit, + ) if arguments.command == "impact": - return index.impact(arguments.node_id, depth=arguments.depth) + return index.impact( + arguments.node_id, + depth=arguments.depth, + limit=arguments.limit, + ) if arguments.command == "context": return compile_context(index, arguments.profile, arguments.budget) if arguments.command == "render": diff --git a/src/docforge/index.py b/src/docforge/index.py index f753a1d..9c7a749 100644 --- a/src/docforge/index.py +++ b/src/docforge/index.py @@ -826,14 +826,20 @@ class ProjectIndex: ORDER BY rank, nodes.node_id LIMIT ? """, - (expression, bounded), + (expression, bounded + 1), ).fetchall() results: list[dict[str, object]] = [] - for row in rows: + for row in rows[:bounded]: payload = _row_to_node(row).as_dict(include_content=False) payload.update({"rank": row["rank"], "snippet": row["snippet"]}) results.append(payload) - return snapshot.result(query=query, count=len(results), results=results) + return snapshot.result( + query=query, + count=len(results), + limit=bounded, + truncated=len(rows) > bounded, + results=results, + ) def filter_nodes( self, @@ -857,66 +863,146 @@ class ProjectIndex: where = f"WHERE {' AND '.join(clauses)}" if clauses else "" with self._read_snapshot() as snapshot: rows = snapshot.connection.execute( - f"SELECT * FROM nodes {where} ORDER BY node_id LIMIT ?", (*values, bounded) + f"SELECT * FROM nodes {where} ORDER BY node_id LIMIT ?", + (*values, bounded + 1), ).fetchall() - results = [_row_to_node(row).as_dict(include_content=False) for row in rows] - return snapshot.result(count=len(results), results=results) + results = [_row_to_node(row).as_dict(include_content=False) for row in rows[:bounded]] + return snapshot.result( + count=len(results), + limit=bounded, + truncated=len(rows) > bounded, + results=results, + ) - def backlinks(self, node_id: str, *, relation: str | None = None) -> dict[str, object]: - return self._edges(node_id, incoming=True, relation=relation) + def backlinks( + self, + node_id: str, + *, + relation: str | None = None, + limit: int | None = None, + ) -> dict[str, object]: + return self._edges(node_id, incoming=True, relation=relation, limit=limit) - def dependencies(self, node_id: str, *, depth: int = 2) -> dict[str, object]: - return self._traverse(node_id, incoming=False, depth=depth, relation="depends_on") + def dependencies( + self, + node_id: str, + *, + depth: int = 2, + limit: int | None = None, + ) -> dict[str, object]: + return self._traverse( + node_id, + incoming=False, + depth=depth, + relation="depends_on", + limit=limit, + ) - def impact(self, node_id: str, *, depth: int = 2) -> dict[str, object]: - return self._traverse(node_id, incoming=True, depth=depth, relation=None) + def impact( + self, + node_id: str, + *, + depth: int = 2, + limit: int | None = None, + ) -> dict[str, object]: + return self._traverse( + node_id, + incoming=True, + depth=depth, + relation=None, + limit=limit, + ) - def _edges(self, node_id: str, *, incoming: bool, relation: str | None) -> dict[str, object]: + def _edges( + self, + node_id: str, + *, + incoming: bool, + relation: str | None, + limit: int | None, + ) -> dict[str, object]: + bounded = _bounded_limit( + limit, + self.project.descriptor.limits.max_results, + default=self.project.descriptor.limits.max_results, + ) source_column = "target_id" if incoming else "source_id" relation_clause = " AND relation = ?" if relation is not None else "" - values: tuple[object, ...] = (node_id, relation) if relation is not None else (node_id,) + values: tuple[object, ...] = ( + (node_id, relation, bounded + 1) if relation is not None else (node_id, bounded + 1) + ) with self._read_snapshot() as snapshot: self._require_node(snapshot.connection, node_id) rows = snapshot.connection.execute( f"SELECT source_id, relation, target_id FROM edges " f"WHERE {source_column} = ?{relation_clause} " - "ORDER BY source_id, relation, target_id", + "ORDER BY source_id, relation, target_id LIMIT ?", values, ).fetchall() - return snapshot.result(edges=[Edge(*row).as_dict() for row in rows]) + truncated = len(rows) > bounded + edges = [Edge(*row).as_dict() for row in rows[:bounded]] + return snapshot.result( + count=len(edges), + limit=bounded, + truncated=truncated, + edges=edges, + ) def _traverse( - self, node_id: str, *, incoming: bool, depth: int, relation: str | None + self, + node_id: str, + *, + incoming: bool, + depth: int, + relation: str | None, + limit: int | None, ) -> dict[str, object]: maximum = self.project.descriptor.limits.max_traversal_depth if type(depth) is not int or depth < 0 or depth > maximum: raise DocForgeError("invalid_depth", "Traversal depth is outside the configured limit") + bounded = _bounded_limit( + limit, + self.project.descriptor.limits.max_results, + default=self.project.descriptor.limits.max_results, + ) + examined_limit = (bounded + 1) ** 2 + source_column = "target_id" if incoming else "source_id" + relation_clause = " AND relation = ?" if relation is not None else "" with self._read_snapshot() as snapshot: self._require_node(snapshot.connection, node_id) - edges = tuple( - Edge(*row) - for row in snapshot.connection.execute( - "SELECT source_id, relation, target_id FROM edges " - "ORDER BY source_id, relation, target_id" - ) - ) queue: deque[tuple[str, int, tuple[str, ...]]] = deque([(node_id, 0, (node_id,))]) seen = {node_id} results: list[dict[str, object]] = [] - while queue: + truncated = False + examined_edges = 0 + while queue and len(results) <= bounded and examined_edges < examined_limit: current, current_depth, path = queue.popleft() if current_depth >= depth: continue - candidates = [ - edge - for edge in edges - if (relation is None or edge.relation == relation) - and ((edge.target_id if incoming else edge.source_id) == current) - ] - for edge in candidates: + remaining = examined_limit - examined_edges + values: tuple[object, ...] = ( + (current, relation, remaining + 1) + if relation is not None + else (current, remaining + 1) + ) + candidates = snapshot.connection.execute( + "SELECT source_id, relation, target_id FROM edges " + f"WHERE {source_column} = ?{relation_clause} " + "ORDER BY source_id, relation, target_id LIMIT ?", + values, + ).fetchall() + if len(candidates) > remaining: + truncated = True + candidates = candidates[:remaining] + examined_edges += len(candidates) + for row in candidates: + edge = Edge(*row) target = edge.source_id if incoming else edge.target_id if target in seen: continue + if len(results) >= bounded: + truncated = True + break seen.add(target) target_path = (*path, target) results.append( @@ -928,10 +1014,16 @@ class ProjectIndex: } ) queue.append((target, current_depth + 1, target_path)) + if queue and examined_edges >= examined_limit: + truncated = True return snapshot.result( root=node_id, depth=depth, count=len(results), + limit=bounded, + truncated=truncated, + examined_edges=examined_edges, + examined_edges_limit=examined_limit, results=results, ) diff --git a/src/docforge/mcp_server.py b/src/docforge/mcp_server.py index 7492f1f..65462e1 100644 --- a/src/docforge/mcp_server.py +++ b/src/docforge/mcp_server.py @@ -577,22 +577,36 @@ def _create_bound_server(service: DocForgeService, *, read_only: bool) -> FastMC ) @server.tool(name="docforge_backlinks") - def backlinks(node_id: str, relation: str | None = None) -> dict[str, Any]: + def backlinks( + node_id: str, + relation: str | None = None, + limit: int | None = None, + ) -> dict[str, Any]: """Return bounded incoming relationships for one exact stable node.""" - return service.invoke(lambda: service.index.backlinks(node_id, relation=relation)) + return service.invoke( + lambda: service.index.backlinks(node_id, relation=relation, limit=limit) + ) @server.tool(name="docforge_dependencies") - def dependencies(node_id: str, depth: int = 2) -> dict[str, Any]: + def dependencies( + node_id: str, + depth: int = 2, + limit: int | None = None, + ) -> dict[str, Any]: """Traverse declared depends_on relationships within the configured depth limit.""" - return service.invoke(lambda: service.index.dependencies(node_id, depth=depth)) + return service.invoke(lambda: service.index.dependencies(node_id, depth=depth, limit=limit)) @server.tool(name="docforge_impact") - def impact(node_id: str, depth: int = 2) -> dict[str, Any]: + def impact( + node_id: str, + depth: int = 2, + limit: int | None = None, + ) -> dict[str, Any]: """Traverse bounded incoming relationships and report exact paths.""" - return service.invoke(lambda: service.index.impact(node_id, depth=depth)) + return service.invoke(lambda: service.index.impact(node_id, depth=depth, limit=limit)) @server.tool(name="docforge_get_context") def get_context(profile: str, budget: int | None = None) -> dict[str, Any]: diff --git a/tests/test_core.py b/tests/test_core.py index e8075cc..dd5ece8 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -333,6 +333,9 @@ class DocForgeCoreTests(unittest.TestCase): self.assertEqual( ["proof.validation"], [item["node_id"] for item in filtered["results"]] ) + limited_filter = index.filter_nodes(limit=1) + self.assertEqual(1, limited_filter["count"]) + self.assertTrue(limited_filter["truncated"]) dependencies = index.dependencies("guide.workflow", depth=2) self.assertEqual( ["guide.foundation"], [item["node_id"] for item in dependencies["results"]] @@ -346,6 +349,23 @@ class DocForgeCoreTests(unittest.TestCase): ["guide.workflow", "proof.validation"], [item["node_id"] for item in impact["results"]], ) + limited = index.impact("guide.foundation", depth=2, limit=1) + self.assertEqual(["guide.workflow"], [item["node_id"] for item in limited["results"]]) + self.assertEqual(1, limited["limit"]) + self.assertTrue(limited["truncated"]) + self.assertLessEqual( + limited["examined_edges"], + limited["examined_edges_limit"], + ) + + complete_limit = index.dependencies("guide.workflow", depth=2, limit=1) + self.assertEqual(1, complete_limit["count"]) + self.assertFalse(complete_limit["truncated"]) + + limited_backlinks = index.backlinks("guide.workflow", limit=1) + self.assertEqual(1, limited_backlinks["count"]) + self.assertEqual(1, limited_backlinks["limit"]) + self.assertFalse(limited_backlinks["truncated"]) def test_query_rechecks_source_identity_before_returning(self) -> None: with tempfile.TemporaryDirectory() as directory: diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index a4d3b40..0fee356 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -131,9 +131,15 @@ class DocForgeMcpTests(unittest.IsolatedAsyncioTestCase): ("docforge_get_logic", {"owner_node_id": "guide.workflow"}), ("docforge_search", {"query": "canonical nodes", "limit": 5}), ("docforge_filter_nodes", {"family": "proof", "tag": "validation"}), - ("docforge_backlinks", {"node_id": "guide.workflow"}), - ("docforge_dependencies", {"node_id": "guide.workflow", "depth": 2}), - ("docforge_impact", {"node_id": "guide.foundation", "depth": 2}), + ("docforge_backlinks", {"node_id": "guide.workflow", "limit": 5}), + ( + "docforge_dependencies", + {"node_id": "guide.workflow", "depth": 2, "limit": 5}, + ), + ( + "docforge_impact", + {"node_id": "guide.foundation", "depth": 2, "limit": 5}, + ), ("docforge_get_context", {"profile": "active", "budget": 180}), ("docforge_validate_project", {}), ("docforge_render_status", {}),