Navegador

feat: knowledge layer MCP tools — get_rationale, find_owners, search_knowledge Adds three new MCP tools for the knowledge layer: - get_rationale: retrieve decision rationale, alternatives, status - find_owners: find people assigned to any node - search_knowledge: search concepts, rules, decisions, wiki pages Includes new Cypher queries (DECISION_RATIONALE, FIND_OWNERS), ContextLoader methods (load_decision, find_owners, search_knowledge), and full test coverage. Closes #6

lmata 2026-03-23 04:46 trunk
Commit ece88cab8dc2f26cb639c8641ed959422c4394bb223a969aa3fe7bbd5a5a4867
Parent 1e364db043bb46a…
5 files changed +74 +25 +73 +109 +88 -2
~ navegador/context/loader.py ~ navegador/graph/queries.py ~ navegador/mcp/server.py ~ tests/test_context.py ~ tests/test_mcp_server.py
M navegador/context/loader.py
+74
--- navegador/context/loader.py
+++ navegador/context/loader.py
@@ -29,10 +29,13 @@
2929
 signature: str | None = None
3030
 source: str | None = None
3131
 description: str | None = None
3232
 domain: str | None = None
3333
 status: str | None = None
34
+ rationale: str | None = None
35
+ alternatives: str | None = None
36
+ date: str | None = None
3437
3538
3639
 @dataclass
3740
 class ContextBundle:
3841
 target: ContextNode
@@ -289,10 +292,81 @@
289292
 """Search functions/classes whose docstring contains the query."""
290293
 result = self.store.query(queries.DOCSTRING_SEARCH, {"query": query, "limit": limit})
291294
 return [
292295
 ContextNode(
293296
 type=row[0], name=row[1], file_path=row[2], line_start=row[3], docstring=row[4]
297
+ )
298
+ for row in (result.result_set or [])
299
+ ]
300
+
301
+ # ── Knowledge: decision rationale ────────────────────────────────────────
302
+
303
+ def load_decision(self, name: str) -> ContextBundle:
304
+ """Decision rationale, alternatives, status, and related nodes."""
305
+ result = self.store.query(queries.DECISION_RATIONALE, {"name": name})
306
+ rows = result.result_set or []
307
+
308
+ if not rows:
309
+ return ContextBundle(
310
+ target=ContextNode(type="Decision", name=name),
311
+ metadata={"query": "decision_rationale", "found": False},
312
+ )
313
+
314
+ row = rows[0]
315
+ target = ContextNode(
316
+ type="Decision",
317
+ name=row[0],
318
+ description=row[1],
319
+ status=row[4],
320
+ domain=row[6],
321
+ )
322
+ target.rationale = row[2]
323
+ target.alternatives = row[3]
324
+ target.date = row[5]
325
+
326
+ nodes: list[ContextNode] = []
327
+ edges: list[dict[str, str]] = []
328
+
329
+ for tname in row[7] or []:
330
+ nodes.append(ContextNode(type="Node", name=tname))
331
+ edges.append({"from": name, "type": "DOCUMENTS", "to": tname})
332
+ for pname in row[8] or []:
333
+ nodes.append(ContextNode(type="Person", name=pname))
334
+ edges.append({"from": name, "type": "DECIDED_BY", "to": pname})
335
+
336
+ return ContextBundle(
337
+ target=target, nodes=nodes, edges=edges, metadata={"query": "decision_rationale"}
338
+ )
339
+
340
+ # ── Knowledge: find owners ────────────────────────────────────────────────
341
+
342
+ def find_owners(self, name: str, file_path: str = "") -> list[ContextNode]:
343
+ """Find people assigned to a named node."""
344
+ result = self.store.query(
345
+ queries.FIND_OWNERS, {"name": name, "file_path": file_path}
346
+ )
347
+ return [
348
+ ContextNode(
349
+ type="Person",
350
+ name=row[2],
351
+ description=f"role={row[4]}, team={row[5]}",
352
+ )
353
+ for row in (result.result_set or [])
354
+ ]
355
+
356
+ # ── Knowledge: search ────────────────────────────────────────────────────
357
+
358
+ def search_knowledge(self, query: str, limit: int = 20) -> list[ContextNode]:
359
+ """Search concepts, rules, decisions, and wiki pages."""
360
+ result = self.store.query(queries.KNOWLEDGE_SEARCH, {"query": query, "limit": limit})
361
+ return [
362
+ ContextNode(
363
+ type=row[0],
364
+ name=row[1],
365
+ description=row[2],
366
+ domain=row[3],
367
+ status=row[4],
294368
 )
295369
 for row in (result.result_set or [])
296370
 ]
297371
298372
 def decorated_by(self, decorator_name: str) -> list[ContextNode]:
299373
--- navegador/context/loader.py
+++ navegador/context/loader.py
@@ -29,10 +29,13 @@
29 signature: str | None = None
30 source: str | None = None
31 description: str | None = None
32 domain: str | None = None
33 status: str | None = None
 
 
 
34
35
36 @dataclass
37 class ContextBundle:
38 target: ContextNode
@@ -289,10 +292,81 @@
289 """Search functions/classes whose docstring contains the query."""
290 result = self.store.query(queries.DOCSTRING_SEARCH, {"query": query, "limit": limit})
291 return [
292 ContextNode(
293 type=row[0], name=row[1], file_path=row[2], line_start=row[3], docstring=row[4]
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
294 )
295 for row in (result.result_set or [])
296 ]
297
298 def decorated_by(self, decorator_name: str) -> list[ContextNode]:
299
--- navegador/context/loader.py
+++ navegador/context/loader.py
@@ -29,10 +29,13 @@
29 signature: str | None = None
30 source: str | None = None
31 description: str | None = None
32 domain: str | None = None
33 status: str | None = None
34 rationale: str | None = None
35 alternatives: str | None = None
36 date: str | None = None
37
38
39 @dataclass
40 class ContextBundle:
41 target: ContextNode
@@ -289,10 +292,81 @@
292 """Search functions/classes whose docstring contains the query."""
293 result = self.store.query(queries.DOCSTRING_SEARCH, {"query": query, "limit": limit})
294 return [
295 ContextNode(
296 type=row[0], name=row[1], file_path=row[2], line_start=row[3], docstring=row[4]
297 )
298 for row in (result.result_set or [])
299 ]
300
301 # ── Knowledge: decision rationale ────────────────────────────────────────
302
303 def load_decision(self, name: str) -> ContextBundle:
304 """Decision rationale, alternatives, status, and related nodes."""
305 result = self.store.query(queries.DECISION_RATIONALE, {"name": name})
306 rows = result.result_set or []
307
308 if not rows:
309 return ContextBundle(
310 target=ContextNode(type="Decision", name=name),
311 metadata={"query": "decision_rationale", "found": False},
312 )
313
314 row = rows[0]
315 target = ContextNode(
316 type="Decision",
317 name=row[0],
318 description=row[1],
319 status=row[4],
320 domain=row[6],
321 )
322 target.rationale = row[2]
323 target.alternatives = row[3]
324 target.date = row[5]
325
326 nodes: list[ContextNode] = []
327 edges: list[dict[str, str]] = []
328
329 for tname in row[7] or []:
330 nodes.append(ContextNode(type="Node", name=tname))
331 edges.append({"from": name, "type": "DOCUMENTS", "to": tname})
332 for pname in row[8] or []:
333 nodes.append(ContextNode(type="Person", name=pname))
334 edges.append({"from": name, "type": "DECIDED_BY", "to": pname})
335
336 return ContextBundle(
337 target=target, nodes=nodes, edges=edges, metadata={"query": "decision_rationale"}
338 )
339
340 # ── Knowledge: find owners ────────────────────────────────────────────────
341
342 def find_owners(self, name: str, file_path: str = "") -> list[ContextNode]:
343 """Find people assigned to a named node."""
344 result = self.store.query(
345 queries.FIND_OWNERS, {"name": name, "file_path": file_path}
346 )
347 return [
348 ContextNode(
349 type="Person",
350 name=row[2],
351 description=f"role={row[4]}, team={row[5]}",
352 )
353 for row in (result.result_set or [])
354 ]
355
356 # ── Knowledge: search ────────────────────────────────────────────────────
357
358 def search_knowledge(self, query: str, limit: int = 20) -> list[ContextNode]:
359 """Search concepts, rules, decisions, and wiki pages."""
360 result = self.store.query(queries.KNOWLEDGE_SEARCH, {"query": query, "limit": limit})
361 return [
362 ContextNode(
363 type=row[0],
364 name=row[1],
365 description=row[2],
366 domain=row[3],
367 status=row[4],
368 )
369 for row in (result.result_set or [])
370 ]
371
372 def decorated_by(self, decorator_name: str) -> list[ContextNode]:
373
M navegador/graph/queries.py
+25
--- navegador/graph/queries.py
+++ navegador/graph/queries.py
@@ -184,10 +184,35 @@
184184
 RETURN type(r) AS rel, labels(neighbor)[0] AS neighbor_type,
185185
 neighbor.name AS neighbor_name,
186186
 coalesce(neighbor.file_path, '') AS neighbor_file_path
187187
 ORDER BY rel, neighbor_name
188188
 """
189
+
190
+# ── Knowledge: decision rationale ─────────────────────────────────────────────
191
+
192
+DECISION_RATIONALE = """
193
+MATCH (d:Decision {name: $name})
194
+OPTIONAL MATCH (d)-[:DOCUMENTS]->(target)
195
+OPTIONAL MATCH (d)-[:DECIDED_BY]->(person:Person)
196
+OPTIONAL MATCH (d)-[:BELONGS_TO]->(domain:Domain)
197
+RETURN
198
+ d.name AS name, d.description AS description,
199
+ d.rationale AS rationale, d.alternatives AS alternatives,
200
+ d.status AS status, d.date AS date, d.domain AS domain,
201
+ collect(DISTINCT target.name) AS documents,
202
+ collect(DISTINCT person.name) AS decided_by,
203
+ collect(DISTINCT domain.name) AS domains
204
+"""
205
+
206
+# ── Knowledge: find owners (ASSIGNED_TO → Person) ────────────────────────────
207
+
208
+FIND_OWNERS = """
209
+MATCH (n)-[:ASSIGNED_TO]->(p:Person)
210
+WHERE n.name = $name AND ($file_path = '' OR n.file_path = $file_path)
211
+RETURN labels(n)[0] AS node_type, n.name AS node_name,
212
+ p.name AS owner, p.email AS email, p.role AS role, p.team AS team
213
+"""
189214
190215
 # ── Stats ─────────────────────────────────────────────────────────────────────
191216
192217
 NODE_TYPE_COUNTS = """
193218
 MATCH (n)
194219
--- navegador/graph/queries.py
+++ navegador/graph/queries.py
@@ -184,10 +184,35 @@
184 RETURN type(r) AS rel, labels(neighbor)[0] AS neighbor_type,
185 neighbor.name AS neighbor_name,
186 coalesce(neighbor.file_path, '') AS neighbor_file_path
187 ORDER BY rel, neighbor_name
188 """
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
189
190 # ── Stats ─────────────────────────────────────────────────────────────────────
191
192 NODE_TYPE_COUNTS = """
193 MATCH (n)
194
--- navegador/graph/queries.py
+++ navegador/graph/queries.py
@@ -184,10 +184,35 @@
184 RETURN type(r) AS rel, labels(neighbor)[0] AS neighbor_type,
185 neighbor.name AS neighbor_name,
186 coalesce(neighbor.file_path, '') AS neighbor_file_path
187 ORDER BY rel, neighbor_name
188 """
189
190 # ── Knowledge: decision rationale ─────────────────────────────────────────────
191
192 DECISION_RATIONALE = """
193 MATCH (d:Decision {name: $name})
194 OPTIONAL MATCH (d)-[:DOCUMENTS]->(target)
195 OPTIONAL MATCH (d)-[:DECIDED_BY]->(person:Person)
196 OPTIONAL MATCH (d)-[:BELONGS_TO]->(domain:Domain)
197 RETURN
198 d.name AS name, d.description AS description,
199 d.rationale AS rationale, d.alternatives AS alternatives,
200 d.status AS status, d.date AS date, d.domain AS domain,
201 collect(DISTINCT target.name) AS documents,
202 collect(DISTINCT person.name) AS decided_by,
203 collect(DISTINCT domain.name) AS domains
204 """
205
206 # ── Knowledge: find owners (ASSIGNED_TO → Person) ────────────────────────────
207
208 FIND_OWNERS = """
209 MATCH (n)-[:ASSIGNED_TO]->(p:Person)
210 WHERE n.name = $name AND ($file_path = '' OR n.file_path = $file_path)
211 RETURN labels(n)[0] AS node_type, n.name AS node_name,
212 p.name AS owner, p.email AS email, p.role AS role, p.team AS team
213 """
214
215 # ── Stats ─────────────────────────────────────────────────────────────────────
216
217 NODE_TYPE_COUNTS = """
218 MATCH (n)
219
M navegador/mcp/server.py
+73
--- navegador/mcp/server.py
+++ navegador/mcp/server.py
@@ -137,10 +137,54 @@
137137
 Tool(
138138
 name="graph_stats",
139139
 description="Return node and edge counts for the current graph.",
140140
 inputSchema={"type": "object", "properties": {}},
141141
 ),
142
+ Tool(
143
+ name="get_rationale",
144
+ description="Return the rationale, alternatives, and status of an architectural decision.",
145
+ inputSchema={
146
+ "type": "object",
147
+ "properties": {
148
+ "name": {"type": "string", "description": "Decision name."},
149
+ "format": {
150
+ "type": "string",
151
+ "enum": ["json", "markdown"],
152
+ "default": "markdown",
153
+ },
154
+ },
155
+ "required": ["name"],
156
+ },
157
+ ),
158
+ Tool(
159
+ name="find_owners",
160
+ description="Find people (owners, stakeholders) assigned to a node.",
161
+ inputSchema={
162
+ "type": "object",
163
+ "properties": {
164
+ "name": {"type": "string", "description": "Node name."},
165
+ "file_path": {
166
+ "type": "string",
167
+ "description": "Narrow to a specific file.",
168
+ "default": "",
169
+ },
170
+ },
171
+ "required": ["name"],
172
+ },
173
+ ),
174
+ Tool(
175
+ name="search_knowledge",
176
+ description="Search concepts, rules, decisions, and wiki pages by name or description.",
177
+ inputSchema={
178
+ "type": "object",
179
+ "properties": {
180
+ "query": {"type": "string", "description": "Search query."},
181
+ "limit": {"type": "integer", "default": 20},
182
+ },
183
+ "required": ["query"],
184
+ },
185
+ ),
142186
 ]
143187
144188
 @server.call_tool()
145189
 async def call_tool(name: str, arguments: dict) -> list[TextContent]:
146190
 loader = _get_loader()
@@ -190,8 +234,37 @@
190234
 "nodes": loader.store.node_count(),
191235
 "edges": loader.store.edge_count(),
192236
 }
193237
 return [TextContent(type="text", text=json.dumps(stats, indent=2))]
194238
239
+ elif name == "get_rationale":
240
+ bundle = loader.load_decision(arguments["name"])
241
+ fmt = arguments.get("format", "markdown")
242
+ text = bundle.to_markdown() if fmt == "markdown" else bundle.to_json()
243
+ return [TextContent(type="text", text=text)]
244
+
245
+ elif name == "find_owners":
246
+ results = loader.find_owners(
247
+ arguments["name"], file_path=arguments.get("file_path", "")
248
+ )
249
+ if not results:
250
+ return [TextContent(type="text", text="No owners found.")]
251
+ lines = [
252
+ f"- **{r.name}** ({r.description})" for r in results
253
+ ]
254
+ return [TextContent(type="text", text="\n".join(lines))]
255
+
256
+ elif name == "search_knowledge":
257
+ results = loader.search_knowledge(
258
+ arguments["query"], limit=arguments.get("limit", 20)
259
+ )
260
+ if not results:
261
+ return [TextContent(type="text", text="No results.")]
262
+ lines = [
263
+ f"- **{r.type}** `{r.name}` — {r.description or ''}"
264
+ for r in results
265
+ ]
266
+ return [TextContent(type="text", text="\n".join(lines))]
267
+
195268
 return [TextContent(type="text", text=f"Unknown tool: {name}")]
196269
197270
 return server
198271
--- navegador/mcp/server.py
+++ navegador/mcp/server.py
@@ -137,10 +137,54 @@
137 Tool(
138 name="graph_stats",
139 description="Return node and edge counts for the current graph.",
140 inputSchema={"type": "object", "properties": {}},
141 ),
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
142 ]
143
144 @server.call_tool()
145 async def call_tool(name: str, arguments: dict) -> list[TextContent]:
146 loader = _get_loader()
@@ -190,8 +234,37 @@
190 "nodes": loader.store.node_count(),
191 "edges": loader.store.edge_count(),
192 }
193 return [TextContent(type="text", text=json.dumps(stats, indent=2))]
194
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
195 return [TextContent(type="text", text=f"Unknown tool: {name}")]
196
197 return server
198
--- navegador/mcp/server.py
+++ navegador/mcp/server.py
@@ -137,10 +137,54 @@
137 Tool(
138 name="graph_stats",
139 description="Return node and edge counts for the current graph.",
140 inputSchema={"type": "object", "properties": {}},
141 ),
142 Tool(
143 name="get_rationale",
144 description="Return the rationale, alternatives, and status of an architectural decision.",
145 inputSchema={
146 "type": "object",
147 "properties": {
148 "name": {"type": "string", "description": "Decision name."},
149 "format": {
150 "type": "string",
151 "enum": ["json", "markdown"],
152 "default": "markdown",
153 },
154 },
155 "required": ["name"],
156 },
157 ),
158 Tool(
159 name="find_owners",
160 description="Find people (owners, stakeholders) assigned to a node.",
161 inputSchema={
162 "type": "object",
163 "properties": {
164 "name": {"type": "string", "description": "Node name."},
165 "file_path": {
166 "type": "string",
167 "description": "Narrow to a specific file.",
168 "default": "",
169 },
170 },
171 "required": ["name"],
172 },
173 ),
174 Tool(
175 name="search_knowledge",
176 description="Search concepts, rules, decisions, and wiki pages by name or description.",
177 inputSchema={
178 "type": "object",
179 "properties": {
180 "query": {"type": "string", "description": "Search query."},
181 "limit": {"type": "integer", "default": 20},
182 },
183 "required": ["query"],
184 },
185 ),
186 ]
187
188 @server.call_tool()
189 async def call_tool(name: str, arguments: dict) -> list[TextContent]:
190 loader = _get_loader()
@@ -190,8 +234,37 @@
234 "nodes": loader.store.node_count(),
235 "edges": loader.store.edge_count(),
236 }
237 return [TextContent(type="text", text=json.dumps(stats, indent=2))]
238
239 elif name == "get_rationale":
240 bundle = loader.load_decision(arguments["name"])
241 fmt = arguments.get("format", "markdown")
242 text = bundle.to_markdown() if fmt == "markdown" else bundle.to_json()
243 return [TextContent(type="text", text=text)]
244
245 elif name == "find_owners":
246 results = loader.find_owners(
247 arguments["name"], file_path=arguments.get("file_path", "")
248 )
249 if not results:
250 return [TextContent(type="text", text="No owners found.")]
251 lines = [
252 f"- **{r.name}** ({r.description})" for r in results
253 ]
254 return [TextContent(type="text", text="\n".join(lines))]
255
256 elif name == "search_knowledge":
257 results = loader.search_knowledge(
258 arguments["query"], limit=arguments.get("limit", 20)
259 )
260 if not results:
261 return [TextContent(type="text", text="No results.")]
262 lines = [
263 f"- **{r.type}** `{r.name}` — {r.description or ''}"
264 for r in results
265 ]
266 return [TextContent(type="text", text="\n".join(lines))]
267
268 return [TextContent(type="text", text=f"Unknown tool: {name}")]
269
270 return server
271
M tests/test_context.py
+109
--- tests/test_context.py
+++ tests/test_context.py
@@ -462,5 +462,114 @@
462462
 types = {e["type"] for e in bundle.edges}
463463
 assert "RELATED_TO" in types
464464
 assert "GOVERNS" in types
465465
 assert "DOCUMENTS" in types
466466
 assert "IMPLEMENTS" in types
467
+
468
+
469
+# ── load_decision ───────────────────────────────────────────────────────────
470
+
471
+class TestContextLoaderDecision:
472
+ def test_load_decision_not_found(self):
473
+ store = _mock_store([])
474
+ loader = ContextLoader(store)
475
+ bundle = loader.load_decision("Nonexistent")
476
+ assert bundle.metadata.get("found") is False
477
+ assert bundle.target.type == "Decision"
478
+
479
+ def test_load_decision_found(self):
480
+ rows = [[
481
+ "Use FalkorDB",
482
+ "Graph DB for navegador",
483
+ "Cypher queries, SQLite backend",
484
+ "Neo4j, ArangoDB",
485
+ "accepted",
486
+ "2026-03-01",
487
+ "infrastructure",
488
+ [], # documents
489
+ [], # decided_by
490
+ [], # domains
491
+ ]]
492
+ store = _mock_store(rows)
493
+ loader = ContextLoader(store)
494
+ bundle = loader.load_decision("Use FalkorDB")
495
+ assert bundle.target.name == "Use FalkorDB"
496
+ assert bundle.target.rationale == "Cypher queries, SQLite backend"
497
+ assert bundle.target.alternatives == "Neo4j, ArangoDB"
498
+ assert bundle.target.status == "accepted"
499
+
500
+ def test_load_decision_with_related_nodes(self):
501
+ rows = [[
502
+ "Use FalkorDB",
503
+ "Graph DB",
504
+ "Cypher",
505
+ "Neo4j",
506
+ "accepted",
507
+ "2026-03-01",
508
+ "infra",
509
+ ["GraphStore"], # documents
510
+ ["Alice"], # decided_by
511
+ ["infrastructure"], # domains
512
+ ]]
513
+ store = _mock_store(rows)
514
+ loader = ContextLoader(store)
515
+ bundle = loader.load_decision("Use FalkorDB")
516
+ names = {n.name for n in bundle.nodes}
517
+ assert "GraphStore" in names
518
+ assert "Alice" in names
519
+ edge_types = {e["type"] for e in bundle.edges}
520
+ assert "DOCUMENTS" in edge_types
521
+ assert "DECIDED_BY" in edge_types
522
+
523
+
524
+# ── find_owners ──────────────────────────────────────────────────────────────
525
+
526
+class TestContextLoaderFindOwners:
527
+ def test_find_owners_empty(self):
528
+ store = _mock_store([])
529
+ loader = ContextLoader(store)
530
+ results = loader.find_owners("AuthService")
531
+ assert results == []
532
+
533
+ def test_find_owners_returns_people(self):
534
+ rows = [["Class", "AuthService", "Alice", "[email protected]", "lead", "auth"]]
535
+ store = _mock_store(rows)
536
+ loader = ContextLoader(store)
537
+ results = loader.find_owners("AuthService")
538
+ assert len(results) == 1
539
+ assert results[0].name == "Alice"
540
+ assert results[0].type == "Person"
541
+
542
+ def test_find_owners_passes_file_path(self):
543
+ store = _mock_store([])
544
+ loader = ContextLoader(store)
545
+ loader.find_owners("foo", file_path="src/foo.py")
546
+ store.query.assert_called_once()
547
+ args = store.query.call_args
548
+ assert args[0][1]["file_path"] == "src/foo.py"
549
+
550
+
551
+# ── search_knowledge ────────────────────────────────────────────────────────
552
+
553
+class TestContextLoaderSearchKnowledge:
554
+ def test_search_knowledge_empty(self):
555
+ store = _mock_store([])
556
+ loader = ContextLoader(store)
557
+ results = loader.search_knowledge("xyz")
558
+ assert results == []
559
+
560
+ def test_search_knowledge_returns_nodes(self):
561
+ rows = [["Concept", "JWT", "Stateless token auth", "auth", "active"]]
562
+ store = _mock_store(rows)
563
+ loader = ContextLoader(store)
564
+ results = loader.search_knowledge("JWT")
565
+ assert len(results) == 1
566
+ assert results[0].name == "JWT"
567
+ assert results[0].type == "Concept"
568
+ assert results[0].domain == "auth"
569
+
570
+ def test_search_knowledge_passes_limit(self):
571
+ store = _mock_store([])
572
+ loader = ContextLoader(store)
573
+ loader.search_knowledge("auth", limit=5)
574
+ args = store.query.call_args
575
+ assert args[0][1]["limit"] == 5
467576
--- tests/test_context.py
+++ tests/test_context.py
@@ -462,5 +462,114 @@
462 types = {e["type"] for e in bundle.edges}
463 assert "RELATED_TO" in types
464 assert "GOVERNS" in types
465 assert "DOCUMENTS" in types
466 assert "IMPLEMENTS" in types
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
467
--- tests/test_context.py
+++ tests/test_context.py
@@ -462,5 +462,114 @@
462 types = {e["type"] for e in bundle.edges}
463 assert "RELATED_TO" in types
464 assert "GOVERNS" in types
465 assert "DOCUMENTS" in types
466 assert "IMPLEMENTS" in types
467
468
469 # ── load_decision ───────────────────────────────────────────────────────────
470
471 class TestContextLoaderDecision:
472 def test_load_decision_not_found(self):
473 store = _mock_store([])
474 loader = ContextLoader(store)
475 bundle = loader.load_decision("Nonexistent")
476 assert bundle.metadata.get("found") is False
477 assert bundle.target.type == "Decision"
478
479 def test_load_decision_found(self):
480 rows = [[
481 "Use FalkorDB",
482 "Graph DB for navegador",
483 "Cypher queries, SQLite backend",
484 "Neo4j, ArangoDB",
485 "accepted",
486 "2026-03-01",
487 "infrastructure",
488 [], # documents
489 [], # decided_by
490 [], # domains
491 ]]
492 store = _mock_store(rows)
493 loader = ContextLoader(store)
494 bundle = loader.load_decision("Use FalkorDB")
495 assert bundle.target.name == "Use FalkorDB"
496 assert bundle.target.rationale == "Cypher queries, SQLite backend"
497 assert bundle.target.alternatives == "Neo4j, ArangoDB"
498 assert bundle.target.status == "accepted"
499
500 def test_load_decision_with_related_nodes(self):
501 rows = [[
502 "Use FalkorDB",
503 "Graph DB",
504 "Cypher",
505 "Neo4j",
506 "accepted",
507 "2026-03-01",
508 "infra",
509 ["GraphStore"], # documents
510 ["Alice"], # decided_by
511 ["infrastructure"], # domains
512 ]]
513 store = _mock_store(rows)
514 loader = ContextLoader(store)
515 bundle = loader.load_decision("Use FalkorDB")
516 names = {n.name for n in bundle.nodes}
517 assert "GraphStore" in names
518 assert "Alice" in names
519 edge_types = {e["type"] for e in bundle.edges}
520 assert "DOCUMENTS" in edge_types
521 assert "DECIDED_BY" in edge_types
522
523
524 # ── find_owners ──────────────────────────────────────────────────────────────
525
526 class TestContextLoaderFindOwners:
527 def test_find_owners_empty(self):
528 store = _mock_store([])
529 loader = ContextLoader(store)
530 results = loader.find_owners("AuthService")
531 assert results == []
532
533 def test_find_owners_returns_people(self):
534 rows = [["Class", "AuthService", "Alice", "[email protected]", "lead", "auth"]]
535 store = _mock_store(rows)
536 loader = ContextLoader(store)
537 results = loader.find_owners("AuthService")
538 assert len(results) == 1
539 assert results[0].name == "Alice"
540 assert results[0].type == "Person"
541
542 def test_find_owners_passes_file_path(self):
543 store = _mock_store([])
544 loader = ContextLoader(store)
545 loader.find_owners("foo", file_path="src/foo.py")
546 store.query.assert_called_once()
547 args = store.query.call_args
548 assert args[0][1]["file_path"] == "src/foo.py"
549
550
551 # ── search_knowledge ────────────────────────────────────────────────────────
552
553 class TestContextLoaderSearchKnowledge:
554 def test_search_knowledge_empty(self):
555 store = _mock_store([])
556 loader = ContextLoader(store)
557 results = loader.search_knowledge("xyz")
558 assert results == []
559
560 def test_search_knowledge_returns_nodes(self):
561 rows = [["Concept", "JWT", "Stateless token auth", "auth", "active"]]
562 store = _mock_store(rows)
563 loader = ContextLoader(store)
564 results = loader.search_knowledge("JWT")
565 assert len(results) == 1
566 assert results[0].name == "JWT"
567 assert results[0].type == "Concept"
568 assert results[0].domain == "auth"
569
570 def test_search_knowledge_passes_limit(self):
571 store = _mock_store([])
572 loader = ContextLoader(store)
573 loader.search_knowledge("auth", limit=5)
574 args = store.query.call_args
575 assert args[0][1]["limit"] == 5
576
M tests/test_mcp_server.py
+88 -2
--- tests/test_mcp_server.py
+++ tests/test_mcp_server.py
@@ -43,11 +43,14 @@
4343
 loader = MagicMock(spec=ContextLoader)
4444
 loader.store = self.store
4545
 loader.load_file.return_value = _bundle("file_target")
4646
 loader.load_function.return_value = _bundle("fn_target")
4747
 loader.load_class.return_value = _bundle("cls_target")
48
+ loader.load_decision.return_value = _bundle("decision_target")
4849
 loader.search.return_value = []
50
+ loader.find_owners.return_value = []
51
+ loader.search_knowledge.return_value = []
4952
 return loader
5053
5154
 def _build(self):
5255
 list_holder = {}
5356
 call_holder = {}
@@ -108,13 +111,13 @@
108111
 class TestListTools:
109112
 def setup_method(self):
110113
 self.fx = _ServerFixture()
111114
112115
 @pytest.mark.asyncio
113
- async def test_returns_seven_tools(self):
116
+ async def test_returns_ten_tools(self):
114117
 tools = await self.fx.list_tools_fn()
115
- assert len(tools) == 7
118
+ assert len(tools) == 10
116119
117120
 @pytest.mark.asyncio
118121
 async def test_tool_names(self):
119122
 tools = await self.fx.list_tools_fn()
120123
 names = {t["name"] for t in tools}
@@ -124,10 +127,13 @@
124127
 "load_function_context",
125128
 "load_class_context",
126129
 "search_symbols",
127130
 "query_graph",
128131
 "graph_stats",
132
+ "get_rationale",
133
+ "find_owners",
134
+ "search_knowledge",
129135
 }
130136
131137
 @pytest.mark.asyncio
132138
 async def test_ingest_repo_requires_path(self):
133139
 tools = await self.fx.list_tools_fn()
@@ -338,10 +344,90 @@
338344
 result = await self.fx.call_tool_fn("graph_stats", {})
339345
 data = json.loads(result[0]["text"])
340346
 assert data["nodes"] == 42
341347
 assert data["edges"] == 17
342348
349
+
350
+# ── call_tool — unknown tool ──────────────────────────────────────────────────
351
+
352
+# ── call_tool — get_rationale ────────────────────────────────────────────────
353
+
354
+class TestCallToolGetRationale:
355
+ def setup_method(self):
356
+ self.fx = _ServerFixture()
357
+
358
+ @pytest.mark.asyncio
359
+ async def test_returns_markdown_by_default(self):
360
+ result = await self.fx.call_tool_fn("get_rationale", {"name": "Use FalkorDB"})
361
+ self.fx.loader.load_decision.assert_called_once_with("Use FalkorDB")
362
+ assert "decision_target" in result[0]["text"]
363
+
364
+ @pytest.mark.asyncio
365
+ async def test_returns_json_when_requested(self):
366
+ result = await self.fx.call_tool_fn(
367
+ "get_rationale", {"name": "Use FalkorDB", "format": "json"}
368
+ )
369
+ data = json.loads(result[0]["text"])
370
+ assert data["target"]["name"] == "decision_target"
371
+
372
+
373
+# ── call_tool — find_owners ──────────────────────────────────────────────────
374
+
375
+class TestCallToolFindOwners:
376
+ def setup_method(self):
377
+ self.fx = _ServerFixture()
378
+
379
+ @pytest.mark.asyncio
380
+ async def test_returns_no_owners_message(self):
381
+ result = await self.fx.call_tool_fn("find_owners", {"name": "AuthService"})
382
+ assert result[0]["text"] == "No owners found."
383
+
384
+ @pytest.mark.asyncio
385
+ async def test_formats_owners(self):
386
+ owner = ContextNode(name="Alice", type="Person", description="role=lead, team=auth")
387
+ self.fx.loader.find_owners.return_value = [owner]
388
+ result = await self.fx.call_tool_fn("find_owners", {"name": "AuthService"})
389
+ assert "Alice" in result[0]["text"]
390
+ assert "role=lead" in result[0]["text"]
391
+
392
+ @pytest.mark.asyncio
393
+ async def test_passes_file_path(self):
394
+ await self.fx.call_tool_fn(
395
+ "find_owners", {"name": "AuthService", "file_path": "auth.py"}
396
+ )
397
+ self.fx.loader.find_owners.assert_called_once_with("AuthService", file_path="auth.py")
398
+
399
+
400
+# ── call_tool — search_knowledge ─────────────────────────────────────────────
401
+
402
+class TestCallToolSearchKnowledge:
403
+ def setup_method(self):
404
+ self.fx = _ServerFixture()
405
+
406
+ @pytest.mark.asyncio
407
+ async def test_returns_no_results_message(self):
408
+ result = await self.fx.call_tool_fn("search_knowledge", {"query": "xyz"})
409
+ assert result[0]["text"] == "No results."
410
+
411
+ @pytest.mark.asyncio
412
+ async def test_formats_results(self):
413
+ hit = ContextNode(name="JWT", type="Concept", description="Stateless auth token")
414
+ self.fx.loader.search_knowledge.return_value = [hit]
415
+ result = await self.fx.call_tool_fn("search_knowledge", {"query": "JWT"})
416
+ assert "JWT" in result[0]["text"]
417
+ assert "Concept" in result[0]["text"]
418
+
419
+ @pytest.mark.asyncio
420
+ async def test_passes_limit(self):
421
+ await self.fx.call_tool_fn("search_knowledge", {"query": "auth", "limit": 5})
422
+ self.fx.loader.search_knowledge.assert_called_once_with("auth", limit=5)
423
+
424
+ @pytest.mark.asyncio
425
+ async def test_limit_defaults_to_twenty(self):
426
+ await self.fx.call_tool_fn("search_knowledge", {"query": "auth"})
427
+ self.fx.loader.search_knowledge.assert_called_once_with("auth", limit=20)
428
+
343429
344430
 # ── call_tool — unknown tool ──────────────────────────────────────────────────
345431
346432
 class TestCallToolUnknown:
347433
 def setup_method(self):
348434
--- tests/test_mcp_server.py
+++ tests/test_mcp_server.py
@@ -43,11 +43,14 @@
43 loader = MagicMock(spec=ContextLoader)
44 loader.store = self.store
45 loader.load_file.return_value = _bundle("file_target")
46 loader.load_function.return_value = _bundle("fn_target")
47 loader.load_class.return_value = _bundle("cls_target")
 
48 loader.search.return_value = []
 
 
49 return loader
50
51 def _build(self):
52 list_holder = {}
53 call_holder = {}
@@ -108,13 +111,13 @@
108 class TestListTools:
109 def setup_method(self):
110 self.fx = _ServerFixture()
111
112 @pytest.mark.asyncio
113 async def test_returns_seven_tools(self):
114 tools = await self.fx.list_tools_fn()
115 assert len(tools) == 7
116
117 @pytest.mark.asyncio
118 async def test_tool_names(self):
119 tools = await self.fx.list_tools_fn()
120 names = {t["name"] for t in tools}
@@ -124,10 +127,13 @@
124 "load_function_context",
125 "load_class_context",
126 "search_symbols",
127 "query_graph",
128 "graph_stats",
 
 
 
129 }
130
131 @pytest.mark.asyncio
132 async def test_ingest_repo_requires_path(self):
133 tools = await self.fx.list_tools_fn()
@@ -338,10 +344,90 @@
338 result = await self.fx.call_tool_fn("graph_stats", {})
339 data = json.loads(result[0]["text"])
340 assert data["nodes"] == 42
341 assert data["edges"] == 17
342
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
343
344 # ── call_tool — unknown tool ──────────────────────────────────────────────────
345
346 class TestCallToolUnknown:
347 def setup_method(self):
348
--- tests/test_mcp_server.py
+++ tests/test_mcp_server.py
@@ -43,11 +43,14 @@
43 loader = MagicMock(spec=ContextLoader)
44 loader.store = self.store
45 loader.load_file.return_value = _bundle("file_target")
46 loader.load_function.return_value = _bundle("fn_target")
47 loader.load_class.return_value = _bundle("cls_target")
48 loader.load_decision.return_value = _bundle("decision_target")
49 loader.search.return_value = []
50 loader.find_owners.return_value = []
51 loader.search_knowledge.return_value = []
52 return loader
53
54 def _build(self):
55 list_holder = {}
56 call_holder = {}
@@ -108,13 +111,13 @@
111 class TestListTools:
112 def setup_method(self):
113 self.fx = _ServerFixture()
114
115 @pytest.mark.asyncio
116 async def test_returns_ten_tools(self):
117 tools = await self.fx.list_tools_fn()
118 assert len(tools) == 10
119
120 @pytest.mark.asyncio
121 async def test_tool_names(self):
122 tools = await self.fx.list_tools_fn()
123 names = {t["name"] for t in tools}
@@ -124,10 +127,13 @@
127 "load_function_context",
128 "load_class_context",
129 "search_symbols",
130 "query_graph",
131 "graph_stats",
132 "get_rationale",
133 "find_owners",
134 "search_knowledge",
135 }
136
137 @pytest.mark.asyncio
138 async def test_ingest_repo_requires_path(self):
139 tools = await self.fx.list_tools_fn()
@@ -338,10 +344,90 @@
344 result = await self.fx.call_tool_fn("graph_stats", {})
345 data = json.loads(result[0]["text"])
346 assert data["nodes"] == 42
347 assert data["edges"] == 17
348
349
350 # ── call_tool — unknown tool ──────────────────────────────────────────────────
351
352 # ── call_tool — get_rationale ────────────────────────────────────────────────
353
354 class TestCallToolGetRationale:
355 def setup_method(self):
356 self.fx = _ServerFixture()
357
358 @pytest.mark.asyncio
359 async def test_returns_markdown_by_default(self):
360 result = await self.fx.call_tool_fn("get_rationale", {"name": "Use FalkorDB"})
361 self.fx.loader.load_decision.assert_called_once_with("Use FalkorDB")
362 assert "decision_target" in result[0]["text"]
363
364 @pytest.mark.asyncio
365 async def test_returns_json_when_requested(self):
366 result = await self.fx.call_tool_fn(
367 "get_rationale", {"name": "Use FalkorDB", "format": "json"}
368 )
369 data = json.loads(result[0]["text"])
370 assert data["target"]["name"] == "decision_target"
371
372
373 # ── call_tool — find_owners ──────────────────────────────────────────────────
374
375 class TestCallToolFindOwners:
376 def setup_method(self):
377 self.fx = _ServerFixture()
378
379 @pytest.mark.asyncio
380 async def test_returns_no_owners_message(self):
381 result = await self.fx.call_tool_fn("find_owners", {"name": "AuthService"})
382 assert result[0]["text"] == "No owners found."
383
384 @pytest.mark.asyncio
385 async def test_formats_owners(self):
386 owner = ContextNode(name="Alice", type="Person", description="role=lead, team=auth")
387 self.fx.loader.find_owners.return_value = [owner]
388 result = await self.fx.call_tool_fn("find_owners", {"name": "AuthService"})
389 assert "Alice" in result[0]["text"]
390 assert "role=lead" in result[0]["text"]
391
392 @pytest.mark.asyncio
393 async def test_passes_file_path(self):
394 await self.fx.call_tool_fn(
395 "find_owners", {"name": "AuthService", "file_path": "auth.py"}
396 )
397 self.fx.loader.find_owners.assert_called_once_with("AuthService", file_path="auth.py")
398
399
400 # ── call_tool — search_knowledge ─────────────────────────────────────────────
401
402 class TestCallToolSearchKnowledge:
403 def setup_method(self):
404 self.fx = _ServerFixture()
405
406 @pytest.mark.asyncio
407 async def test_returns_no_results_message(self):
408 result = await self.fx.call_tool_fn("search_knowledge", {"query": "xyz"})
409 assert result[0]["text"] == "No results."
410
411 @pytest.mark.asyncio
412 async def test_formats_results(self):
413 hit = ContextNode(name="JWT", type="Concept", description="Stateless auth token")
414 self.fx.loader.search_knowledge.return_value = [hit]
415 result = await self.fx.call_tool_fn("search_knowledge", {"query": "JWT"})
416 assert "JWT" in result[0]["text"]
417 assert "Concept" in result[0]["text"]
418
419 @pytest.mark.asyncio
420 async def test_passes_limit(self):
421 await self.fx.call_tool_fn("search_knowledge", {"query": "auth", "limit": 5})
422 self.fx.loader.search_knowledge.assert_called_once_with("auth", limit=5)
423
424 @pytest.mark.asyncio
425 async def test_limit_defaults_to_twenty(self):
426 await self.fx.call_tool_fn("search_knowledge", {"query": "auth"})
427 self.fx.loader.search_knowledge.assert_called_once_with("auth", limit=20)
428
429
430 # ── call_tool — unknown tool ──────────────────────────────────────────────────
431
432 class TestCallToolUnknown:
433 def setup_method(self):
434

Keyboard Shortcuts

Open search /
Next entry (timeline) j
Previous entry (timeline) k
Open focused entry Enter
Show this help ?
Toggle theme Top nav button