aaronrene / gabriel-muse public
knowtation.py python
469 lines 18.6 KB
Raw
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 14 days ago
1 """Knowtation MCP namespace — 9 tools, 8 resources, 4 prompts (§KD-5a.B/C/D)."""
2
3 from __future__ import annotations
4
5 import json
6 import pathlib
7 import re
8 from datetime import datetime, timedelta, timezone
9 from typing import Any, Callable
10
11 from muse.core.attestation import (
12 AttestationRequiredError,
13 get_attestation_provider,
14 )
15 from muse.core.store import get_commit_snapshot_manifest, read_commit
16 from muse.core.validation import sanitize_display
17 from muse.mcp.errors import MCPApplicationError
18 from muse.plugins.code._query import walk_commits_bfs
19 from muse.plugins.knowtation.code_intel import note_impact
20 from muse.plugins.knowtation.link_index import build_link_index
21 from muse.plugins.knowtation.mcp_helpers import (
22 PathValidationError,
23 clamp_tool_int,
24 consolidate_receipt,
25 create_hub_proposal,
26 enrich_prime,
27 link_index_to_json,
28 list_vault_notes,
29 projects_summary,
30 read_note_at_ref,
31 recent_memory_events,
32 resolve_ref,
33 validate_note_path,
34 )
35 from muse.plugins.knowtation.note_metrics import make_snapshot_loader, note_dead, note_hotspots
36 from muse.plugins.knowtation.search import hybrid_search
37 from muse.cli.commands.commit import _attestation_path_for_commit
38 from muse.cli.commands.symbol_log import SymbolEvent, _find_events_in_commit
39
40 TOOL_NAMES: tuple[str, ...] = (
41 "knowtation/search",
42 "knowtation/get-note",
43 "knowtation/history",
44 "knowtation/consolidate",
45 "knowtation/impact",
46 "knowtation/dead",
47 "knowtation/prime",
48 "knowtation/propose",
49 "knowtation/attest",
50 )
51
52 RESOURCE_URIS: tuple[str, ...] = (
53 "knowtation://vault",
54 "knowtation://prime",
55 "knowtation://graph",
56 "knowtation://memory/recent",
57 "knowtation://projects",
58 "knowtation://hotspots",
59 )
60
61 PROMPT_NAMES: tuple[str, ...] = (
62 "vault/daily-brief",
63 "vault/search-synthesis",
64 "vault/memory-session",
65 "vault/knowledge-gaps",
66 )
67
68 _VAULT_PATH_RE = re.compile(r"^knowtation://vault/(.+?)(?:@(.+))?$")
69
70
71 def _tool_error(reason: str, message: str, detail: str | None = None) -> MCPApplicationError:
72 return MCPApplicationError(reason, message, detail=detail)
73
74
75 def _parse_path_error(exc: PathValidationError) -> MCPApplicationError:
76 return _tool_error(exc.reason, str(exc))
77
78
79 class KnowtationNamespace:
80 """Registers and dispatches the frozen knowtation MCP surface."""
81
82 def __init__(self, root: pathlib.Path) -> None:
83 self.root = root
84
85 # ------------------------------------------------------------------
86 # Catalogues
87 # ------------------------------------------------------------------
88
89 def list_tools(self) -> list[dict[str, Any]]:
90 return [
91 {"name": n, "description": f"Knowtation domain tool: {n}", "inputSchema": {"type": "object"}}
92 for n in TOOL_NAMES
93 ]
94
95 def list_resources(self) -> list[dict[str, Any]]:
96 static = [
97 {"uri": u, "name": u.replace("knowtation://", ""), "mimeType": "application/json"}
98 for u in RESOURCE_URIS
99 ]
100 static.append(
101 {
102 "uri": "knowtation://vault/{path}",
103 "name": "vault-note",
104 "mimeType": "application/json",
105 "description": "Template: knowtation://vault/{path} or @{ref}",
106 }
107 )
108 return static
109
110 def list_prompts(self) -> list[dict[str, Any]]:
111 return [{"name": n, "description": f"Knowtation vault prompt: {n}"} for n in PROMPT_NAMES]
112
113 # ------------------------------------------------------------------
114 # Dispatch
115 # ------------------------------------------------------------------
116
117 def call_tool(self, name: str, arguments: dict[str, Any]) -> dict[str, Any]:
118 handlers: dict[str, Callable[[dict[str, Any]], dict[str, Any]]] = {
119 "knowtation/search": self._tool_search,
120 "knowtation/get-note": self._tool_get_note,
121 "knowtation/history": self._tool_history,
122 "knowtation/consolidate": self._tool_consolidate,
123 "knowtation/impact": self._tool_impact,
124 "knowtation/dead": self._tool_dead,
125 "knowtation/prime": self._tool_prime,
126 "knowtation/propose": self._tool_propose,
127 "knowtation/attest": self._tool_attest,
128 }
129 handler = handlers.get(name)
130 if handler is None:
131 raise _tool_error("METHOD_NOT_FOUND", f"unknown tool: {name}")
132 return handler(arguments or {})
133
134 def read_resource(self, uri: str) -> dict[str, Any]:
135 if uri == "knowtation://vault":
136 return list_vault_notes(self.root)
137 if uri == "knowtation://prime":
138 return enrich_prime(self.root)
139 if uri == "knowtation://graph":
140 idx = build_link_index(self.root)
141 return link_index_to_json(idx)
142 if uri == "knowtation://memory/recent":
143 return recent_memory_events(self.root)
144 if uri == "knowtation://projects":
145 return projects_summary(self.root)
146 if uri == "knowtation://hotspots":
147 head = resolve_ref(self.root, "HEAD")
148 commits, _ = walk_commits_bfs(self.root, head.commit_id, 500)
149 loader = make_snapshot_loader(self.root)
150 return note_hotspots(commits, loader, top=10)
151
152 m = _VAULT_PATH_RE.match(uri)
153 if m:
154 path_part = m.group(1)
155 ref = m.group(2) or "HEAD"
156 try:
157 path = validate_note_path(path_part)
158 except PathValidationError as exc:
159 raise _parse_path_error(exc) from exc
160 note = read_note_at_ref(self.root, path, ref)
161 if not note.get("exists"):
162 raise _tool_error("NOTE_NOT_FOUND", "note not found at ref")
163 return note
164
165 raise _tool_error("RESOURCE_NOT_FOUND", "unknown resource uri")
166
167 def get_prompt(self, name: str, arguments: dict[str, Any]) -> dict[str, Any]:
168 args = arguments or {}
169 if name == "vault/daily-brief":
170 return self._prompt_daily_brief(args)
171 if name == "vault/search-synthesis":
172 return self._prompt_search_synthesis(args)
173 if name == "vault/memory-session":
174 return self._prompt_memory_session(args)
175 if name == "vault/knowledge-gaps":
176 return self._prompt_knowledge_gaps(args)
177 raise _tool_error("PROMPT_NOT_FOUND", f"unknown prompt: {name}")
178
179 # ------------------------------------------------------------------
180 # Tools
181 # ------------------------------------------------------------------
182
183 def _tool_search(self, args: dict[str, Any]) -> dict[str, Any]:
184 query = args.get("query")
185 if not isinstance(query, str) or not query.strip():
186 raise _tool_error("QUERY_INVALID", "query is required")
187 top_k = clamp_tool_int(args.get("top_k"), 10, 1, 100, "top_k")
188 mode = str(args.get("mode") or "hybrid")
189 if mode not in ("hybrid", "lexical", "semantic"):
190 raise _tool_error("QUERY_INVALID", "invalid mode")
191 ref = str(args.get("ref") or "HEAD")
192 try:
193 resolve_ref(self.root, ref)
194 except PathValidationError as exc:
195 raise _parse_path_error(exc) from exc
196 project = args.get("project")
197 project_s = str(project) if project is not None else None
198 return hybrid_search(
199 self.root,
200 query.strip(),
201 top_k=top_k,
202 mode=mode,
203 project=project_s,
204 )
205
206 def _tool_get_note(self, args: dict[str, Any]) -> dict[str, Any]:
207 path = args.get("path")
208 if not isinstance(path, str):
209 raise _tool_error("PATH_INVALID", "path is required")
210 ref = str(args.get("ref") or "HEAD")
211 fmt = str(args.get("format") or "json")
212 try:
213 validate_note_path(path)
214 resolve_ref(self.root, ref)
215 note = read_note_at_ref(self.root, path, ref, fmt=fmt)
216 if not note.get("exists"):
217 raise _tool_error("NOTE_NOT_FOUND", "note not found at ref")
218 return note
219 except PathValidationError as exc:
220 if exc.reason == "REF_NOT_FOUND":
221 raise _parse_path_error(exc) from exc
222 if exc.reason == "NOT_A_NOTE":
223 raise _parse_path_error(exc) from exc
224 raise _parse_path_error(exc) from exc
225
226 def _tool_history(self, args: dict[str, Any]) -> dict[str, Any]:
227 address = args.get("address")
228 if not isinstance(address, str) or "::" not in address:
229 raise _tool_error("ADDRESS_INVALID", "address must contain '::'")
230 from_ref = str(args.get("from_ref") or "HEAD")
231 max_commits = clamp_tool_int(args.get("max"), 500, 1, 100000, "max")
232 try:
233 start = resolve_ref(self.root, from_ref)
234 except PathValidationError as exc:
235 raise _parse_path_error(exc) from exc
236
237 commits, truncated = walk_commits_bfs(self.root, start.commit_id, max_commits)
238 current = address
239 all_events: list[SymbolEvent] = []
240 for commit in commits:
241 evs, current = _find_events_in_commit(commit, current)
242 all_events.extend(evs)
243
244 return {
245 "address": address,
246 "start_ref": from_ref,
247 "total_commits_scanned": len(commits),
248 "truncated": truncated,
249 "events": [e.to_dict() for e in reversed(all_events)],
250 }
251
252 def _tool_consolidate(self, args: dict[str, Any]) -> dict[str, Any]:
253 return consolidate_receipt(self.root, **args)
254
255 def _tool_impact(self, args: dict[str, Any]) -> dict[str, Any]:
256 path = args.get("path")
257 if not isinstance(path, str):
258 raise _tool_error("PATH_INVALID", "path is required")
259 try:
260 normalised = validate_note_path(path)
261 except PathValidationError as exc:
262 raise _parse_path_error(exc) from exc
263 direction = str(args.get("direction") or "reverse")
264 forward = direction == "forward"
265 depth = args.get("depth", 0)
266 try:
267 depth_i = int(depth)
268 except (TypeError, ValueError):
269 depth_i = 0
270 if depth_i < 0:
271 raise _tool_error("DEPTH_NEGATIVE", "depth must be non-negative")
272 include_attachments = bool(args.get("include_attachments", True))
273 idx = build_link_index(self.root)
274 if normalised not in idx.forward and normalised not in idx.backward:
275 # note may exist without edges — still valid
276 pass
277 try:
278 return note_impact(
279 idx,
280 normalised,
281 forward=forward,
282 depth=depth_i,
283 include_attachments=include_attachments,
284 )
285 except ValueError as exc:
286 raise _tool_error("NOTE_NOT_FOUND", str(exc)) from exc
287
288 def _tool_dead(self, args: dict[str, Any]) -> dict[str, Any]:
289 from muse.plugins.registry import read_domain
290
291 if read_domain(self.root) != "knowtation":
292 raise _tool_error("NOT_KNOWTATION_REPO", "active domain is not knowtation")
293 use_mtime = bool(args.get("use_mtime", True))
294 threshold = clamp_tool_int(args.get("mtime_days_threshold"), 180, 0, 100000, "mtime_days_threshold")
295 idx = build_link_index(self.root)
296 result = note_dead(
297 idx,
298 root=self.root,
299 use_mtime=use_mtime,
300 mtime_days_threshold=threshold,
301 )
302 project = args.get("project")
303 if project:
304 prefix = str(project)
305 filtered = [n for n in result.get("dead_notes", []) if n.get("path", "").startswith(prefix)]
306 result = {**result, "dead_notes": filtered, "dead_count": len(filtered)}
307 return result
308
309 def _tool_prime(self, args: dict[str, Any]) -> dict[str, Any]:
310 max_commits = clamp_tool_int(args.get("max_commits"), 100, 1, 100000, "max_commits")
311 hot_note_count = clamp_tool_int(args.get("hot_note_count"), 10, 1, 100, "hot_note_count")
312 return enrich_prime(
313 self.root,
314 max_commits=max_commits,
315 hot_note_count=hot_note_count,
316 )
317
318 def _tool_propose(self, args: dict[str, Any]) -> dict[str, Any]:
319 path = args.get("path")
320 title = args.get("title")
321 if not isinstance(path, str):
322 raise _tool_error("PATH_INVALID", "path is required")
323 if not isinstance(title, str) or not title.strip():
324 raise _tool_error("VALIDATION_FAILED", "title is required")
325 scope = str(args.get("scope") or "personal")
326 if scope not in ("personal", "project", "org"):
327 raise _tool_error("VALIDATION_FAILED", "invalid scope")
328 try:
329 return create_hub_proposal(
330 self.root,
331 path=path,
332 title=title,
333 proposed_body=args.get("proposed_body") if isinstance(args.get("proposed_body"), str) else None,
334 proposed_frontmatter=args.get("proposed_frontmatter")
335 if isinstance(args.get("proposed_frontmatter"), dict)
336 else None,
337 base_ref=str(args.get("base_ref") or "HEAD"),
338 scope=scope,
339 rationale=str(args.get("rationale")) if args.get("rationale") else None,
340 )
341 except PathValidationError as exc:
342 raise _parse_path_error(exc) from exc
343
344 def _tool_attest(self, args: dict[str, Any]) -> dict[str, Any]:
345 commit_id = args.get("commit_id")
346 if not isinstance(commit_id, str) or not commit_id.strip():
347 raise _tool_error("COMMIT_NOT_FOUND", "commit_id is required")
348
349 provider = get_attestation_provider("knowtation")
350 if provider is None:
351 raise _tool_error("PROVIDER_UNREGISTERED", "attestation provider not registered")
352
353 commit = read_commit(self.root, commit_id.strip())
354 if commit is None:
355 raise _tool_error("COMMIT_NOT_FOUND", "commit not found")
356
357 manifest = get_commit_snapshot_manifest(self.root, commit.commit_id) or {}
358 parent_manifest: dict[str, str] = {}
359 if commit.parent_commit_id:
360 parent_manifest = get_commit_snapshot_manifest(self.root, commit.parent_commit_id) or {}
361
362 action = str(args.get("action") or "write")
363 required_override = args.get("required")
364 cfg_required = bool(getattr(provider, "config", {}).get("required", False))
365 required = bool(required_override) if required_override is not None else cfg_required
366
367 commit_meta: dict[str, object] = {
368 "action": action,
369 "path": _attestation_path_for_commit(manifest, parent_manifest),
370 "content_hash": commit.snapshot_id,
371 "timestamp": commit.committed_at.isoformat(),
372 }
373
374 try:
375 record = provider.compute_attestation(commit.snapshot_id, commit_meta)
376 except AttestationRequiredError as exc:
377 raise _tool_error("ATTESTATION_REQUIRED", str(exc)) from exc
378
379 anchored = False
380 anchor: dict[str, str] | None = None
381 anchor_error: str | None = None
382 try:
383 receipt = provider.anchor(record)
384 anchored = True
385 anchor = {
386 "tx_id": receipt.get("tx_id", ""),
387 "anchor_url": receipt.get("anchor_url", ""),
388 "provider_id": record.get("provider_id", "knowtation"),
389 }
390 except Exception as exc:
391 anchor_error = sanitize_display(str(exc))
392
393 return {
394 "record": record,
395 "anchored": anchored,
396 "anchor": anchor,
397 "anchor_error": anchor_error,
398 "provider_registered": True,
399 }
400
401 # ------------------------------------------------------------------
402 # Prompts
403 # ------------------------------------------------------------------
404
405 def _prompt_daily_brief(self, args: dict[str, Any]) -> dict[str, Any]:
406 since_hours = clamp_tool_int(args.get("since_hours"), 24, 1, 168, "since_hours")
407 cutoff = datetime.now(tz=timezone.utc) - timedelta(hours=since_hours)
408 head = resolve_ref(self.root, "HEAD")
409 commits, _ = walk_commits_bfs(self.root, head.commit_id, 500)
410 lines: list[str] = []
411 for commit in commits:
412 if commit.committed_at < cutoff:
413 continue
414 lines.append(
415 f"- {commit.commit_id[:8]} {commit.committed_at.isoformat()}: "
416 f"{sanitize_display(commit.message)}"
417 )
418 recent = recent_memory_events(self.root)
419 text = (
420 f"Daily brief (last {since_hours}h):\n"
421 + "\n".join(lines[:50])
422 + "\n\nRecent memory events:\n"
423 + json.dumps(recent.get("events", [])[:10], indent=2)
424 )
425 return {"messages": [{"role": "user", "content": {"type": "text", "text": text}}]}
426
427 def _prompt_search_synthesis(self, args: dict[str, Any]) -> dict[str, Any]:
428 query = args.get("query")
429 if not isinstance(query, str) or not query.strip():
430 raise _tool_error("QUERY_INVALID", "query is required")
431 top_k = clamp_tool_int(args.get("top_k"), 5, 1, 20, "top_k")
432 results = hybrid_search(self.root, query.strip(), top_k=top_k, mode="hybrid")
433 text = (
434 f"Synthesise an answer from these vault search results for query: {query!r}\n\n"
435 + json.dumps(results.get("results", []), indent=2)
436 + "\n\nProvide a concise synthesis citing only the listed note paths."
437 )
438 return {"messages": [{"role": "user", "content": {"type": "text", "text": text}}]}
439
440 def _prompt_memory_session(self, args: dict[str, Any]) -> dict[str, Any]:
441 del args
442 prime = enrich_prime(self.root)
443 recent = recent_memory_events(self.root)
444 text = (
445 "Resume this knowtation vault session using prime context and recent memory:\n\n"
446 f"Prime:\n{json.dumps(prime, indent=2)}\n\n"
447 f"Recent memory:\n{json.dumps(recent, indent=2)}"
448 )
449 return {"messages": [{"role": "user", "content": {"type": "text", "text": text}}]}
450
451 def _prompt_knowledge_gaps(self, args: dict[str, Any]) -> dict[str, Any]:
452 del args
453 idx = build_link_index(self.root)
454 dead = note_dead(idx, root=self.root, use_mtime=True)
455 entities = {k: sorted(v) for k, v in idx.entity_index.items()}
456 text = (
457 "Identify knowledge gaps by cross-referencing dead notes and entity coverage:\n\n"
458 f"Dead notes:\n{json.dumps(dead.get('dead_notes', []), indent=2)}\n\n"
459 f"Entity index:\n{json.dumps(entities, indent=2)}"
460 )
461 return {"messages": [{"role": "user", "content": {"type": "text", "text": text}}]}
462
463
464 __all__ = [
465 "KnowtationNamespace",
466 "PROMPT_NAMES",
467 "RESOURCE_URIS",
468 "TOOL_NAMES",
469 ]
File History 1 commit
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 14 days ago