compounding-intelligence/scripts/test_knowledge_synthesizer.py

#!/usr/bin/env python3
"""
Tests for knowledge_synthesizer.py — zero-shot knowledge synthesis pipeline.

Run with: python3 scripts/test_knowledge_synthesizer.py
Or via pytest: pytest scripts/test_knowledge_synthesizer.py
"""

import json
import os
import sys
import os
import tempfile
from pathlib import Path

# Add scripts dir to path for importing sibling module
SCRIPT_DIR = Path(__file__).resolve().parent
sys.path.insert(0, str(SCRIPT_DIR))

import importlib.util
spec = importlib.util.spec_from_file_location(
    "ks", os.path.join(str(SCRIPT_DIR), "knowledge_synthesizer.py")
)
ks = importlib.util.module_from_spec(spec)
spec.loader.exec_module(ks)


# ── Test data helpers ─────────────────────────────────────────────

SAMPLE_FACTS = [
    {
        "id": "global:pitfall:001",
        "fact": "Branch protection requires 1 approval on main for Gitea merges",
        "category": "pitfall",
        "domain": "global",
        "confidence": 0.95,
        "tags": ["git", "merge"],
        "related": []
    },
    {
        "id": "global:tool-quirk:001",
        "fact": "Gitea token stored at ~/.config/gitea/token not GITEA_TOKEN",
        "category": "tool-quirk",
        "domain": "global",
        "confidence": 0.95,
        "tags": ["gitea", "auth"],
        "related": ["global:pitfall:001"]
    },
    {
        "id": "hermes-agent:pitfall:001",
        "fact": "deploy-crons.py leaves jobs in mixed model format",
        "category": "pitfall",
        "domain": "hermes-agent",
        "confidence": 0.95,
        "tags": ["cron"],
        "related": []
    },
]


def make_index(facts, tmp_dir: Path) -> Path:
    index = {
        "version": 1,
        "last_updated": "2026-04-13T20:00:00Z",
        "total_facts": len(facts),
        "facts": facts,
    }
    path = tmp_dir / "index.json"
    with open(path, "w") as f:
        json.dump(index, f)
    return path


# ── Unit tests ────────────────────────────────────────────────────

def test_next_sequence():
    facts = SAMPLE_FACTS[:2]
    seq = ks.next_sequence(facts, "global", "pitfall")
    assert seq == 2, f"Expected 2, got {seq}"

    seq2 = ks.next_sequence(facts, "hermes-agent", "pitfall")
    assert seq2 == 1, f"Expected 1, got {seq2}"


def test_generate_id():
    facts = SAMPLE_FACTS[:2]
    fid = ks.generate_id("global", "fact", facts)
    assert fid == "global:fact:001", f"Got {fid}"


def test_facts_are_unrelated():
    f1 = SAMPLE_FACTS[0]  # unrelated to hermes-agent pitfall
    f2 = SAMPLE_FACTS[2]
    assert ks.facts_are_unrelated(f1, f2) is True

    f3 = SAMPLE_FACTS[1]  # related to f1
    assert ks.facts_are_unrelated(f1, f3) is False


def test_find_candidate_pair():
    facts = SAMPLE_FACTS
    pair = ks.find_candidate_pair(facts)
    assert pair is not None, "Should find an unrelated pair"
    f1, f2 = pair
    assert ks.facts_are_unrelated(f1, f2), "Returned pair must be unrelated"


def test_parse_synthesis_response_raw_json():
    content = '{"hypothesis": "test connection", "plausibility": 0.8, "bridging_concepts": ["x"], "suggested_tags": ["a"]}'
    result = ks.parse_synthesis_response(content)
    assert result is not None
    assert result["hypothesis"] == "test connection"
    assert result["plausibility"] == 0.8


def test_parse_synthesis_response_markdown_wrapped():
    content = '```json\n{"hypothesis": "wrapped", "plausibility": 0.5}\n```'
    result = ks.parse_synthesis_response(content)
    assert result is not None
    assert result["hypothesis"] == "wrapped"


def test_parse_synthesis_response_invalid():
    assert ks.parse_synthesis_response("not json") is None
    assert ks.parse_synthesis_response('{"nohypothesis": 1}') is None


def test_heuristic_synthesis():
    f1 = SAMPLE_FACTS[0]
    f2 = SAMPLE_FACTS[2]
    result = ks.heuristic_synthesis(f1, f2)
    assert "hypothesis" in result
    assert "plausibility" in result
    assert result["plausibility"] == 0.4
    assert "bridging_concepts" in result
    assert "suggested_tags" in result


def test_is_duplicate():
    facts = [{"fact": "existing fact", "id": "test:1"}]
    assert ks.is_duplicate("existing fact", facts) is True
    assert ks.is_duplicate("new fact", facts) is False


def test_store_synthesis_integration():
    """Integration test: pick a real candidate pair and store a mock synthesis."""
    with tempfile.TemporaryDirectory() as tmp:
        tmp_path = Path(tmp)
        # Create fake knowledge dir with index
        kdir = tmp_path / "knowledge"
        kdir.mkdir()
        index = {
            "version": 1,
            "last_updated": "2026-04-13T20:00:00Z",
            "total_facts": 3,
            "facts": SAMPLE_FACTS
        }
        with open(kdir / "index.json", "w") as f:
            json.dump(index, f)

        # Mock synthesis
        synth = {
            "hypothesis": "Test synthesized pattern",
            "plausibility": 0.8,
            "bridging_concepts": ["test"],
            "suggested_tags": ["test"]
        }
        source_ids = [SAMPLE_FACTS[0]['id'], SAMPLE_FACTS[2]['id']]

        # Temporarily override KNOWLEDGE_DIR path for test
        original_kdir = ks.KNOWLEDGE_DIR
        ks.KNOWLEDGE_DIR = kdir
        try:
            stored = ks.store_synthesis(synth, source_ids, index, threshold=0.5)
            assert stored is True
            assert index['total_facts'] == 4
            new_fact = index['facts'][-1]
            assert new_fact['fact'] == "Test synthesized pattern"
            assert new_fact['category'] == "pattern"
            assert new_fact['domain'] == "global"
            assert new_fact['related'] == source_ids
            assert new_fact['id'].startswith("global:pattern:")

            # Check YAML appended
            yaml_path = kdir / "global" / "patterns.yaml"
            assert yaml_path.exists()
            content = yaml_path.read_text()
            assert "Test synthesized pattern" in content
        finally:
            ks.KNOWLEDGE_DIR = original_kdir


# ── Smoke test ────────────────────────────────────────────────────

def test_smoke_synthesizer_info():
    """Sanity check: script can at least load and report current knowledge state."""
    index = ks.load_index()
    total = index.get('total_facts', 0)
    facts = index.get('facts', [])
    print(f"\nKnowledge store contains {total} facts across {len(set(f['domain'] for f in facts))} domains")
    assert total >= 0

# Import os for test
import os

if __name__ == "__main__":
    print("Running knowledge_synthesizer tests...\n")
    passed = 0
    failed = 0

    tests = [
        test_next_sequence,
        test_generate_id,
        test_facts_are_unrelated,
        test_find_candidate_pair,
        test_parse_synthesis_response_raw_json,
        test_parse_synthesis_response_markdown_wrapped,
        test_parse_synthesis_response_invalid,
        test_heuristic_synthesis,
        test_is_duplicate,
        test_store_synthesis_integration,
        test_smoke_synthesizer_info,
    ]

    for test in tests:
        try:
            test()
            print(f"  ✓ {test.__name__}")
            passed += 1
        except Exception as e:
            import traceback; traceback.print_exc(); print(f"  ✗ {test.__name__}: {e}")
            failed += 1

    print(f"\n{passed} passed, {failed} failed")
    sys.exit(0 if failed == 0 else 1)