diff --git a/config.yaml.example b/config.yaml.example index 47d2fda2b..c19e4e7c6 100644 --- a/config.yaml.example +++ b/config.yaml.example @@ -27,6 +27,14 @@ pageindex_threshold: 20 # PDF pages threshold for PageIndex # - dataset # - model +# Optional: how concept/entity pages absorb new documents on `openkb add`. +# rewrite default — send the existing page's full body to the LLM and +# let it rewrite the whole page to incorporate the new document. +# append cheaper for large/mature wikis — instead of a full rewrite, +# generate a short note about the new document and append it to +# the page under a "## Notes" heading (no full-page LLM rewrite). +# concept_update_mode: rewrite + # Optional: LLM / LiteLLM tuning. Keys are forwarded to LiteLLM; `timeout` and # `extra_headers` apply per request, the rest are set as litellm.. # litellm: diff --git a/openkb/agent/compiler.py b/openkb/agent/compiler.py index d0c9f878d..33463b3f1 100644 --- a/openkb/agent/compiler.py +++ b/openkb/agent/compiler.py @@ -30,6 +30,7 @@ import litellm from openkb import frontmatter +from openkb.agent import compiler_notes from openkb.config import ( DEFAULT_ENTITY_TYPES, get_extra_headers, @@ -1334,6 +1335,8 @@ def _remove_doc_from_pages( ``## Related Documents`` section. - Remove any standalone ``See also: [[summaries/{doc_name}]]`` lines (left by ``_add_related_link``). + - Remove this doc's ``## Notes`` line, if any (left by + ``concept_update_mode="append"``; a no-op for "rewrite"-mode pages). - If the ``sources:`` list becomes empty AND ``keep_empty`` is False, delete the page entirely. @@ -1390,6 +1393,17 @@ def _remove_doc_from_pages( flags=re.MULTILINE, ) + # Drop this doc's "## Notes" line (left by + # ``compiler_notes.append_concept_note``/``append_entity_note`` under + # ``concept_update_mode="append"``) — a no-op on "rewrite"-mode pages, + # which never contain this line shape. + new_text = re.sub( + rf"^- \*\*.*\(\[\[{re.escape(bare_source)}\]\]\)[ \t]*\n?", + "", + new_text, + flags=re.MULTILINE, + ) + if sources_empty and not keep_empty: path.unlink() deleted.append(path.stem) @@ -1603,6 +1617,7 @@ async def _compile_concepts( doc_type: str = "short", rewrite_summary: bool = False, entity_types: list[str] | None = None, + concept_update_mode: str = "rewrite", bundle=None, ) -> None: """Shared Steps 2-4: concepts plan → generate/update → index. @@ -1613,6 +1628,14 @@ async def _compile_concepts( written to disk. When ``rewrite_summary=True`` (short-doc path), the summary is rewritten by the LLM after concepts are finalized so its wikilinks reflect the actual concept pages on disk. + + ``concept_update_mode`` (see ``openkb.config.resolve_concept_update_mode``) + controls how EXISTING concept/entity pages absorb this document: the + default ``"rewrite"`` sends the full page back to the LLM for a rewrite; + ``"append"`` generates a short note instead (the LLM never sees the + existing page) and appends it via ``openkb.agent.compiler_notes`` — no + LLM call for the write itself. New pages are generated the same way in + both modes for "rewrite" (full content) vs. a note for "append". """ source_file = f"summaries/{doc_name}.md" @@ -1963,17 +1986,140 @@ async def _gen_entity_update(ent: dict) -> tuple[str, str, str, str]: _require_nonempty_content(content, name) return name, content, brief, etype_out + # --- "append" mode closures: a short note instead of a full-page rewrite. + # The LLM never sees the existing page (no existing_content read, no + # known_targets_msg turn — notes stay plain text, see compiler_notes.py). + # Return shapes are IDENTICAL to the four closures above (name, + # content-or-note, is_update-or-brief, brief-or-type), so every downstream + # step (gather, ghost-link stripping, index bookkeeping) is shared between + # modes — only the final disk write branches (see below). + async def _gen_note_create(concept: dict) -> tuple[str, str, bool, str]: + name = concept["name"] + title = concept.get("title", name) + async with semaphore: + raw = await _llm_call_page_async( + model, + [ + system_msg, + doc_msg, # cached (BP1) + summary_msg, # cached (BP2) + { + "role": "user", + "content": compiler_notes._CONCEPT_NOTE_CREATE_USER.format( + title=title, + doc_name=doc_name, + ), + }, + ], + f"concept-note: {name}", + response_format=_JSON_RESPONSE_FORMAT, + bundle=bundle, + ) + description, note = compiler_notes.note_fields(raw) + _require_nonempty_content(note, name) + return name, note, False, description + + async def _gen_note_update(concept: dict) -> tuple[str, str, bool, str]: + name = concept["name"] + title = concept.get("title", name) + async with semaphore: + raw = await _llm_call_page_async( + model, + [ + system_msg, + doc_msg, # cached (BP1) + summary_msg, # cached (BP2) + { + "role": "user", + "content": compiler_notes._CONCEPT_NOTE_UPDATE_USER.format( + title=title, + doc_name=doc_name, + ), + }, + ], + f"concept-note-update: {name}", + response_format=_JSON_RESPONSE_FORMAT, + bundle=bundle, + ) + _, note = compiler_notes.note_fields(raw) + _require_nonempty_content(note, name) + return name, note, True, "" + + async def _gen_entity_note_create(ent: dict) -> tuple[str, str, str, str]: + name = ent["name"] + title = ent.get("title", name) + etype = ent.get("type", "other") + async with semaphore: + raw = await _llm_call_page_async( + model, + [ + system_msg, + doc_msg, # cached (BP1) + summary_msg, # cached (BP2) + { + "role": "user", + "content": compiler_notes._ENTITY_NOTE_CREATE_USER.format( + title=title, + type=etype, + doc_name=doc_name, + ), + }, + ], + f"entity-note: {name}", + response_format=_JSON_RESPONSE_FORMAT, + bundle=bundle, + ) + description, note = compiler_notes.note_fields(raw) + _require_nonempty_content(note, name) + return name, note, description, etype + + async def _gen_entity_note_update(ent: dict) -> tuple[str, str, str, str]: + name = ent["name"] + title = ent.get("title", name) + etype = ent.get("type", "other") + async with semaphore: + raw = await _llm_call_page_async( + model, + [ + system_msg, + doc_msg, # cached (BP1) + summary_msg, # cached (BP2) + { + "role": "user", + "content": compiler_notes._ENTITY_NOTE_UPDATE_USER.format( + title=title, + type=etype, + doc_name=doc_name, + ), + }, + ], + f"entity-note-update: {name}", + response_format=_JSON_RESPONSE_FORMAT, + bundle=bundle, + ) + _, note = compiler_notes.note_fields(raw) + _require_nonempty_content(note, name) + return name, note, "", etype + tasks = [] - tasks.extend(_gen_create(c) for c in create_items) - tasks.extend(_gen_update(c) for c in update_items) + if concept_update_mode == "append": + tasks.extend(_gen_note_create(c) for c in create_items) + tasks.extend(_gen_note_update(c) for c in update_items) + else: + tasks.extend(_gen_create(c) for c in create_items) + tasks.extend(_gen_update(c) for c in update_items) # --- Step 3 (entities): build the entity task list up front so it can be # gathered concurrently with the concept tasks below. Entity coroutines # return 4-arity tuples (name, content, brief, type), so their results are # processed in their own loop rather than mixed with the concept tuples. entity_tasks = [] - entity_tasks.extend(_gen_entity_create(e) for e in entity_create) - entity_tasks.extend(_gen_entity_update(e) for e in entity_update) + if concept_update_mode == "append": + entity_tasks.extend(_gen_entity_note_create(e) for e in entity_create) + entity_tasks.extend(_gen_entity_note_update(e) for e in entity_update) + else: + entity_tasks.extend(_gen_entity_create(e) for e in entity_create) + entity_tasks.extend(_gen_entity_update(e) for e in entity_update) concept_names: list[str] = [] concept_briefs_map: dict[str, str] = {} @@ -2063,7 +2209,12 @@ async def _gen_entity_update(ent: dict) -> tuple[str, str, str, str]: ) safe = _sanitize_concept_name(name) is_update = (wiki_dir / "entities" / f"{safe}.md").exists() - _write_entity(wiki_dir, name, cleaned, source_file, is_update, brief=brief, type_=etype) + if concept_update_mode == "append": + compiler_notes.append_entity_note( + wiki_dir, name, cleaned, source_file, doc_name, description=brief, type_=etype + ) + else: + _write_entity(wiki_dir, name, cleaned, source_file, is_update, brief=brief, type_=etype) entity_names.append(safe) entity_meta[safe] = (etype, brief) @@ -2154,14 +2305,19 @@ async def _gen_entity_update(ent: dict) -> tuple[str, str, str, str]: # --- Write concept pages to disk --- for name, page_content, is_update, brief in pending_writes: - _write_concept( - wiki_dir, - name, - page_content, - source_file, - is_update, - brief=brief, - ) + if concept_update_mode == "append": + compiler_notes.append_concept_note( + wiki_dir, name, page_content, source_file, doc_name, description=brief + ) + else: + _write_concept( + wiki_dir, + name, + page_content, + source_file, + is_update, + brief=brief, + ) # --- Step 3b: Process related items (code only, no LLM) --- sanitized_related = [_sanitize_concept_name(s) for s in related_items] @@ -2215,7 +2371,7 @@ async def compile_short_doc( Step 1: Build base context A (schema + doc content), generate summary. Steps 2-4: Delegated to ``_compile_concepts``. """ - from openkb.config import resolve_effective_config + from openkb.config import resolve_concept_update_mode, resolve_effective_config config = resolve_effective_config(kb_dir)[0] language: str = config.get("language", "en") @@ -2280,6 +2436,7 @@ async def compile_short_doc( doc_type="short", rewrite_summary=True, entity_types=entity_types, + concept_update_mode=resolve_concept_update_mode(config), bundle=bundle, ) finally: @@ -2303,7 +2460,7 @@ async def compile_long_doc( The summary page is already written by the indexer. This function generates concept pages and updates the index. """ - from openkb.config import resolve_effective_config + from openkb.config import resolve_concept_update_mode, resolve_effective_config config = resolve_effective_config(kb_dir)[0] language: str = config.get("language", "en") @@ -2364,6 +2521,7 @@ async def compile_long_doc( doc_brief=doc_description, doc_type="pageindex", entity_types=entity_types, + concept_update_mode=resolve_concept_update_mode(config), bundle=bundle, ) finally: diff --git a/openkb/agent/compiler_notes.py b/openkb/agent/compiler_notes.py new file mode 100644 index 000000000..7b2df2c5e --- /dev/null +++ b/openkb/agent/compiler_notes.py @@ -0,0 +1,257 @@ +"""Append-only note ingest for OpenKB's wiki compiler (``concept_update_mode="append"``). + +Companion to ``openkb.agent.compiler``: instead of sending an existing concept/ +entity page's full body back to the LLM for a rewrite (``_gen_update``/ +``_gen_entity_update`` in ``compiler.py``), this module generates a short note +about the new document — the LLM never sees the existing page — and appends +it deterministically (no LLM call for the write itself) as a dated, +source-linked line under a ``## Notes`` heading in the same page. Reconciling +the accumulated notes back into curated prose is a separate, future concern — +out of scope here. +""" + +from __future__ import annotations + +import datetime +import json +import logging +import re +from pathlib import Path + +from openkb import frontmatter +from openkb.locks import atomic_write_text + +logger = logging.getLogger(__name__) + +_NOTES_HEADING = "## Notes" + +# --------------------------------------------------------------------------- +# Prompt templates — deliberately slim: no existing page content, no wikilink +# whitelist (notes stay plain text; see module docstring / SKILL.md for why). +# --------------------------------------------------------------------------- + +_CONCEPT_NOTE_CREATE_USER = """\ +This is a NEW concept page: {title} + +This concept was just identified in document "{doc_name}" (summarized above). + +Return a JSON object with two keys: +- "description": A single sentence (under 100 chars) defining this concept +- "note": Some short sentences (or just a few keywords if that's enough) \ +capturing what THIS document says about {title} — it will be appended to a \ +running list of notes, not written as prose. Do NOT use [[wikilinks]]. + +Return ONLY valid JSON, no fences. +""" + +_CONCEPT_NOTE_UPDATE_USER = """\ +Concept page: {title} + +Document "{doc_name}" (summarized above) mentions this concept. + +Return a JSON object with one key: +- "note": Some short sentences (or just a few keywords if that's enough) \ +capturing what THIS document adds about {title} — it will be appended to a \ +running list of notes, not merged into the existing page (which you do not \ +see). Do NOT use [[wikilinks]]. + +Return ONLY valid JSON, no fences. +""" + +_ENTITY_NOTE_CREATE_USER = """\ +This is a NEW entity page: {title} (type: {type}) + +This entity was just identified in document "{doc_name}" (summarized above). + +Return a JSON object with two keys: +- "description": A single sentence (under 100 chars) identifying this entity +- "note": Some short sentences (or just a few keywords if that's enough) \ +capturing what THIS document says about {title} — it will be appended to a \ +running list of notes, not written as prose. Do NOT use [[wikilinks]]. + +Return ONLY valid JSON, no fences. +""" + +_ENTITY_NOTE_UPDATE_USER = """\ +Entity page: {title} (type: {type}) + +Document "{doc_name}" (summarized above) mentions this entity. + +Return a JSON object with one key: +- "note": Some short sentences (or just a few keywords if that's enough) \ +capturing what THIS document adds about {title} — it will be appended to a \ +running list of notes, not merged into the existing page (which you do not \ +see). Do NOT use [[wikilinks]]. + +Return ONLY valid JSON, no fences. +""" + + +def note_fields(raw: str) -> tuple[str, str]: + """Map a note LLM response to ``(description, note)``. + + Mirrors ``compiler._page_fields`` for the smaller note shape: not-JSON + responses fall back to using the raw text as the note itself (a model + that ignores the JSON instruction still produces a usable short note). + """ + from openkb.agent import compiler as _compiler # local: avoid import cycle + + try: + obj = _compiler._parse_page_json(raw) + except (json.JSONDecodeError, ValueError): + return "", raw.strip() + if obj is None: + return "", "" + return obj.get("description", ""), (obj.get("note") or "").strip() + + +def _upsert_note_line(body: str, doc_name: str, note: str, heading: str = _NOTES_HEADING) -> str: + """Insert/replace the note line for ``doc_name`` right after ``heading``. + + Keyed by the ``[[summaries/{doc_name}]]`` source marker (not the note + text), so re-ingesting an updated version of the same document replaces + its own line instead of accumulating duplicates. New/replaced lines land + directly after the heading — newest first, matching the ``sources:`` + frontmatter convention. + """ + marker = f"[[summaries/{doc_name}]]" + date = datetime.date.today().isoformat() + line = f"- **{date}** {note} ({marker})" + + lines = body.split("\n") + line_re = re.compile(rf"^- \*\*.*\({re.escape(marker)}\)\s*$") + lines = [ln for ln in lines if not line_re.match(ln)] + + heading_idx = next((i for i, ln in enumerate(lines) if ln.strip() == heading), None) + if heading_idx is None: + while lines and lines[-1].strip() == "": + lines.pop() + if lines: + lines.append("") + lines.append(heading) + lines.append("") + lines.append(line) + else: + insert_at = heading_idx + 1 + if insert_at < len(lines) and lines[insert_at].strip() == "": + insert_at += 1 + lines.insert(insert_at, line) + + return "\n".join(lines) + + +def _build_frontmatter(fm_lines: list[str]) -> str: + """Build a fresh frontmatter block (delimiters + trailing blank line).""" + return "---\n" + "\n".join(fm_lines) + "\n---\n\n" + + +def append_concept_note( + wiki_dir: Path, + name: str, + note: str, + source_file: str, + doc_name: str, + description: str = "", +) -> None: + """Append a short note about ``doc_name`` to a concept page (no LLM write). + + Creates the page (with ``description`` in its frontmatter, if given) when + it doesn't exist yet; on an existing page, only ``sources:`` is updated + and a note line is upserted — ``description`` is never touched once set. + """ + from openkb.agent import compiler as _compiler # local: avoid import cycle + + concepts_dir = wiki_dir / "concepts" + concepts_dir.mkdir(parents=True, exist_ok=True) + safe_name = _compiler._sanitize_concept_name(name) + path = (concepts_dir / f"{safe_name}.md").resolve() + if not path.is_relative_to(concepts_dir.resolve()): + logger.warning("Concept name escapes concepts dir: %s", name) + return + + if path.exists(): + existing = path.read_text(encoding="utf-8") + if source_file not in existing: + existing = _compiler._prepend_source_to_frontmatter(existing, source_file) + parts = frontmatter.split(existing) + if parts is not None: + fm_block, body = parts + new_body = _upsert_note_line(body.lstrip("\n"), doc_name, note) + atomic_write_text(path, fm_block + "\n" + new_body) + else: + # Malformed/absent frontmatter: rebuild rather than write a bare + # body (mirrors compiler._write_concept's recovery path). + fm_block = _build_frontmatter( + [ + frontmatter.kv_line("type", "Concept"), + frontmatter.list_line("sources", [source_file]), + ] + ) + new_body = _upsert_note_line(existing, doc_name, note) + atomic_write_text(path, fm_block + new_body) + return + + fm_lines = [ + frontmatter.kv_line("type", "Concept"), + frontmatter.list_line("sources", [source_file]), + ] + if description: + fm_lines.append(frontmatter.kv_line("description", description)) + body = _upsert_note_line(f"{_NOTES_HEADING}\n\n", doc_name, note) + atomic_write_text(path, _build_frontmatter(fm_lines) + body) + + +def append_entity_note( + wiki_dir: Path, + name: str, + note: str, + source_file: str, + doc_name: str, + description: str = "", + type_: str = "other", +) -> None: + """Append a short note about ``doc_name`` to an entity page (no LLM write). + + Mirrors :func:`append_concept_note`; ``type_`` is only used to seed a new + page's frontmatter (no re-classification on update, unlike the rewrite + path's ``_gen_entity_update``) — a deliberate scope simplification for the + append mode. + """ + from openkb.agent import compiler as _compiler # local: avoid import cycle + + entities_dir = wiki_dir / "entities" + entities_dir.mkdir(parents=True, exist_ok=True) + safe_name = _compiler._sanitize_concept_name(name) + path = (entities_dir / f"{safe_name}.md").resolve() + if not path.is_relative_to(entities_dir.resolve()): + logger.warning("Entity name escapes entities dir: %s", name) + return + + if path.exists(): + existing = path.read_text(encoding="utf-8") + if source_file not in existing: + existing = _compiler._prepend_source_to_frontmatter(existing, source_file) + parts = frontmatter.split(existing) + if parts is not None: + fm_block, body = parts + new_body = _upsert_note_line(body.lstrip("\n"), doc_name, note) + atomic_write_text(path, fm_block + "\n" + new_body) + else: + fm_block = _build_frontmatter( + [ + frontmatter.list_line("sources", [source_file]), + frontmatter.kv_line("type", (type_ or "other").title()), + ] + ) + new_body = _upsert_note_line(existing, doc_name, note) + atomic_write_text(path, fm_block + new_body) + return + + fm_lines = [ + frontmatter.list_line("sources", [source_file]), + frontmatter.kv_line("type", (type_ or "other").title()), + ] + if description: + fm_lines.append(frontmatter.kv_line("description", description)) + body = _upsert_note_line(f"{_NOTES_HEADING}\n\n", doc_name, note) + atomic_write_text(path, _build_frontmatter(fm_lines) + body) diff --git a/openkb/agent/consolidator.py b/openkb/agent/consolidator.py new file mode 100644 index 000000000..79c936495 --- /dev/null +++ b/openkb/agent/consolidator.py @@ -0,0 +1,259 @@ +"""Consolidate a concept/entity page's accumulated "## Notes" into prose. + +Companion to ``openkb.agent.compiler_notes`` (``concept_update_mode="append"``): +once a page has accumulated one or more notes, ``openkb consolidate`` folds +them into the page's existing prose with a single LLM call — no new source +document, no concept/entity classification step, since the page is already +fixed. Contradictions between notes (or between notes and existing prose) are +described directly in the rewritten text rather than silently resolved. The +"## Notes" section is replaced entirely: after consolidation the page reads +like an ordinary ``concept_update_mode="rewrite"`` page — new notes appended +later (see ``compiler_notes``) start a fresh "## Notes" section, so the next +consolidation run only ever sees what changed since the last one. +""" + +from __future__ import annotations + +import asyncio +import logging +import time +from pathlib import Path + +from openkb import frontmatter +from openkb.agent.compiler_notes import _NOTES_HEADING +from openkb.lint import list_existing_wiki_targets, strip_ghost_wikilinks +from openkb.locks import atomic_write_text +from openkb.schema import get_agents_md + +logger = logging.getLogger(__name__) + +_CONSOLIDATE_CONCEPT_USER = """\ +Consolidate the concept page: {title} + +Existing prose on this page (may be empty if this is the first consolidation): +{existing_content} + +Accumulated notes to fold in, newest first (each tied to a source document): +{notes_content} + +Rewrite the ENTIRE page as a single, coherent Markdown page that: +- Preserves every distinct fact from both the existing prose and the notes. +- If notes conflict with each other or with the existing prose, describe the \ +conflict directly in the text (which source said what, and when) instead of \ +silently picking one side. +- Uses [[wikilinks]] to related concepts/entities, per the whitelist message \ +above. +- Does NOT include a "## Notes" section or any raw note lines — fold their \ +content into the prose instead. + +Return a JSON object with two keys: +- "description": A single sentence (under 100 chars) defining this concept +- "content": The rewritten full concept page in Markdown + +Return ONLY valid JSON, no fences. +""" + +_CONSOLIDATE_ENTITY_USER = """\ +Consolidate the entity page: {title} (type: {type}) + +Existing prose on this page (may be empty if this is the first consolidation): +{existing_content} + +Accumulated notes to fold in, newest first (each tied to a source document): +{notes_content} + +Rewrite the ENTIRE page as a single, coherent Markdown page that: +- Preserves every distinct fact from both the existing prose and the notes. +- If notes conflict with each other or with the existing prose, describe the \ +conflict directly in the text (which source said what, and when) instead of \ +silently picking one side. +- Uses [[wikilinks]] to related concepts/entities, per the whitelist message \ +above. +- Does NOT include a "## Notes" section or any raw note lines — fold their \ +content into the prose instead. + +Return a JSON object with two keys: +- "description": A single sentence (under 100 chars) identifying this entity +- "content": The rewritten full entity page in Markdown + +Return ONLY valid JSON, no fences. +""" + + +def _title_from_slug(slug: str) -> str: + return slug.replace("-", " ").title() + + +def _split_notes(body: str) -> tuple[str, str]: + """Split a page body into ``(existing_prose, notes_content)``. + + ``notes_content`` is empty when there is no ``## Notes`` heading (nothing + pending) — callers treat that as "skip, no notes to consolidate". + """ + lines = body.split("\n") + idx = next((i for i, ln in enumerate(lines) if ln.strip() == _NOTES_HEADING), None) + if idx is None: + return body.strip(), "" + existing = "\n".join(lines[:idx]).strip() + notes = "\n".join(lines[idx + 1 :]).strip() + return existing, notes + + +def count_pending_notes(text: str) -> int: + """Count ``## Notes`` bullet lines in a page's raw text (0 if none).""" + _, notes = _split_notes(text) + if not notes: + return 0 + return sum(1 for ln in notes.split("\n") if ln.lstrip().startswith("- **")) + + +def find_consolidation_candidates(wiki_dir: Path, min_notes: int = 1) -> list[tuple[str, str, int]]: + """Return ``(page_dir, slug, note_count)`` for pages with pending notes. + + ``page_dir`` is ``"concepts"`` or ``"entities"``. Only pages with at least + ``min_notes`` pending note lines are included. + """ + candidates: list[tuple[str, str, int]] = [] + for page_dir in ("concepts", "entities"): + dir_path = wiki_dir / page_dir + if not dir_path.is_dir(): + continue + for path in sorted(dir_path.glob("*.md")): + count = count_pending_notes(path.read_text(encoding="utf-8")) + if count >= min_notes: + candidates.append((page_dir, path.stem, count)) + return candidates + + +def resolve_page(wiki_dir: Path, name: str) -> list[tuple[str, str]]: + """Resolve ``name`` to ``[(page_dir, slug)]`` matches (exact slug first). + + Returns an empty list when nothing matches, or more than one entry when + ``name`` is an ambiguous substring across concepts/entities — callers + decide how to report each case. + """ + for page_dir in ("concepts", "entities"): + if (wiki_dir / page_dir / f"{name}.md").exists(): + return [(page_dir, name)] + + matches: list[tuple[str, str]] = [] + for page_dir in ("concepts", "entities"): + dir_path = wiki_dir / page_dir + if not dir_path.is_dir(): + continue + for path in sorted(dir_path.glob("*.md")): + if name.lower() in path.stem.lower(): + matches.append((page_dir, path.stem)) + return matches + + +async def consolidate_page( + wiki_dir: Path, page_dir: str, slug: str, model: str, language: str = "en" +) -> bool: + """Fold ``page_dir/slug``'s pending notes into curated prose. + + Returns ``False`` (no-op, no LLM call) when the page has no ``## Notes`` + section. Raises on LLM/parse failure — the CLI command treats a raised + exception for one page as a per-page failure, not a whole-batch abort + (mirrors ``recompile``). Async so the CLI can consolidate several pages + concurrently, the same way concept/entity generation does during ingest. + """ + from openkb.agent import compiler as _compiler + + path = wiki_dir / page_dir / f"{slug}.md" + text = path.read_text(encoding="utf-8") + parts = frontmatter.split(text) + if parts is None: + logger.warning("Skipping %s/%s: malformed or missing frontmatter.", page_dir, slug) + return False + fm_block, body = parts + existing_content, notes_content = _split_notes(body.lstrip("\n")) + if not notes_content: + return False + + fm = frontmatter.parse(text) + title = _title_from_slug(slug) + known_targets = list_existing_wiki_targets(wiki_dir) + known_targets_str = _compiler._format_known_targets(known_targets) + + system_msg = { + "role": "system", + "content": _compiler._SYSTEM_TEMPLATE.format( + schema_md=get_agents_md(wiki_dir), + language=language, + ), + } + known_targets_msg = { + "role": "user", + "content": _compiler._KNOWN_TARGETS_USER.format(known_targets=known_targets_str), + } + if page_dir == "entities": + etype = fm.get("type", "other") + user_content = _CONSOLIDATE_ENTITY_USER.format( + title=title, + type=etype, + existing_content=existing_content or "(none — first consolidation for this page)", + notes_content=notes_content, + ) + else: + user_content = _CONSOLIDATE_CONCEPT_USER.format( + title=title, + existing_content=existing_content or "(none — first consolidation for this page)", + notes_content=notes_content, + ) + + raw = await _compiler._llm_call_async( + model, + [system_msg, known_targets_msg, {"role": "user", "content": user_content}], + f"consolidate: {page_dir}/{slug}", + response_format=_compiler._JSON_RESPONSE_FORMAT, + ) + description, content, _obj = _compiler._page_fields(raw) + _compiler._require_nonempty_content(content, slug) + + clean_parts = frontmatter.split(content) + clean = clean_parts[1].lstrip("\n") if clean_parts is not None else content + cleaned, ghosts = strip_ghost_wikilinks(clean, known_targets) + if ghosts: + logger.info( + "stripped %d ghost wikilink(s) from consolidated %s/%s: %s", + len(ghosts), + page_dir, + slug, + ghosts[:5], + ) + + if description: + fm_block = frontmatter.set_line(fm_block, "description", description) + atomic_write_text(path, fm_block + "\n" + cleaned) + return True + + +async def consolidate_pages( + wiki_dir: Path, + targets: list[tuple[str, str, int]], + model: str, + language: str, + max_concurrency: int, +) -> list[tuple[bool | None, Exception | None, float]]: + """Consolidate several pages concurrently, bounded by ``max_concurrency``. + + Mirrors the concept/entity generation concurrency model used during + ingest (``openkb.agent.compiler._compile_concepts``). Returns one + ``(ok, error, elapsed_seconds)`` per target, in the same order as + ``targets`` regardless of completion order — a per-page exception is + captured here rather than propagated, so one page's failure never + aborts the batch (mirrors ``recompile``). + """ + semaphore = asyncio.Semaphore(max_concurrency) + + async def _run_one(page_dir: str, slug: str) -> tuple[bool | None, Exception | None, float]: + start = time.time() + async with semaphore: + try: + ok = await consolidate_page(wiki_dir, page_dir, slug, model, language=language) + except Exception as exc: + return None, exc, time.time() - start + return ok, None, time.time() - start + + return await asyncio.gather(*(_run_one(page_dir, slug) for page_dir, slug, _ in targets)) diff --git a/openkb/cli.py b/openkb/cli.py index c9e543181..17c3ad8ac 100644 --- a/openkb/cli.py +++ b/openkb/cli.py @@ -2055,6 +2055,141 @@ def _classify(meta: dict) -> str: append_log(wiki_dir, "recompile", f"recompiled {recompiled}, skipped {skipped}") +@cli.command() +@click.argument("page_name", required=False) +@click.option( + "--all", "all_pages", is_flag=True, default=False, help="Consolidate every pending page." +) +@click.option( + "--min-notes", + type=int, + default=1, + show_default=True, + help="With --all, only consider pages with at least this many pending notes.", +) +@click.option( + "--dry-run", + is_flag=True, + default=False, + help="List the pages that would be consolidated; no LLM calls, no writes.", +) +@click.option( + "--yes", "-y", is_flag=True, default=False, help="Skip the --all confirmation prompt." +) +@click.pass_context +@_with_kb_lock(exclusive=True) +def consolidate(ctx, page_name, all_pages, min_notes, dry_run, yes): + """Fold a concept/entity page's accumulated "## Notes" into curated prose. + + Only relevant under ``concept_update_mode: append`` (see ``openkb add``): + each ingested document appends a short dated note to the concept/entity + pages it touches instead of rewriting them in full. This command folds + those notes into the page's prose with a single LLM call per page — no + new source document, no concept/entity classification, since the page is + already fixed. Contradictions between notes (or between notes and + existing prose) are described directly in the rewritten text rather than + silently resolved. + + PAGE_NAME resolves like ``openkb remove`` — exact slug first, else a + unique substring match across ``wiki/concepts/`` and ``wiki/entities/``. + ``--all`` consolidates every page with at least ``--min-notes`` pending + notes. Exactly one of PAGE_NAME or ``--all`` is required. With ``--all``, + pages are consolidated concurrently (bounded by the ``concurrency:`` + config key, same as concept/entity generation during ingest). + + Side effect: this replaces the page's "## Notes" section with prose — + manual edits inside that section are overwritten. Existing prose above + it, ``sources:``, and ``type:`` are preserved; only ``description:`` may + be refreshed. + """ + from openkb.agent import consolidator + + if bool(page_name) == bool(all_pages): + click.echo("Specify exactly one of PAGE_NAME or --all.") + return + + kb_dir = _find_kb_dir(ctx.obj.get("kb_dir_override")) + if kb_dir is None: + click.echo("No knowledge base found. Run `openkb init` first.") + return + wiki_dir = kb_dir / "wiki" + + if page_name: + matches = consolidator.resolve_page(wiki_dir, page_name) + if not matches: + click.echo(f"No concept/entity page matching '{page_name}' found.") + return + if len(matches) > 1: + click.echo(f"'{page_name}' matches multiple pages:") + for page_dir, slug in matches: + click.echo(f" - {page_dir}/{slug}") + return + page_dir, slug = matches[0] + targets = [ + ( + page_dir, + slug, + consolidator.count_pending_notes( + (wiki_dir / page_dir / f"{slug}.md").read_text(encoding="utf-8") + ), + ) + ] + else: + targets = consolidator.find_consolidation_candidates(wiki_dir, min_notes=min_notes) + if not targets: + click.echo("No pages with pending notes found.") + return + + if dry_run: + click.echo(f"Would consolidate {len(targets)} page(s):") + for page_dir, slug, count in targets: + click.echo(f" - {page_dir}/{slug} ({count} note(s))") + click.echo("(dry-run — nothing modified)") + return + + if all_pages and not yes and len(targets) > 1: + click.echo( + f"This will consolidate {len(targets)} page(s), replacing each " + 'page\'s "## Notes" section with rewritten prose.' + ) + if not click.confirm("Proceed?", default=False): + click.echo("Aborted.") + return + + _setup_llm_key(kb_dir) + config = resolve_effective_config(kb_dir)[0] + model: str = config.get("model", DEFAULT_CONFIG["model"]) + language: str = config.get("language", "en") + max_concurrency = resolve_concurrency(config) or DEFAULT_COMPILE_CONCURRENCY + + total = len(targets) + click.echo(f"Consolidating {total} page(s) (concurrency={max_concurrency})...") + for page_dir, slug, count in targets: + click.echo(f" - {page_dir}/{slug} ({count} note(s))") + results = asyncio.run( + consolidator.consolidate_pages(wiki_dir, targets, model, language, max_concurrency) + ) + + consolidated = 0 + skipped = 0 + for (page_dir, slug, _count), (ok, exc, elapsed) in zip(targets, results): + if exc is not None: + click.echo(f" [ERROR] Consolidation failed: {exc}") + logging.getLogger(__name__).debug( + "Consolidate traceback for %s/%s:", page_dir, slug, exc_info=exc + ) + skipped += 1 + elif ok: + click.echo(f" [OK] {page_dir}/{slug} ({elapsed:.1f}s)") + consolidated += 1 + else: + click.echo(f" [SKIP] {page_dir}/{slug} (no pending notes).") + skipped += 1 + + click.echo(f"\nDone: consolidated {consolidated}, skipped {skipped}.") + append_log(wiki_dir, "consolidate", f"consolidated {consolidated}, skipped {skipped}") + + async def iter_recompile( kb_dir: Path, doc_name: str | None = None, diff --git a/openkb/config.py b/openkb/config.py index 95ca8691f..46173c840 100644 --- a/openkb/config.py +++ b/openkb/config.py @@ -36,8 +36,14 @@ # global/KB list overrides it wholesale; resolve_entity_types cleans the # effective value on read. "entity_types": list(DEFAULT_ENTITY_TYPES), + # How concept/entity pages absorb new documents on `openkb add` — see + # resolve_concept_update_mode(). KB config.yaml only (like `debug`), not + # in GLOBAL_SCALAR_KEYS. + "concept_update_mode": "rewrite", } +VALID_CONCEPT_UPDATE_MODES: tuple[str, ...] = ("rewrite", "append") + GLOBAL_CONFIG_DIR = Path.home() / ".config" / "openkb" GLOBAL_CONFIG_PATH = GLOBAL_CONFIG_DIR / "global.yaml" GLOBAL_CONFIG_LOCK_PATH = GLOBAL_CONFIG_DIR / "global.lock" @@ -140,6 +146,23 @@ def resolve_entity_types(config: dict, *, warn: bool = True) -> list[str]: return cleaned +def resolve_concept_update_mode(config: dict) -> str: + """Resolve ``concept_update_mode:`` — ``"rewrite"`` (default, full-page + LLM rewrite on update) or ``"append"`` (short note instead, see + ``openkb.agent.compiler_notes``). Invalid values degrade to ``"rewrite"`` + with a warning. + """ + value = config.get("concept_update_mode", "rewrite") + if value not in VALID_CONCEPT_UPDATE_MODES: + logger.warning( + "config: 'concept_update_mode' must be one of %s, got %r — using 'rewrite'.", + VALID_CONCEPT_UPDATE_MODES, + value, + ) + return "rewrite" + return value + + def resolve_extra_headers(config: dict) -> dict[str, str]: """Resolve the optional ``extra_headers:`` config key into a str→str dict. diff --git a/tests/test_compiler.py b/tests/test_compiler.py index 95a57cc4c..dc9579530 100644 --- a/tests/test_compiler.py +++ b/tests/test_compiler.py @@ -1220,6 +1220,64 @@ async def test_full_pipeline(self, tmp_path): assert "[[summaries/test-doc]]" in index_text assert "[[concepts/transformer]]" in index_text + @pytest.mark.asyncio + async def test_append_mode_full_pipeline_writes_note_not_full_page(self, tmp_path): + """concept_update_mode="append" end-to-end, through the SAME mocked + acompletion path as the "rewrite" pipeline above — proves the note + closures work under this branch's _llm_call_page_async. + """ + wiki = tmp_path / "wiki" + (wiki / "sources").mkdir(parents=True) + (wiki / "summaries").mkdir(parents=True) + (wiki / "concepts").mkdir(parents=True) + (wiki / "index.md").write_text( + "# Index\n\n## Documents\n\n## Concepts\n\n## Explorations\n", + encoding="utf-8", + ) + source_path = wiki / "sources" / "test-doc.md" + source_path.write_text("# Test Doc\n\nA support ticket about approvals.", encoding="utf-8") + (tmp_path / ".openkb").mkdir() + (tmp_path / ".openkb" / "config.yaml").write_text( + "concept_update_mode: append\n", encoding="utf-8" + ) + + summary_response = json.dumps( + {"description": "A ticket about approvals", "content": "# Summary\n\nApproval ticket."} + ) + concepts_plan_response = json.dumps( + { + "create": [{"name": "approval-workflows", "title": "Approval Workflows"}], + "update": [], + "related": [], + } + ) + summary_rewrite_response = ( + "# Summary\n\nApproval ticket about [[concepts/approval-workflows]]." + ) + note_response = json.dumps( + { + "description": "How approvals are routed.", + "note": "This ticket reports a timeout during approval.", + } + ) + + with patch("openkb.agent.compiler.litellm") as mock_litellm: + mock_litellm.completion = MagicMock( + side_effect=_mock_completion( + [summary_response, concepts_plan_response, summary_rewrite_response] + ) + ) + mock_litellm.acompletion = AsyncMock(side_effect=_mock_acompletion([note_response])) + await compile_short_doc("test-doc", source_path, tmp_path, "gpt-4o-mini") + + concept_path = wiki / "concepts" / "approval-workflows.md" + assert concept_path.exists() + text = concept_path.read_text(encoding="utf-8") + assert "## Notes" in text + assert "This ticket reports a timeout during approval." in text + assert 'description: "How approvals are routed."' in text + assert 'sources: ["summaries/test-doc.md"]' in text + @pytest.mark.asyncio async def test_handles_bad_json(self, tmp_path): wiki = tmp_path / "wiki" @@ -2252,6 +2310,28 @@ def test_strips_standalone_see_also_line(self, tmp_path): assert "See also" not in shared assert "summaries/other" in shared + def test_strips_append_mode_note_line_keeps_other_notes(self, tmp_path): + # A "concept_update_mode=append" page (compiler_notes.append_entity_note) + # keeps its notes under "## Notes", one dated line per source doc. + # Removing one doc must strip only ITS line, not the whole section. + ent = tmp_path / "entities" + ent.mkdir() + (ent / "shared.md").write_text( + "---\ntype: organization\nsources: [summaries/doc.md, summaries/other.md]\n---\n\n" + "## Notes\n\n" + "- **2026-09-07** Mentioned in doc. ([[summaries/doc]])\n" + "- **2026-09-01** Mentioned in other. ([[summaries/other]])\n", + encoding="utf-8", + ) + result = remove_doc_from_entity_pages(tmp_path, "doc") + assert result == {"modified": ["shared"], "deleted": []} + shared = (ent / "shared.md").read_text(encoding="utf-8") + assert "summaries/doc" not in shared + assert "Mentioned in doc." not in shared + assert "## Notes" in shared + assert "Mentioned in other." in shared + assert "summaries/other" in shared + class TestCompileEntitiesEndToEnd: @pytest.mark.asyncio diff --git a/tests/test_compiler_notes.py b/tests/test_compiler_notes.py new file mode 100644 index 000000000..23d25803f --- /dev/null +++ b/tests/test_compiler_notes.py @@ -0,0 +1,172 @@ +"""Tests for openkb.agent.compiler_notes (concept_update_mode="append").""" + +from __future__ import annotations + +from openkb.agent.compiler_notes import ( + _NOTES_HEADING, + _upsert_note_line, + append_concept_note, + append_entity_note, + note_fields, +) + + +class TestNoteFields: + def test_parses_description_and_note(self): + raw = '{"description": "A greeting", "note": "Says hello."}' + assert note_fields(raw) == ("A greeting", "Says hello.") + + def test_update_shape_has_no_description(self): + raw = '{"note": "Adds a detail."}' + assert note_fields(raw) == ("", "Adds a detail.") + + def test_non_json_falls_back_to_raw_text_as_note(self): + assert note_fields(" Just a plain note. ") == ("", "Just a plain note.") + + def test_malformed_shape_returns_empty(self): + # A JSON array of scalars is valid JSON but not a usable object. + assert note_fields("[1, 2, 3]") == ("", "") + + +class TestUpsertNoteLine: + def test_seeds_heading_and_inserts_first_note(self): + body = _upsert_note_line(f"{_NOTES_HEADING}\n\n", "jira-1", "First note.") + expected = f"{_NOTES_HEADING}\n\n- **{_today()}** First note. ([[summaries/jira-1]])" + assert body.rstrip("\n") == expected + + def test_second_doc_inserted_above_first(self): + body = _upsert_note_line(f"{_NOTES_HEADING}\n\n", "jira-1", "First note.") + body = _upsert_note_line(body, "jira-2", "Second note.") + lines = body.split("\n") + note_lines = [ln for ln in lines if ln.startswith("- **")] + assert len(note_lines) == 2 + assert "jira-2" in note_lines[0] # newest first + assert "jira-1" in note_lines[1] + + def test_reingesting_same_doc_replaces_not_duplicates(self): + body = _upsert_note_line(f"{_NOTES_HEADING}\n\n", "jira-1", "Old text.") + body = _upsert_note_line(body, "jira-2", "Unrelated.") + body = _upsert_note_line(body, "jira-1", "Updated text.") + note_lines = [ln for ln in body.split("\n") if ln.startswith("- **")] + assert len(note_lines) == 2 + assert any("Updated text." in ln for ln in note_lines) + assert not any("Old text." in ln for ln in note_lines) + + def test_creates_missing_heading(self): + body = _upsert_note_line("Some unrelated body.", "jira-1", "A note.") + assert body.split("\n") == [ + "Some unrelated body.", + "", + _NOTES_HEADING, + "", + f"- **{_today()}** A note. ([[summaries/jira-1]])", + ] + + +class TestAppendConceptNote: + def test_creates_new_page_with_description(self, tmp_path): + wiki = tmp_path / "wiki" + wiki.mkdir() + append_concept_note( + wiki, + "approval-workflows", + "Customer reports a timeout.", + "summaries/jira-1.md", + "jira-1", + description="How approvals are routed.", + ) + path = wiki / "concepts" / "approval-workflows.md" + text = path.read_text(encoding="utf-8") + assert 'type: "Concept"' in text + assert 'sources: ["summaries/jira-1.md"]' in text + assert 'description: "How approvals are routed."' in text + assert _NOTES_HEADING in text + assert "Customer reports a timeout." in text + assert "[[summaries/jira-1]]" in text + + def test_creates_new_page_without_description(self, tmp_path): + wiki = tmp_path / "wiki" + wiki.mkdir() + append_concept_note(wiki, "approval-workflows", "A note.", "summaries/jira-1.md", "jira-1") + text = (wiki / "concepts" / "approval-workflows.md").read_text(encoding="utf-8") + assert "description:" not in text + + def test_update_appends_source_and_note_keeps_description(self, tmp_path): + wiki = tmp_path / "wiki" + wiki.mkdir() + append_concept_note( + wiki, + "approval-workflows", + "First ticket note.", + "summaries/jira-1.md", + "jira-1", + description="Original description.", + ) + append_concept_note( + wiki, + "approval-workflows", + "Second ticket note.", + "summaries/jira-2.md", + "jira-2", + description="Ignored — page already exists.", + ) + text = (wiki / "concepts" / "approval-workflows.md").read_text(encoding="utf-8") + assert '"summaries/jira-1.md"' in text + assert '"summaries/jira-2.md"' in text + assert "First ticket note." in text + assert "Second ticket note." in text + # description is frozen at first creation, never overwritten. + assert 'description: "Original description."' in text + assert "Ignored" not in text + + def test_reingest_same_doc_replaces_note_not_source_duplicate(self, tmp_path): + wiki = tmp_path / "wiki" + wiki.mkdir() + append_concept_note( + wiki, "approval-workflows", "Old note.", "summaries/jira-1.md", "jira-1" + ) + append_concept_note( + wiki, "approval-workflows", "Updated note.", "summaries/jira-1.md", "jira-1" + ) + text = (wiki / "concepts" / "approval-workflows.md").read_text(encoding="utf-8") + assert text.count("summaries/jira-1.md") == 1 # sources: list stays deduped + assert "Updated note." in text + assert "Old note." not in text + + +class TestAppendEntityNote: + def test_creates_new_page_with_capitalized_type(self, tmp_path): + wiki = tmp_path / "wiki" + wiki.mkdir() + append_entity_note( + wiki, + "acme-corp", + "Mentioned as the customer.", + "summaries/jira-1.md", + "jira-1", + description="A customer organization.", + type_="organization", + ) + text = (wiki / "entities" / "acme-corp.md").read_text(encoding="utf-8") + assert 'type: "Organization"' in text + assert 'description: "A customer organization."' in text + assert "Mentioned as the customer." in text + + def test_update_keeps_original_type_regardless_of_new_value(self, tmp_path): + wiki = tmp_path / "wiki" + wiki.mkdir() + append_entity_note( + wiki, "acme-corp", "First note.", "summaries/jira-1.md", "jira-1", type_="organization" + ) + append_entity_note( + wiki, "acme-corp", "Second note.", "summaries/jira-2.md", "jira-2", type_="person" + ) + text = (wiki / "entities" / "acme-corp.md").read_text(encoding="utf-8") + assert 'type: "Organization"' in text + assert "Person" not in text + + +def _today() -> str: + import datetime + + return datetime.date.today().isoformat() diff --git a/tests/test_config.py b/tests/test_config.py index 65572d6b9..8254d8f68 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -12,6 +12,7 @@ kb_root_dir, load_config, registered_kbs, + resolve_concept_update_mode, resolve_concurrency, resolve_effective_config, resolve_extra_headers, @@ -145,6 +146,35 @@ def test_default_config_values(): assert DEFAULT_CONFIG["pageindex_threshold"] == 20 +# --- concept_update_mode ------------------------------------------------------- + + +def test_concept_update_mode_default_in_config(): + assert DEFAULT_CONFIG["concept_update_mode"] == "rewrite" + + +def test_concept_update_mode_not_in_global_scalar_keys(): + # KB config.yaml only (like `debug`/`insert_mode`) — not workbench/global- + # editable, so it must never leak into the global.yaml layering. + assert "concept_update_mode" not in GLOBAL_SCALAR_KEYS + + +def test_resolve_concept_update_mode_absent_is_default(): + assert resolve_concept_update_mode({}) == "rewrite" + + +def test_resolve_concept_update_mode_valid_values(): + assert resolve_concept_update_mode({"concept_update_mode": "rewrite"}) == "rewrite" + assert resolve_concept_update_mode({"concept_update_mode": "append"}) == "append" + + +def test_resolve_concept_update_mode_rejects_invalid(caplog): + with caplog.at_level(logging.WARNING, logger="openkb.config"): + result = resolve_concept_update_mode({"concept_update_mode": "bogus"}) + assert result == "rewrite" + assert "concept_update_mode" in caplog.text + + def test_concurrency_not_in_default_config(): # Like the other optional tuning knobs (timeout, extra_headers, # parallel_tool_calls), concurrency stays out of DEFAULT_CONFIG — diff --git a/tests/test_consolidate_cli.py b/tests/test_consolidate_cli.py new file mode 100644 index 000000000..3168b102d --- /dev/null +++ b/tests/test_consolidate_cli.py @@ -0,0 +1,124 @@ +"""Tests for the `openkb consolidate` CLI command.""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, patch + +from click.testing import CliRunner + +from openkb.cli import cli + + +def _invoke(kb_dir, args): + return CliRunner().invoke(cli, ["--kb-dir", str(kb_dir), *args]) + + +def _seed_page_with_notes(kb_dir, page_dir="concepts", slug="approval-workflows"): + d = kb_dir / "wiki" / page_dir + d.mkdir(parents=True, exist_ok=True) + (d / f"{slug}.md").write_text( + '---\nsources: ["summaries/a.md"]\n---\n\nExisting prose.\n\n## Notes\n\n' + "- **2026-01-01** A note. ([[summaries/a]])\n", + encoding="utf-8", + ) + (kb_dir / "wiki" / "log.md").write_text("# Log\n\n", encoding="utf-8") + + +class TestConsolidateArgValidation: + def test_requires_exactly_one_of_name_or_all(self, kb_dir): + _seed_page_with_notes(kb_dir) + result = _invoke(kb_dir, ["consolidate"]) + assert result.exit_code == 0 + assert "exactly one" in result.output.lower() + + def test_unknown_page_name(self, kb_dir): + _seed_page_with_notes(kb_dir) + result = _invoke(kb_dir, ["consolidate", "nonexistent"]) + assert "No concept/entity page matching" in result.output + + +class TestConsolidateDryRun: + def test_dry_run_lists_candidates_no_calls_no_writes(self, kb_dir): + _seed_page_with_notes(kb_dir) + path = kb_dir / "wiki" / "concepts" / "approval-workflows.md" + before = path.read_text(encoding="utf-8") + with patch( + "openkb.agent.consolidator.consolidate_page", new_callable=AsyncMock + ) as mock_consolidate: + result = _invoke(kb_dir, ["consolidate", "--all", "--dry-run"]) + + assert result.exit_code == 0, result.output + mock_consolidate.assert_not_called() + assert "approval-workflows" in result.output + assert "1 note" in result.output + assert path.read_text(encoding="utf-8") == before + + def test_min_notes_filters_dry_run_candidates(self, kb_dir): + _seed_page_with_notes(kb_dir) + with patch( + "openkb.agent.consolidator.consolidate_page", new_callable=AsyncMock + ) as mock_consolidate: + result = _invoke(kb_dir, ["consolidate", "--all", "--min-notes", "5", "--dry-run"]) + + assert result.exit_code == 0, result.output + mock_consolidate.assert_not_called() + assert "No pages with pending notes found." in result.output + + +class TestConsolidateExecution: + def test_single_page_by_name_dispatches_consolidate_page(self, kb_dir): + _seed_page_with_notes(kb_dir) + with patch( + "openkb.agent.consolidator.consolidate_page", + new_callable=AsyncMock, + return_value=True, + ) as mock_c: + result = _invoke(kb_dir, ["consolidate", "approval-workflows"]) + + assert result.exit_code == 0, result.output + mock_c.assert_called_once() + args = mock_c.call_args.args + assert args[1] == "concepts" + assert args[2] == "approval-workflows" + assert "Done: consolidated 1, skipped 0." in result.output + log_text = (kb_dir / "wiki" / "log.md").read_text(encoding="utf-8") + assert "consolidate" in log_text + + def test_all_with_yes_skips_confirmation(self, kb_dir): + _seed_page_with_notes(kb_dir) + with patch( + "openkb.agent.consolidator.consolidate_page", + new_callable=AsyncMock, + return_value=True, + ) as mock_c: + result = _invoke(kb_dir, ["consolidate", "--all", "--yes"]) + + assert result.exit_code == 0, result.output + mock_c.assert_called_once() + assert "Done: consolidated 1, skipped 0." in result.output + + def test_skip_result_counts_as_skipped(self, kb_dir): + _seed_page_with_notes(kb_dir) + with patch( + "openkb.agent.consolidator.consolidate_page", + new_callable=AsyncMock, + return_value=False, + ): + result = _invoke(kb_dir, ["consolidate", "approval-workflows"]) + + assert result.exit_code == 0, result.output + assert "Done: consolidated 0, skipped 1." in result.output + + def test_exception_in_one_page_reported_as_error_not_fatal(self, kb_dir): + _seed_page_with_notes(kb_dir, slug="approval-workflows") + _seed_page_with_notes(kb_dir, slug="second-page") + with patch( + "openkb.agent.consolidator.consolidate_page", + new_callable=AsyncMock, + side_effect=[ValueError("boom"), True], + ): + result = _invoke(kb_dir, ["consolidate", "--all", "--yes"]) + + assert result.exit_code == 0, result.output + assert "[ERROR] Consolidation failed: boom" in result.output + assert "Done: consolidated 1, skipped 1." in result.output diff --git a/tests/test_consolidator.py b/tests/test_consolidator.py new file mode 100644 index 000000000..a9914039b --- /dev/null +++ b/tests/test_consolidator.py @@ -0,0 +1,171 @@ +"""Tests for openkb.agent.consolidator.""" + +from __future__ import annotations + +import json +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from openkb.agent.consolidator import ( + _split_notes, + consolidate_page, + count_pending_notes, + find_consolidation_candidates, + resolve_page, +) + + +def _mock_acompletion(response: str): + async def side_effect(*args, **kwargs): + mock_resp = MagicMock() + mock_resp.choices = [MagicMock()] + mock_resp.choices[0].message.content = response + mock_resp.usage = MagicMock(prompt_tokens=100, completion_tokens=50) + mock_resp.usage.prompt_tokens_details = None + return mock_resp + + return side_effect + + +def _write_page(wiki, page_dir: str, slug: str, body: str): + d = wiki / page_dir + d.mkdir(parents=True, exist_ok=True) + path = d / f"{slug}.md" + path.write_text(body, encoding="utf-8") + return path + + +class TestSplitNotesAndCount: + def test_no_notes_heading(self): + existing, notes = _split_notes("# Attention\n\nSome prose.") + assert existing == "# Attention\n\nSome prose." + assert notes == "" + + def test_splits_prose_and_notes(self): + body = ( + "# Attention\n\nSome prose.\n\n## Notes\n\n- **2026-01-01** A note. ([[summaries/x]])" + ) + existing, notes = _split_notes(body) + assert existing == "# Attention\n\nSome prose." + assert "A note." in notes + + def test_count_pending_notes_zero_without_heading(self): + assert count_pending_notes("# Attention\n\nSome prose.") == 0 + + def test_count_pending_notes_counts_bullets(self): + text = ( + "---\nsources: [a]\n---\n\n## Notes\n\n" + "- **2026-01-02** Second. ([[summaries/b]])\n" + "- **2026-01-01** First. ([[summaries/a]])\n" + ) + assert count_pending_notes(text) == 2 + + +class TestFindConsolidationCandidates: + def test_finds_pages_with_enough_notes(self, tmp_path): + wiki = tmp_path / "wiki" + _write_page( + wiki, + "concepts", + "approval-workflows", + '---\nsources: ["a"]\n---\n\n## Notes\n\n- **2026-01-01** Note. ([[summaries/a]])\n', + ) + _write_page(wiki, "concepts", "no-notes", '---\nsources: ["a"]\n---\n\n# Prose only.\n') + candidates = find_consolidation_candidates(wiki, min_notes=1) + assert candidates == [("concepts", "approval-workflows", 1)] + + def test_min_notes_filters_out_thin_pages(self, tmp_path): + wiki = tmp_path / "wiki" + _write_page( + wiki, + "entities", + "acme-corp", + '---\nsources: ["a"]\n---\n\n## Notes\n\n- **2026-01-01** Note. ([[summaries/a]])\n', + ) + assert find_consolidation_candidates(wiki, min_notes=2) == [] + + +class TestResolvePage: + def test_exact_slug_match(self, tmp_path): + wiki = tmp_path / "wiki" + _write_page(wiki, "concepts", "approval-workflows", "---\nsources: []\n---\n\nBody.") + assert resolve_page(wiki, "approval-workflows") == [("concepts", "approval-workflows")] + + def test_no_match(self, tmp_path): + wiki = tmp_path / "wiki" + wiki.mkdir() + assert resolve_page(wiki, "nonexistent") == [] + + def test_ambiguous_substring_match(self, tmp_path): + wiki = tmp_path / "wiki" + _write_page(wiki, "concepts", "approval-workflows", "---\nsources: []\n---\n\nBody.") + _write_page(wiki, "entities", "approval-bot", "---\nsources: []\n---\n\nBody.") + matches = resolve_page(wiki, "approval") + assert set(matches) == {("concepts", "approval-workflows"), ("entities", "approval-bot")} + + +class TestConsolidatePage: + @pytest.mark.asyncio + async def test_returns_false_without_notes_section(self, tmp_path): + wiki = tmp_path / "wiki" + _write_page(wiki, "concepts", "approval-workflows", '---\nsources: ["a"]\n---\n\nProse.\n') + with patch("openkb.agent.compiler.litellm") as mock_litellm: + mock_litellm.acompletion = AsyncMock(side_effect=AssertionError("should not be called")) + result = await consolidate_page(wiki, "concepts", "approval-workflows", "gpt-4o-mini") + assert result is False + + @pytest.mark.asyncio + async def test_consolidates_and_replaces_notes_section(self, tmp_path): + wiki = tmp_path / "wiki" + _write_page( + wiki, + "concepts", + "approval-workflows", + '---\nsources: ["summaries/a.md", "summaries/b.md"]\ntype: "Concept"\n' + 'description: "Old description"\n---\n\n' + "Existing prose.\n\n## Notes\n\n" + "- **2026-01-02** Second ticket note. ([[summaries/b]])\n" + "- **2026-01-01** First ticket note. ([[summaries/a]])\n", + ) + response = json.dumps( + { + "description": "How approvals are routed and escalated.", + "content": "# Approval Workflows\n\nConsolidated prose covering both tickets.", + } + ) + with patch("openkb.agent.compiler.litellm") as mock_litellm: + mock_litellm.acompletion = AsyncMock(side_effect=_mock_acompletion(response)) + result = await consolidate_page(wiki, "concepts", "approval-workflows", "gpt-4o-mini") + + assert result is True + text = (wiki / "concepts" / "approval-workflows.md").read_text(encoding="utf-8") + assert "## Notes" not in text + assert "Consolidated prose covering both tickets." in text + assert 'description: "How approvals are routed and escalated."' in text + # sources: untouched by consolidation. + assert '"summaries/a.md"' in text + assert '"summaries/b.md"' in text + + @pytest.mark.asyncio + async def test_strips_ghost_wikilinks_from_consolidated_content(self, tmp_path): + wiki = tmp_path / "wiki" + _write_page( + wiki, + "concepts", + "approval-workflows", + '---\nsources: ["summaries/a.md"]\n---\n\n## Notes\n\n' + "- **2026-01-01** Note. ([[summaries/a]])\n", + ) + response = json.dumps( + { + "description": "Desc.", + "content": "Mentions [[concepts/nonexistent-page]] which doesn't exist.", + } + ) + with patch("openkb.agent.compiler.litellm") as mock_litellm: + mock_litellm.acompletion = AsyncMock(side_effect=_mock_acompletion(response)) + await consolidate_page(wiki, "concepts", "approval-workflows", "gpt-4o-mini") + + text = (wiki / "concepts" / "approval-workflows.md").read_text(encoding="utf-8") + assert "[[concepts/nonexistent-page]]" not in text