From d6dcabe4f72e2ae1e18d56d13bf09f4b29dce51c Mon Sep 17 00:00:00 2001 From: Crespin Alexandre <114501859+kunail0804@users.noreply.github.com> Date: Tue, 7 Jul 2026 19:31:16 +0200 Subject: [PATCH] feat(knowledge-base): add Knowledge Base validation tooling (#9) (#52) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implement whole-Knowledge-Base validation per MASTER_SPEC §15.10, §26.4: compiler/common/kb_validation.py (importable by the loader) plus a CLI at scripts/validate_knowledge_base.py. validate_knowledge_base(entries) collects every problem (never stops at the first); each Message carries a code, the offending id, and a description: - duplicate canonical id -> SC0005 - invalid category -> SC0008 (drawn from the shared 19-category constant) - circular expansion (incl. self-expansion) -> SC0003, via DFS cycle detection - duplicate alias, alias colliding with a canonical id (no chains), missing expansion target, and per-entry schema nonconformance (empty tags, unknown fields, ...) -> umbrella SC0004 Findings are deterministically ordered. The CLI loads every *.json (array of entries) from a directory, reports invalid JSON / non-array files as SC0004, writes JSON findings to stdout, and exits non-zero when invalid. Tests: a passing fixture plus a fixture per failure mode, multi-problem reporting, determinism, and the CLI on valid/invalid/broken-JSON inputs. Co-authored-by: kunail0804 Co-authored-by: Claude Opus 4.8 --- compiler/common/kb_validation.py | 221 +++++++++++++++++++++++++++++ scripts/validate_knowledge_base.py | 79 +++++++++++ tests/test_kb_validation.py | 128 +++++++++++++++++ tests/test_kb_validation_cli.py | 49 +++++++ 4 files changed, 477 insertions(+) create mode 100644 compiler/common/kb_validation.py create mode 100644 scripts/validate_knowledge_base.py create mode 100644 tests/test_kb_validation.py create mode 100644 tests/test_kb_validation_cli.py diff --git a/compiler/common/kb_validation.py b/compiler/common/kb_validation.py new file mode 100644 index 0000000..836b081 --- /dev/null +++ b/compiler/common/kb_validation.py @@ -0,0 +1,221 @@ +"""Knowledge Base validation tooling (MASTER_SPEC §15.10, §26.4). + +Validates a whole Knowledge Base — both per-entry structure and cross-entry +integrity — and returns every problem found (it does not stop at the first). +Each problem is a :class:`Message` carrying a code, the offending ``id`` in its +context, and a description. + +This core is importable by the Knowledge Base loader (#7) and driven by the CLI +in ``scripts/validate_knowledge_base.py``. + +Codes (Appendix B): duplicate canonical id → ``SC0005``; invalid category → +``SC0008``; circular expansion → ``SC0003``. The remaining Knowledge-Base +structural problems have no dedicated Appendix B code and are reported under the +umbrella ``SC0004`` (Knowledge Base Load Failure): duplicate alias, an alias +colliding with a canonical id (which would create a resolution chain), a missing +expansion target, and any per-entry schema nonconformance (including empty tags +and unknown fields). +""" + +from __future__ import annotations + +from collections.abc import Iterable, Mapping, Sequence +from typing import Any + +from compiler.common.categories import is_valid_category +from compiler.common.result import Message, Severity +from schemas.validation import validate_document + +_ENTRY_SCHEMA = "knowledge_base_entry" + + +def _sc0004(title: str, description: str, entry_id: str, **context: Any) -> Message: + return Message( + code="SC0004", + severity=Severity.FATAL, + title=title, + description=description, + context={"id": entry_id, **context}, + ) + + +def _schema_messages(entry: Mapping[str, Any], entry_id: str) -> list[Message]: + """Per-entry schema conformance, minus the category (checked explicitly).""" + messages: list[Message] = [] + for issue in validate_document(entry, _ENTRY_SCHEMA): + if issue.path == "category": + continue # reported as SC0008 by the explicit category check + messages.append( + _sc0004( + title="Invalid Knowledge Base entry", + description=f"Entry '{entry_id}' failed schema validation: {issue.message}", + entry_id=entry_id, + path=issue.path, + ) + ) + return messages + + +def _category_messages(entry: Mapping[str, Any], entry_id: str) -> list[Message]: + category = entry.get("category") + if category is not None and not is_valid_category(category): + return [ + Message( + code="SC0008", + severity=Severity.ERROR, + title="Invalid Category", + description=f"Entry '{entry_id}' references unknown category '{category}'.", + context={"id": entry_id, "category": category}, + ) + ] + return [] + + +def _duplicate_id_messages(entries: Sequence[Mapping[str, Any]]) -> list[Message]: + seen: set[str] = set() + messages: list[Message] = [] + for entry in entries: + entry_id = entry.get("id") + if not isinstance(entry_id, str): + continue + if entry_id in seen: + messages.append( + Message( + code="SC0005", + severity=Severity.ERROR, + title="Duplicate Canonical ID", + description=f"Canonical id '{entry_id}' is defined more than once.", + context={"id": entry_id}, + ) + ) + seen.add(entry_id) + return messages + + +def _alias_messages(entries: Sequence[Mapping[str, Any]]) -> list[Message]: + """Aliases must be globally unique and never collide with a canonical id.""" + ids = {e["id"] for e in entries if isinstance(e.get("id"), str)} + messages: list[Message] = [] + seen_aliases: set[str] = set() + for entry in entries: + entry_id = entry.get("id", "") + for alias in entry.get("aliases", []): + if alias in ids: + messages.append( + _sc0004( + title="Alias collides with a canonical id", + description=( + f"Alias '{alias}' on entry '{entry_id}' equals a canonical id; " + "aliases must resolve directly (no chains)." + ), + entry_id=entry_id, + alias=alias, + ) + ) + if alias in seen_aliases: + messages.append( + _sc0004( + title="Duplicate alias", + description=( + f"Alias '{alias}' (entry '{entry_id}') is defined more than once." + ), + entry_id=entry_id, + alias=alias, + ) + ) + seen_aliases.add(alias) + return messages + + +def _expansion_messages(entries: Sequence[Mapping[str, Any]]) -> list[Message]: + """Expansion targets must exist and must not form cycles.""" + ids = {e["id"] for e in entries if isinstance(e.get("id"), str)} + graph = { + e["id"]: [t for t in e.get("expand", [])] for e in entries if isinstance(e.get("id"), str) + } + messages: list[Message] = [] + + # Missing targets. + for entry in entries: + entry_id = entry.get("id", "") + for target in entry.get("expand", []): + if target not in ids: + messages.append( + _sc0004( + title="Expansion target does not exist", + description=( + f"Entry '{entry_id}' expands to unknown canonical id '{target}'." + ), + entry_id=entry_id, + target=target, + ) + ) + + # Cycles (only over targets that exist), reported once per distinct cycle. + reported: set[frozenset[str]] = set() + for cycle in _find_cycles(graph, ids): + key = frozenset(cycle) + if key in reported: + continue + reported.add(key) + messages.append( + Message( + code="SC0003", + severity=Severity.ERROR, + title="Circular Expansion", + description=f"Expansion cycle detected: {' -> '.join(cycle)}.", + context={"id": cycle[0], "cycle": cycle}, + ) + ) + return messages + + +def _find_cycles(graph: Mapping[str, list[str]], ids: set[str]) -> list[list[str]]: + """Return the ids of each expansion cycle, deterministically.""" + cycles: list[list[str]] = [] + on_stack: dict[str, int] = {} + stack: list[str] = [] + visited: set[str] = set() + + def dfs(node: str) -> None: + visited.add(node) + on_stack[node] = len(stack) + stack.append(node) + for target in graph.get(node, []): + if target not in ids: + continue + if target in on_stack: + cycles.append(stack[on_stack[target] :]) + elif target not in visited: + dfs(target) + stack.pop() + del on_stack[node] + + for node in graph: # insertion order -> deterministic + if node not in visited: + dfs(node) + return cycles + + +def validate_knowledge_base(entries: Iterable[Mapping[str, Any]]) -> list[Message]: + """Validate a whole Knowledge Base and return every problem found. + + Args: + entries: All Knowledge Base entries (parsed JSON), across every file. + + Returns: + A deterministically-ordered list of :class:`Message`; empty when the + Knowledge Base is valid. + """ + entries = list(entries) + messages: list[Message] = [] + + for entry in entries: + entry_id = entry.get("id", "") + messages.extend(_schema_messages(entry, entry_id)) + messages.extend(_category_messages(entry, entry_id)) + + messages.extend(_duplicate_id_messages(entries)) + messages.extend(_alias_messages(entries)) + messages.extend(_expansion_messages(entries)) + return messages diff --git a/scripts/validate_knowledge_base.py b/scripts/validate_knowledge_base.py new file mode 100644 index 0000000..7e2217d --- /dev/null +++ b/scripts/validate_knowledge_base.py @@ -0,0 +1,79 @@ +"""CLI to validate a Knowledge Base directory (issue #9). + +Loads every ``*.json`` file under the given directory (each a JSON array of +entries), validates the whole Knowledge Base, and writes the findings as JSON to +stdout. Exit code is 0 when the Knowledge Base is valid, 1 otherwise. + +Usage: + python scripts/validate_knowledge_base.py knowledge_base/ +""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path + +# Allow running directly (python scripts/validate_knowledge_base.py) without install. +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from compiler.common.kb_validation import validate_knowledge_base # noqa: E402 +from compiler.common.result import Message, Severity # noqa: E402 + + +def load_entries(directory: Path) -> tuple[list[dict], list[Message]]: + """Load all entries from ``*.json`` files; return (entries, load_errors).""" + entries: list[dict] = [] + errors: list[Message] = [] + for path in sorted(directory.glob("*.json")): + try: + data = json.loads(path.read_text(encoding="utf-8")) + except json.JSONDecodeError as exc: + errors.append( + Message( + code="SC0004", + severity=Severity.FATAL, + title="Invalid JSON", + description=f"{path.name}: {exc}", + context={"id": path.name}, + ) + ) + continue + if isinstance(data, list): + entries.extend(data) + else: + errors.append( + Message( + code="SC0004", + severity=Severity.FATAL, + title="Invalid Knowledge Base file", + description=f"{path.name}: expected a JSON array of entries.", + context={"id": path.name}, + ) + ) + return entries, errors + + +def main(argv: list[str]) -> int: + if len(argv) != 1: + sys.stderr.write("usage: validate_knowledge_base.py \n") + return 2 + directory = Path(argv[0]) + if not directory.is_dir(): + sys.stderr.write(f"not a directory: {directory}\n") + return 2 + + entries, findings = load_entries(directory) + findings = findings + validate_knowledge_base(entries) + + report = { + "valid": not findings, + "entry_count": len(entries), + "findings": [m.to_json() for m in findings], + } + sys.stdout.write(json.dumps(report, indent=2, sort_keys=True) + "\n") + return 0 if not findings else 1 + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv[1:])) diff --git a/tests/test_kb_validation.py b/tests/test_kb_validation.py new file mode 100644 index 0000000..912faca --- /dev/null +++ b/tests/test_kb_validation.py @@ -0,0 +1,128 @@ +"""Tests for the Knowledge Base validation tooling (issue #9, §15.10, §26.4).""" + +from __future__ import annotations + +from compiler.common.kb_validation import validate_knowledge_base +from compiler.common.result import Severity + + +def entry(id_: str, **overrides) -> dict: + base = {"id": id_, "tags": [id_.replace("_", " ")], "category": "hair"} + base.update(overrides) + return base + + +def codes(messages) -> list[str]: + return [m.code for m in messages] + + +# --- passing --------------------------------------------------------------- + + +def test_valid_knowledge_base_passes_cleanly() -> None: + entries = [ + entry("long_hair", aliases=["lengthy hair"]), + entry("twin_braids", aliases=["double braids"], expand=["long_hair"]), + ] + assert validate_knowledge_base(entries) == [] + + +def test_empty_knowledge_base_passes() -> None: + assert validate_knowledge_base([]) == [] + + +# --- duplicate canonical id (SC0005) --------------------------------------- + + +def test_duplicate_id_reports_sc0005() -> None: + messages = validate_knowledge_base([entry("long_hair"), entry("long_hair")]) + assert "SC0005" in codes(messages) + dup = next(m for m in messages if m.code == "SC0005") + assert dup.context["id"] == "long_hair" + + +# --- invalid category (SC0008) --------------------------------------------- + + +def test_invalid_category_reports_sc0008() -> None: + messages = validate_knowledge_base([entry("x", category="hairstyle")]) + assert "SC0008" in codes(messages) + + +def test_valid_category_accepted() -> None: + assert validate_knowledge_base([entry("x", category="eyes")]) == [] + + +# --- circular expansion (SC0003) ------------------------------------------- + + +def test_circular_expansion_reports_sc0003() -> None: + entries = [entry("a", expand=["b"]), entry("b", expand=["a"])] + messages = validate_knowledge_base(entries) + assert "SC0003" in codes(messages) + + +def test_self_expansion_is_circular() -> None: + messages = validate_knowledge_base([entry("a", expand=["a"])]) + assert "SC0003" in codes(messages) + + +# --- other KB-structural failures (SC0004) --------------------------------- + + +def test_duplicate_alias_reports_sc0004() -> None: + entries = [entry("a", aliases=["shared"]), entry("b", aliases=["shared"])] + assert "SC0004" in codes(validate_knowledge_base(entries)) + + +def test_alias_colliding_with_id_reports_sc0004() -> None: + # "long_hair" is a canonical id; using it as an alias would create a chain. + entries = [entry("long_hair"), entry("b", aliases=["long_hair"])] + assert "SC0004" in codes(validate_knowledge_base(entries)) + + +def test_missing_expansion_target_reports_sc0004() -> None: + messages = validate_knowledge_base([entry("a", expand=["does_not_exist"])]) + assert "SC0004" in codes(messages) + + +def test_empty_tags_reported() -> None: + messages = validate_knowledge_base([entry("a", tags=[])]) + assert messages != [] + + +def test_unknown_field_reported() -> None: + messages = validate_knowledge_base([entry("a", colour="brown")]) + assert messages != [] + + +# --- reporting behaviour --------------------------------------------------- + + +def test_reports_all_problems_not_just_first() -> None: + entries = [ + entry("dup"), + entry("dup"), # duplicate id (SC0005) + entry("bad", category="nope"), # invalid category (SC0008) + ] + found = set(codes(validate_knowledge_base(entries))) + assert {"SC0005", "SC0008"}.issubset(found) + + +def test_every_message_has_code_id_and_description() -> None: + messages = validate_knowledge_base([entry("a", category="nope", expand=["missing"])]) + assert messages + for m in messages: + assert m.code + assert m.description + assert "id" in m.context + + +def test_findings_are_deterministic() -> None: + entries = [entry("dup"), entry("dup"), entry("bad", category="nope")] + assert codes(validate_knowledge_base(entries)) == codes(validate_knowledge_base(entries)) + + +def test_findings_carry_error_or_fatal_severity() -> None: + messages = validate_knowledge_base([entry("x", category="nope")]) + assert all(m.severity in (Severity.ERROR, Severity.FATAL) for m in messages) diff --git a/tests/test_kb_validation_cli.py b/tests/test_kb_validation_cli.py new file mode 100644 index 0000000..4f22b13 --- /dev/null +++ b/tests/test_kb_validation_cli.py @@ -0,0 +1,49 @@ +"""Tests for the Knowledge Base validation CLI (issue #9).""" + +from __future__ import annotations + +import importlib.util +import json +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parent.parent +SCRIPT = REPO_ROOT / "scripts" / "validate_knowledge_base.py" + + +def _load_cli(): + spec = importlib.util.spec_from_file_location("kb_cli", SCRIPT) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) # type: ignore[union-attr] + return module + + +def test_cli_passes_on_valid_kb(tmp_path, capsys) -> None: + (tmp_path / "hair.json").write_text( + json.dumps([{"id": "long_hair", "tags": ["long hair"], "category": "hair"}]), + encoding="utf-8", + ) + exit_code = _load_cli().main([str(tmp_path)]) + report = json.loads(capsys.readouterr().out) + assert exit_code == 0 + assert report["valid"] is True + assert report["entry_count"] == 1 + + +def test_cli_fails_on_invalid_kb(tmp_path, capsys) -> None: + (tmp_path / "bad.json").write_text( + json.dumps([{"id": "x", "tags": ["x"], "category": "nope"}]), + encoding="utf-8", + ) + exit_code = _load_cli().main([str(tmp_path)]) + report = json.loads(capsys.readouterr().out) + assert exit_code == 1 + assert report["valid"] is False + assert any(f["code"] == "SC0008" for f in report["findings"]) + + +def test_cli_reports_invalid_json(tmp_path, capsys) -> None: + (tmp_path / "broken.json").write_text("{not json", encoding="utf-8") + exit_code = _load_cli().main([str(tmp_path)]) + report = json.loads(capsys.readouterr().out) + assert exit_code == 1 + assert any(f["code"] == "SC0004" for f in report["findings"])