#!/usr/bin/env python3 """ agent.py — Vault Tagger + Embeddings Agent (Step 5) Enriches ONE markdown doc with frontmatter properties (+ optional related links) using the local vault-tagger model (Ollama) and bge-m3 embeddings (TEI). Usage: python3 engine/lib/agent.py --dry-run # print diff, write nothing python3 engine/lib/agent.py # apply if confidence >= threshold Importable: ollama_tag(), tei_embed(), process_doc(), load_config(), load_vocab() stdlib only — no pip installs; all HTTP via urllib. """ from __future__ import annotations import argparse import difflib import json import math import os import re import sys import urllib.error import urllib.request from datetime import datetime, timezone from pathlib import Path # --------------------------------------------------------------------------- # Config loader (minimal stdlib YAML parser — same approach as vocab_gen.py) # --------------------------------------------------------------------------- def load_config(config_path: str | Path | None = None) -> dict: """Load config.yaml; fall back to defaults if file is missing or unparseable.""" defaults: dict = { "vault_dir": "/home/zvx/projects/.ref/vault", "engine_dir": "/home/zvx/projects/.ref/engine", "topic_categories": [ "mesh", "matrix", "recon", "media", "auth", "dns", "vpn", "storage", "proxmox", "ai", "mail", ], "models": { "tagger": { "ollama_endpoint": "http://localhost:11434", "model": "vault-tagger", "temperature": 0.1, "format": "json", }, "embeddings": { "tei_endpoint": "http://localhost:8090", "model": "bge-m3", }, }, "frontmatter_schema": { "required": ["title", "type", "tags", "updated"], "optional": ["aliases", "related", "status"], "types": ["reference", "runbook", "project", "note", "index", "session"], }, "behavior": { "auto_apply": True, "log_changes": True, "changelog": "/home/zvx/projects/.ref/engine/changelog.md", "confidence_threshold": 0.6, }, } if config_path is None: # Search relative to this file's location here = Path(__file__).parent candidates = [ here.parent / "config.yaml", Path("/home/zvx/projects/.ref/engine/config.yaml"), ] for c in candidates: if c.exists(): config_path = c break if config_path is None or not Path(config_path).exists(): return defaults cfg = dict(defaults) try: with open(config_path) as f: lines = f.readlines() for line in lines: m = re.match(r'^(vault_dir|engine_dir):\s*(.+)', line) if m: cfg[m.group(1)] = m.group(2).strip() # topic_categories list in_topics = False topics: list[str] = [] for line in lines: if re.match(r'^topic_categories:', line): in_topics = True continue if in_topics: mm = re.match(r'^\s+-\s+(\S+)', line) if mm: topics.append(mm.group(1).strip()) elif line.strip() and not line.startswith(' '): in_topics = False if topics: cfg["topic_categories"] = topics # models.tagger / models.embeddings (best-effort) in_models = False in_tagger = False in_embed = False in_behavior = False for line in lines: if re.match(r'^models:', line): in_models = True continue if re.match(r'^behavior:', line): in_models = False in_behavior = True continue if in_models: if re.match(r' tagger:', line): in_tagger = True in_embed = False elif re.match(r' embeddings:', line): in_embed = True in_tagger = False elif not line.startswith(' '): in_models = False if in_tagger: mm = re.match(r'\s+ollama_endpoint:\s*(\S+)', line) if mm: cfg["models"]["tagger"]["ollama_endpoint"] = mm.group(1).strip() mm = re.match(r'\s+model:\s*(\S+)', line) if mm: cfg["models"]["tagger"]["model"] = mm.group(1).strip() mm = re.match(r'\s+temperature:\s*([\d.]+)', line) if mm: cfg["models"]["tagger"]["temperature"] = float(mm.group(1)) if in_embed: mm = re.match(r'\s+tei_endpoint:\s*(\S+)', line) if mm: cfg["models"]["embeddings"]["tei_endpoint"] = mm.group(1).strip() if in_behavior: mm = re.match(r'\s+auto_apply:\s*(\S+)', line) if mm: cfg["behavior"]["auto_apply"] = mm.group(1).strip().lower() == 'true' mm = re.match(r'\s+confidence_threshold:\s*([\d.]+)', line) if mm: cfg["behavior"]["confidence_threshold"] = float(mm.group(1)) mm = re.match(r'\s+changelog:\s*(.+)', line) if mm: cfg["behavior"]["changelog"] = mm.group(1).strip() except Exception as e: print(f"[warn] Could not fully parse {config_path}: {e}; using defaults", file=sys.stderr) return cfg # --------------------------------------------------------------------------- # Vocab loader # --------------------------------------------------------------------------- def load_vocab(engine_dir: str | Path) -> dict: """Load vocab.json from engine_dir. Returns dict with topic_categories, entities, acronyms.""" vocab_path = Path(engine_dir) / "vocab.json" if not vocab_path.exists(): print(f"[warn] vocab.json not found at {vocab_path}", file=sys.stderr) return {"topic_categories": [], "entities": [], "acronyms": []} with open(vocab_path) as f: return json.load(f) # --------------------------------------------------------------------------- # Frontmatter parser (stdlib) # --------------------------------------------------------------------------- def parse_frontmatter(text: str) -> tuple[dict, str]: """ Parse YAML frontmatter between --- delimiters. Returns (fm_dict, body_text). Supports: scalars, simple lists (- item format and [a, b, c] inline). """ fm: dict = {} body = text m = re.match(r'^---\r?\n(.*?)\r?\n---\r?\n?(.*)', text, re.DOTALL) if not m: return fm, body fm_raw = m.group(1) body = m.group(2) # Parse line by line lines = fm_raw.splitlines() i = 0 while i < len(lines): line = lines[i] # Skip blank/comment lines if not line.strip() or line.strip().startswith('#'): i += 1 continue # Key: value kv = re.match(r'^(\w[\w-]*):\s*(.*)', line) if not kv: i += 1 continue key = kv.group(1) val_raw = kv.group(2).strip() # Inline list: [a, b, c] if val_raw.startswith('['): inner = re.sub(r'^\[|\]$', '', val_raw).strip() if inner: items = [x.strip().strip('"\'') for x in inner.split(',') if x.strip()] else: items = [] fm[key] = items i += 1 continue # Check if next lines are a multi-line list if not val_raw: lst = [] j = i + 1 while j < len(lines) and re.match(r'^\s+-\s+(.*)', lines[j]): item_m = re.match(r'^\s+-\s+(.*)', lines[j]) lst.append(item_m.group(1).strip().strip('"\'')) j += 1 if lst: fm[key] = lst i = j continue # Scalar val = val_raw.strip('"\'') # Try int/float try: if re.match(r'^\d+$', val): fm[key] = int(val) elif re.match(r'^\d+\.\d+$', val): fm[key] = float(val) else: fm[key] = val except Exception: fm[key] = val i += 1 return fm, body def render_frontmatter(fm: dict) -> str: """Render a frontmatter dict to YAML string (between --- markers).""" lines = ["---"] key_order = ["title", "type", "tags", "aliases", "related", "updated"] written = set() for key in key_order: if key not in fm: continue written.add(key) val = fm[key] if isinstance(val, list): if not val: lines.append(f"{key}: []") else: lines.append(f"{key}:") for item in val: lines.append(f" - {item}") else: # Scalar — quote if contains special chars sv = str(val) if any(c in sv for c in ':#{}[]|>&*!,?'): lines.append(f'{key}: "{sv}"') else: lines.append(f"{key}: {sv}") # Append any extra keys not in key_order (preserve human-set extras) for key, val in fm.items(): if key in written: continue if isinstance(val, list): if not val: lines.append(f"{key}: []") else: lines.append(f"{key}:") for item in val: lines.append(f" - {item}") else: sv = str(val) if any(c in sv for c in ':#{}[]|>&*!,?'): lines.append(f'{key}: "{sv}"') else: lines.append(f"{key}: {sv}") lines.append("---") return "\n".join(lines) + "\n" # --------------------------------------------------------------------------- # HTTP helpers (urllib only) # --------------------------------------------------------------------------- def _http_post(url: str, payload: dict, timeout: int = 120) -> dict | list: """POST JSON to url; return parsed response. Raises on HTTP error.""" data = json.dumps(payload).encode("utf-8") req = urllib.request.Request( url, data=data, headers={"Content-Type": "application/json"}, method="POST", ) with urllib.request.urlopen(req, timeout=timeout) as resp: return json.loads(resp.read().decode("utf-8")) def _http_get(url: str, timeout: int = 10) -> dict: req = urllib.request.Request(url, method="GET") with urllib.request.urlopen(req, timeout=timeout) as resp: return json.loads(resp.read().decode("utf-8")) # --------------------------------------------------------------------------- # Ollama tagger # --------------------------------------------------------------------------- _SYSTEM_PROMPT = """You are the Echo6 vault tagger. Given a markdown document and a controlled vocabulary (topic categories + an entity lexicon), you classify the document and extract structured metadata. You output ONLY valid JSON matching the requested schema. You never invent tags outside the provided vocabulary.""" def _build_vocab_block(vocab: dict) -> str: """Build a compact vocab summary for the prompt.""" cats = vocab.get("topic_categories", []) entities = vocab.get("entities", []) acronyms = vocab.get("acronyms", []) lines = ["## Vocabulary"] lines.append("") lines.append(f"topic_categories (use ONLY these as tags): {', '.join(cats)}") lines.append("") lines.append("### Entities (names + aliases)") for ent in entities: aliases = ent.get("aliases", []) alias_str = f" [{', '.join(str(a) for a in aliases[:3])}]" if aliases else "" lines.append(f"- {ent['name']} ({ent['type']}){alias_str}") lines.append("") lines.append("### Acronyms") for acro in acronyms: exp = acro.get("expansion") or "unknown" lines.append(f"- {acro['acronym']}: {exp}") return "\n".join(lines) def _build_prompt(doc_text: str, vocab: dict) -> str: """Build the full prompt: system + vocab block + doc body.""" vocab_block = _build_vocab_block(vocab) doc_truncated = doc_text[:6000] if len(doc_text) > 6000: doc_truncated += "\n\n[... truncated ...]" return f"""{_SYSTEM_PROMPT} {vocab_block} ## Document {doc_truncated} ## Task Classify this document and return ONLY a JSON object with these fields: - "type": one of reference|runbook|project|note|index|session - "tags": array of strings, each MUST be in topic_categories - "entities": array of entity names from the entity lexicon found in this doc - "glossary_proposals": array of unknown acronyms/terms worth adding - "confidence": float 0.0-1.0 Respond with ONLY the JSON. No prose, no markdown fences.""" def ollama_tag(doc_text: str, vocab: dict, config: dict) -> dict: """ POST to Ollama /api/generate with vault-tagger model. Returns dict with {type, tags, entities, glossary_proposals, confidence}. """ tagger_cfg = config.get("models", {}).get("tagger", {}) endpoint = tagger_cfg.get("ollama_endpoint", "http://localhost:11434") model = tagger_cfg.get("model", "vault-tagger") temperature = tagger_cfg.get("temperature", 0.1) prompt = _build_prompt(doc_text, vocab) payload = { "model": model, "prompt": prompt, "format": "json", "stream": False, "options": {"temperature": temperature}, } url = f"{endpoint}/api/generate" try: resp = _http_post(url, payload, timeout=120) except urllib.error.HTTPError as e: raise RuntimeError(f"Ollama HTTP {e.code}: {e.reason}") from e except Exception as e: raise RuntimeError(f"Ollama request failed: {e}") from e raw_response = resp.get("response", "") if isinstance(resp, dict) else "" # Parse JSON from response (be defensive) try: result = json.loads(raw_response) except json.JSONDecodeError: # Try to extract JSON object from noisy response m = re.search(r'\{.*\}', raw_response, re.DOTALL) if m: try: result = json.loads(m.group(0)) except json.JSONDecodeError: result = {} else: result = {} # Validate and sanitize fields valid_types = {"reference", "runbook", "project", "note", "index", "session"} valid_cats = set(vocab.get("topic_categories", [])) out: dict = { "type": None, "tags": [], "entities": [], "glossary_proposals": [], "confidence": 0.0, } if isinstance(result.get("type"), str) and result["type"] in valid_types: out["type"] = result["type"] if isinstance(result.get("tags"), list): out["tags"] = [t for t in result["tags"] if isinstance(t, str) and t in valid_cats] if isinstance(result.get("entities"), list): out["entities"] = [e for e in result["entities"] if isinstance(e, str)] if isinstance(result.get("glossary_proposals"), list): out["glossary_proposals"] = [p for p in result["glossary_proposals"] if isinstance(p, str)] try: conf = float(result.get("confidence", 0.0)) out["confidence"] = max(0.0, min(1.0, conf)) except (TypeError, ValueError): out["confidence"] = 0.0 return out # --------------------------------------------------------------------------- # TEI embedding # --------------------------------------------------------------------------- def _detect_tei_route(endpoint: str) -> str: """ Detect the correct TEI embed route. Try /embed; if that fails with 404/405 try /embeddings. Returns the working path prefix (e.g. '/embed'). """ for route in ["/embed", "/embeddings"]: try: resp = _http_post(f"{endpoint}{route}", {"inputs": "test"}, timeout=10) if isinstance(resp, list): return route except urllib.error.HTTPError as e: if e.code not in (404, 405): return route # some other error but route exists except Exception: continue return "/embed" # fall back _TEI_ROUTE_CACHE: dict[str, str] = {} def tei_embed(text: str, endpoint: str) -> list[float]: """ POST to TEI bge-m3 embed endpoint; return flat vector. Auto-detects route (/embed vs /embeddings) on first call per endpoint. """ if endpoint not in _TEI_ROUTE_CACHE: _TEI_ROUTE_CACHE[endpoint] = _detect_tei_route(endpoint) route = _TEI_ROUTE_CACHE[endpoint] resp = _http_post(f"{endpoint}{route}", {"inputs": text}, timeout=30) # TEI may return [[vec]] (batch) or [vec] (single) or vec directly if isinstance(resp, list): if resp and isinstance(resp[0], list): return resp[0] # batch of 1 → first element return resp # already flat vector raise RuntimeError(f"Unexpected TEI response type: {type(resp)}") # --------------------------------------------------------------------------- # Embedding cache # --------------------------------------------------------------------------- def _cache_path(engine_dir: Path) -> Path: return engine_dir / ".embcache.json" def load_embed_cache(engine_dir: Path) -> dict: p = _cache_path(engine_dir) if p.exists(): try: with open(p) as f: return json.load(f) except Exception: return {} return {} def save_embed_cache(cache: dict, engine_dir: Path) -> None: p = _cache_path(engine_dir) with open(p, "w") as f: json.dump(cache, f) def _embed_cache_key(rel_path: str, mtime: float) -> str: return f"{rel_path}::{mtime:.3f}" def get_or_embed( doc_path: Path, vault_dir: Path, tei_endpoint: str, cache: dict, doc_text: str | None = None, ) -> tuple[list[float], bool]: """ Return (vector, cache_hit). Reads doc_path if doc_text is None. Updates cache in place (caller must save). """ rel = str(doc_path.relative_to(vault_dir)) mtime = doc_path.stat().st_mtime key = _embed_cache_key(rel, mtime) if key in cache: return cache[key], True if doc_text is None: try: doc_text = doc_path.read_text(errors="replace") except Exception as e: raise RuntimeError(f"Cannot read {doc_path}: {e}") from e # Truncate to ~4000 chars for embedding (TEI handles long but be kind to it) text_for_embed = doc_text[:4000] vec = tei_embed(text_for_embed, tei_endpoint) cache[key] = vec return vec, False # --------------------------------------------------------------------------- # Cosine similarity + related-doc finder # --------------------------------------------------------------------------- def cosine_sim(a: list[float], b: list[float]) -> float: if len(a) != len(b): return 0.0 dot = sum(x * y for x, y in zip(a, b)) mag_a = math.sqrt(sum(x * x for x in a)) mag_b = math.sqrt(sum(x * x for x in b)) if mag_a == 0 or mag_b == 0: return 0.0 return dot / (mag_a * mag_b) def find_related( doc_path: Path, doc_vec: list[float], vault_dir: Path, tei_endpoint: str, cache: dict, k: int = 5, skip_archive: bool = True, ) -> list[str]: """ Embed all other vault docs (using cache), compute cosine similarity, return top-k as [[basename]] wikilinks. """ sims: list[tuple[float, str]] = [] for md in vault_dir.rglob("*.md"): if md == doc_path: continue if skip_archive and "archive" in md.parts: continue try: vec, _ = get_or_embed(md, vault_dir, tei_endpoint, cache) except Exception: continue sim = cosine_sim(doc_vec, vec) sims.append((sim, md.stem)) sims.sort(reverse=True) return [f"[[{stem}]]" for _, stem in sims[:k]] # --------------------------------------------------------------------------- # Entity dictionary match # --------------------------------------------------------------------------- def match_entities(doc_text: str, vocab: dict) -> list[str]: """ Deterministic: find entity names/aliases from vocab that appear in doc_text. Returns list of entity names (canonical). """ body_lower = doc_text.lower() matched: list[str] = [] for ent in vocab.get("entities", []): name = ent.get("name", "") aliases = ent.get("aliases", []) candidates = [name] + [str(a) for a in aliases] for candidate in candidates: if candidate.lower() in body_lower: if name not in matched: matched.append(name) break return matched # --------------------------------------------------------------------------- # Type inference from folder # --------------------------------------------------------------------------- def infer_type_from_path(doc_path: Path, vault_dir: Path) -> str | None: """Infer document type from vault folder path.""" try: rel = doc_path.relative_to(vault_dir) except ValueError: return None parts = rel.parts if not parts: return None folder = parts[0].lower() if len(parts) > 1 else "" filename = doc_path.stem.lower() if filename in ("index", "readme"): return "index" folder_map = { "runbooks": "runbook", "projects": "project", "docs": "reference", "notes": "note", "session-resume": "session", } return folder_map.get(folder) # --------------------------------------------------------------------------- # Title extraction # --------------------------------------------------------------------------- def extract_title(fm: dict, body: str, doc_path: Path) -> str: if fm.get("title"): return str(fm["title"]) # First H1 in body m = re.search(r'^#\s+(.+)', body, re.MULTILINE) if m: return m.group(1).strip() # Prettify filename stem = doc_path.stem return re.sub(r'[-_]+', ' ', stem).title() # --------------------------------------------------------------------------- # Frontmatter merge # --------------------------------------------------------------------------- def merge_frontmatter( existing_fm: dict, tagger_result: dict, matched_entities: list[str], related_links: list[str], doc_path: Path, body: str, vocab: dict, config: dict, ) -> dict: """ Merge tagger results into existing frontmatter. NEVER clobber existing human-set keys; only fill missing / augment. """ valid_types = set( config.get("frontmatter_schema", {}).get("types", ["reference", "runbook", "project", "note", "index", "session"]) ) today = datetime.now(timezone.utc).strftime("%Y-%m-%d") vault_dir = Path(config["vault_dir"]) merged = dict(existing_fm) # title: existing wins if not merged.get("title"): merged["title"] = extract_title(existing_fm, body, doc_path) # type: existing valid value wins; else use tagger; else infer from folder existing_type = str(merged.get("type", "")).strip() if existing_type in valid_types: pass # keep elif tagger_result.get("type") and tagger_result["type"] in valid_types: merged["type"] = tagger_result["type"] else: inferred = infer_type_from_path(doc_path, vault_dir) if inferred: merged["type"] = inferred # tags: tagger tags ∩ topic_categories; merge with existing (union, dedup) valid_cats = set(vocab.get("topic_categories", [])) tagger_tags = [t for t in tagger_result.get("tags", []) if t in valid_cats] existing_tags = merged.get("tags", []) if isinstance(existing_tags, str): existing_tags = [existing_tags] combined_tags = list(dict.fromkeys(existing_tags + tagger_tags)) # preserve order, dedup merged["tags"] = combined_tags # aliases: only add if missing entirely (human-set wins) if "aliases" not in merged: merged["aliases"] = [] # related: tagger-derived embedding links (augment, don't clobber) if related_links and not merged.get("related"): merged["related"] = related_links elif related_links and merged.get("related"): # Augment: add new links not already present existing_related = merged["related"] for link in related_links: if link not in existing_related: existing_related.append(link) merged["related"] = existing_related[:5] # cap at 5 # updated: always set to today merged["updated"] = today return merged # --------------------------------------------------------------------------- # Diff helper # --------------------------------------------------------------------------- def frontmatter_diff(old_text: str, new_text: str, path: Path) -> str: """Return unified diff of old vs new full doc text.""" old_lines = old_text.splitlines(keepends=True) new_lines = new_text.splitlines(keepends=True) return "".join( difflib.unified_diff( old_lines, new_lines, fromfile=f"a/{path.name}", tofile=f"b/{path.name}", n=3, ) ) # --------------------------------------------------------------------------- # Changelog writer # --------------------------------------------------------------------------- def append_changelog( doc_path: Path, old_fm: dict, new_fm: dict, tagger_result: dict, config: dict, applied: bool, ) -> None: changelog = config.get("behavior", {}).get("changelog", "") if not changelog: return ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") confidence = tagger_result.get("confidence", 0.0) action = "applied" if applied else "proposal (below threshold)" old_tags = old_fm.get("tags", []) new_tags = new_fm.get("tags", []) old_type = old_fm.get("type", "") new_type = new_fm.get("type", "") changes = [] if old_type != new_type: changes.append(f"type: {old_type!r} → {new_type!r}") if old_tags != new_tags: changes.append(f"tags: {old_tags} → {new_tags}") if not changes: changes.append("updated date refreshed") entry = ( f"\n## {ts} — {doc_path.name}\n" f"- file: `{doc_path}`\n" f"- action: {action}\n" f"- confidence: {confidence:.2f}\n" f"- changes: {'; '.join(changes)}\n" ) Path(changelog).parent.mkdir(parents=True, exist_ok=True) with open(changelog, "a") as f: f.write(entry) # --------------------------------------------------------------------------- # Core process function # --------------------------------------------------------------------------- def process_doc( doc_path: Path | str, config: dict, vocab: dict, dry_run: bool = True, cache: dict | None = None, ) -> dict: """ Process a single vault document. Returns a result dict with all computed data for reporting/review. In dry_run mode, nothing is written. """ doc_path = Path(doc_path).resolve() vault_dir = Path(config["vault_dir"]).resolve() engine_dir = Path(config["engine_dir"]).resolve() tei_endpoint = config["models"]["embeddings"]["tei_endpoint"] threshold = config["behavior"]["confidence_threshold"] if cache is None: cache = load_embed_cache(engine_dir) # --- Read doc --- try: doc_text = doc_path.read_text(errors="replace") except Exception as e: raise RuntimeError(f"Cannot read {doc_path}: {e}") from e existing_fm, body = parse_frontmatter(doc_text) # --- Tagger --- print(f"[tagger] Calling vault-tagger on {doc_path.name} ...", file=sys.stderr) tagger_result = ollama_tag(doc_text, vocab, config) print(f"[tagger] confidence={tagger_result['confidence']:.2f} type={tagger_result['type']} tags={tagger_result['tags']}", file=sys.stderr) # --- Entity match (deterministic) --- matched_entities = match_entities(doc_text, vocab) # --- Embeddings + related links --- print(f"[embed] Embedding {doc_path.name} ...", file=sys.stderr) try: doc_vec, cache_hit = get_or_embed(doc_path, vault_dir, tei_endpoint, cache, doc_text=doc_text) if not cache_hit: save_embed_cache(cache, engine_dir) print(f"[embed] Vector dim={len(doc_vec)} cache_hit={cache_hit}", file=sys.stderr) print(f"[embed] Finding related docs (scanning vault) ...", file=sys.stderr) related_links = find_related(doc_path, doc_vec, vault_dir, tei_endpoint, cache, k=5) save_embed_cache(cache, engine_dir) # save updated cache print(f"[embed] Found {len(related_links)} related docs", file=sys.stderr) except Exception as e: print(f"[warn] Embedding/related failed: {e}", file=sys.stderr) doc_vec = [] related_links = [] # --- Merge frontmatter --- new_fm = merge_frontmatter( existing_fm, tagger_result, matched_entities, related_links, doc_path, body, vocab, config, ) # --- Build new doc text --- new_fm_text = render_frontmatter(new_fm) new_doc_text = new_fm_text + body # --- Diff --- diff = frontmatter_diff(doc_text, new_doc_text, doc_path) # --- Apply or log --- applied = False if not dry_run: auto_apply = config["behavior"].get("auto_apply", True) if auto_apply and tagger_result["confidence"] >= threshold: doc_path.write_text(new_doc_text) applied = True print(f"[apply] Wrote {doc_path}", file=sys.stderr) else: print( f"[skip] confidence={tagger_result['confidence']:.2f} < threshold={threshold} " f"or auto_apply=False — logged as proposal", file=sys.stderr, ) if config["behavior"].get("log_changes", True): append_changelog(doc_path, existing_fm, new_fm, tagger_result, config, applied) return { "path": str(doc_path), "tagger_result": tagger_result, "matched_entities": matched_entities, "related_links": related_links, "tei_route": _TEI_ROUTE_CACHE.get(tei_endpoint, "(not called)"), "proposed_frontmatter": new_fm_text, "diff": diff, "applied": applied, "dry_run": dry_run, "existing_fm": existing_fm, "new_fm": new_fm, } # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def main() -> None: parser = argparse.ArgumentParser( description="Vault tagger agent — enrich a markdown doc with frontmatter." ) parser.add_argument("path", help="Path to the vault markdown document") parser.add_argument( "--dry-run", action="store_true", help="Print tagger JSON, matched entities, proposed related, diff — write nothing", ) args = parser.parse_args() # Load config here = Path(__file__).parent config_path = here.parent / "config.yaml" if not config_path.exists(): config_path = Path("/home/zvx/projects/.ref/engine/config.yaml") config = load_config(config_path) vocab = load_vocab(config["engine_dir"]) doc_path = Path(args.path).resolve() if not doc_path.exists(): print(f"[error] File not found: {doc_path}", file=sys.stderr) sys.exit(1) result = process_doc(doc_path, config, vocab, dry_run=args.dry_run) # --- Print report --- print("\n" + "=" * 70) print(f"TAGGER AGENT {'DRY-RUN ' if args.dry_run else ''}REPORT: {doc_path.name}") print("=" * 70) print("\n## Raw tagger JSON:") print(json.dumps(result["tagger_result"], indent=2)) print("\n## Matched entities (deterministic):") if result["matched_entities"]: for e in result["matched_entities"]: print(f" - {e}") else: print(" (none)") print("\n## Proposed related links (top-5 by cosine similarity):") if result["related_links"]: for link in result["related_links"]: print(f" {link}") else: print(" (none — embedding not available)") print(f"\n## TEI route used: {result['tei_route']}") print("\n## Proposed frontmatter block:") print(result["proposed_frontmatter"]) if result["diff"]: print("## Unified diff vs current:") print(result["diff"]) else: print("## Unified diff: (no changes)") if args.dry_run: print("\n[DRY-RUN] No files were modified.") else: status = "applied" if result["applied"] else "not applied (below threshold or auto_apply=False)" print(f"\n[APPLY] Status: {status}") if __name__ == "__main__": main()