"""Markdown-based hierarchical knowledge store. Stores classified documents as human-readable Markdown files with YAML frontmatter. Maintains hierarchy.md tree for agent navigation (per ByteRover and NeuRAG research). Storage layout: knowledge_docs/ ├── _index.md # root overview tree ├── ai/ │ ├── security/ │ │ └── audit_protocols.md │ └── architecture/ │ └── design.md ├── research/ ├── notes/ # uncategorized / low-confidence └── reports/ Frontmatter fields: topic, subtopic, tags[], summary, confidence (0-1), entities[], intent ("actionable"|"reference"|"idea"), created, updated, sha256 Research references: - ByteRover (arXiv:2604.01599) — Context Tree as markdown hierarchy - NeuRAG — trigger-navigable hierarchy via markdown structure - secondmem — markdown files with tags for agent consumption """ from __future__ import annotations import hashlib import logging import time from dataclasses import dataclass, field from pathlib import Path from typing import Any try: import yaml except ImportError: # pragma: no cover - yaml optional for basic operation yaml = None from cyberbrain26.knowledge_plane.classifier import ClassificationResult logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Hierarchy node # --------------------------------------------------------------------------- @dataclass(frozen=True) class HierarchyNode: """A node in the knowledge hierarchy tree. Used by hierarchy.md generation and navigation. """ name: str path: str level: int # 0 = root, 1 = type, 2 = subtopic, 3 = document document_count: int tags: list[str] = field(default_factory=list) # --------------------------------------------------------------------------- # Markdown store implementation # --------------------------------------------------------------------------- class MarkdownStore: """Hierarchical Markdown knowledge store. Writes classified documents as Markdown with YAML frontmatter. Maintains hierarchy.md tree for agent navigation. Research-backed design: - Hierarchical folder structure (per ByteRover Context Tree) - YAML frontmatter for structured metadata - SHA-256 dedup gate (per Grimore-MD two-tier change detection) - hierarchy.md maintained in sync with filesystem tree """ def __init__(self, root: Path | str, auto_classify: bool = True) -> None: self.root = Path(root).expanduser().resolve() self.root.mkdir(parents=True, exist_ok=True) self.auto_classify = auto_classify self._index_path = self.root / "_index.md" self._document_index: dict[str, dict[str, Any]] = {} self._rebuild_index() # ------------------------------------------------------------------ # Internal: index rebuild from filesystem # ------------------------------------------------------------------ def _rebuild_index(self) -> None: """Rebuild internal document index from existing markdown files.""" self._document_index.clear() if not self.root.exists(): return for md_file in self.root.rglob("*.md"): if md_file.name == "_index.md": continue try: doc_meta = self._read_frontmatter(md_file) if doc_meta: self._document_index[md_file.stem] = { "path": str(md_file.relative_to(self.root)), "metadata": doc_meta, } except Exception as exc: # noqa: BLE001 - skip broken files logger.warning("Skipping broken markdown file %s: %s", md_file, exc) # ------------------------------------------------------------------ # Frontmatter read/write helpers # ------------------------------------------------------------------ def _read_frontmatter(self, path: Path) -> dict[str, Any] | None: """Read YAML frontmatter from a Markdown file.""" content = path.read_text(encoding="utf-8") if not content.startswith("---\n"): return None end_idx = content.find("\n---\n", 4) if end_idx == -1: return None frontmatter_text = content[4:end_idx] if yaml is not None: try: return yaml.safe_load(frontmatter_text) or {} except yaml.YAMLError as exc: logger.warning("YAML parse error in %s: %s", path, exc) return None # Minimal YAML-free fallback: parse key: value lines result: dict[str, Any] = {} for line in frontmatter_text.splitlines(): if ":" in line: key, _, value = line.partition(":") key = key.strip() value = value.strip().strip('"').strip("'") if value.startswith("[") and value.endswith("]"): # Simple list parsing inner = value[1:-1].strip() if inner: result[key] = [x.strip().strip('"').strip("'") for x in inner.split(",")] else: result[key] = [] else: result[key] = value return result def _write_document( self, path: Path, content: str, frontmatter: dict[str, Any], ) -> None: """Write a Markdown document with YAML frontmatter.""" path.parent.mkdir(parents=True, exist_ok=True) lines: list[str] = ["---"] if yaml is not None: yaml_text = yaml.dump(frontmatter, default_flow_style=False, allow_unicode=True).strip() lines.append(yaml_text) else: for key, value in frontmatter.items(): if isinstance(value, list): items = ", ".join(f'"{v}"' for v in value) lines.append(f'{key}: [{items}]') else: lines.append(f'{key}: "{value}"') lines.append("---") lines.append("") lines.append(content) path.write_text("\n".join(lines) + "\n", encoding="utf-8") logger.debug("Wrote markdown document: %s", path) # ------------------------------------------------------------------ # SHA-256 hash gate (incremental processing) # ------------------------------------------------------------------ def _file_sha256(self, path: Path) -> str: """Return SHA-256 hex digest of file contents.""" h = hashlib.sha256() with path.open("rb") as f: for chunk in iter(lambda: f.read(65536), b""): h.update(chunk) return h.hexdigest() def should_process(self, filepath: Path) -> bool: """Check if a file should be processed (not already stored with same hash).""" h = self._file_sha256(filepath) for doc_path in self.root.rglob("*.md"): if doc_path.name == "_index.md": continue try: doc_meta = self._read_frontmatter(doc_path) if doc_meta and doc_meta.get("sha256") == h: return False # duplicate hash found except Exception: # noqa: BLE001 continue return True # ------------------------------------------------------------------ # Document storage with classification # ------------------------------------------------------------------ def store_classified( self, classification: ClassificationResult, content: str, source_uri: str, source_id: str, filename: str | None = None, ) -> Path: """Store a classified document as Markdown with frontmatter. Writes to: {root}/{topic}/{subtopic}/{slug}.md Returns the path to the written file. """ # Build slug from filename or first words of summary if filename: stem = Path(filename).stem else: stem = classification.summary[:30].strip().replace(" ", "_") # Sanitize slug slug = "".join(c.lower() if c.isalnum() or c in "-_" else "-" for c in stem).strip("-") # Determine folder path topic_dir = self.root / classification.topic subtopic_dir = topic_dir / classification.subtopic target_path = subtopic_dir / f"{slug}.md" # Build frontmatter now = time.time() frontmatter: dict[str, Any] = { "topic": classification.topic, "subtopic": classification.subtopic, "tags": classification.tags, "summary": classification.summary, "confidence": classification.confidence, "entities": classification.entities, "intent": classification.intent, "created": now, "updated": now, "sha256": hashlib.sha256(content.encode("utf-8")).hexdigest(), "source_id": source_id, "source_uri": source_uri, } self._write_document(target_path, content, frontmatter) # Rebuild internal index self._document_index[slug] = { "path": str(target_path.relative_to(self.root)), "metadata": frontmatter, } # Update hierarchy.md self._rebuild_hierarchy() return target_path # ------------------------------------------------------------------ # Hierarchy maintenance # ------------------------------------------------------------------ def _rebuild_hierarchy(self) -> None: """Rebuild _index.md hierarchy tree from current documents.""" lines: list[str] = ["# Knowledge Hierarchy", ""] lines.append(f"Root: `{self.root.name}`") lines.append("") # Walk filesystem tree topics: dict[str, dict[str, list[str]]] = {} for md_file in self.root.rglob("*.md"): if md_file.name == "_index.md": continue doc_meta = self._read_frontmatter(md_file) if not doc_meta: continue topic = doc_meta.get("topic", "general") subtopic = doc_meta.get("subtopic", "general") doc_name = md_file.name topics.setdefault(topic, {}).setdefault(subtopic, []).append(doc_name) for topic in sorted(topics): lines.append(f"## {topic}") for subtopic in sorted(topics[topic]): lines.append(f" - **{subtopic}**") for doc in sorted(topics[topic][subtopic]): lines.append(f" - {doc}") lines.append("") self._index_path.write_text("\n".join(lines) + "\n", encoding="utf-8") logger.debug("Rebuilt hierarchy.md with %d topics", len(topics)) # ------------------------------------------------------------------ # Document lookup # ------------------------------------------------------------------ def list_documents(self, topic: str | None = None) -> list[dict[str, Any]]: """List all documents, optionally filtered by topic.""" results = [] for md_file in self.root.rglob("*.md"): if md_file.name == "_index.md": continue doc_meta = self._read_frontmatter(md_file) if not doc_meta: continue if topic and doc_meta.get("topic") != topic: continue results.append( { "path": str(md_file.relative_to(self.root)), "metadata": doc_meta, } ) return results def get_document(self, relative_path: str) -> dict[str, Any] | None: """Read a document by its relative path from root.""" full_path = self.root / relative_path if not full_path.exists(): return None content = full_path.read_text(encoding="utf-8") frontmatter = self._read_frontmatter(full_path) or {} # Extract body after frontmatter body = content if content.startswith("---\n"): end_idx = content.find("\n---\n", 4) if end_idx != -1: body = content[end_idx + 5:] return {"content": body.strip(), "metadata": frontmatter, "path": relative_path} def search_by_tag(self, tag: str) -> list[dict[str, Any]]: """Search documents by tag (simple containment match).""" results = [] for md_file in self.root.rglob("*.md"): if md_file.name == "_index.md": continue doc_meta = self._read_frontmatter(md_file) if not doc_meta: continue tags = doc_meta.get("tags", []) if isinstance(tags, str): tags = [t.strip() for t in tags.split(",")] if tag.lower() in [t.lower() for t in tags]: results.append({"path": str(md_file.relative_to(self.root)), "metadata": doc_meta}) return results def search_by_topic(self, topic: str) -> list[dict[str, Any]]: """Search documents by topic.""" return self.list_documents(topic=topic) def close(self) -> None: """No-op — MarkdownStore holds no persistent connections.""" pass def __enter__(self) -> MarkdownStore: return self def __exit__(self, *_exc: object) -> None: self.close()