"""Tests for compiler/stages/capture_import.py.""" from __future__ import annotations import hashlib import json import struct import wave from pathlib import Path import numpy as np import yaml from compiler.stages.capture_import import import_capture from compiler.stages.index import build_index from runtime.conversation.retrieval import retrieve from runtime.providers.fake import FakeProvider from shared.capture import load_capture class _HashEmbedder: def encode( self, sentences: list[str], *, show_progress_bar: bool = False, normalize_embeddings: bool = False, ) -> np.ndarray: vecs = [] for s in sentences: h = hashlib.sha256(s.encode()).digest() raw = (h * 2)[:32] vec = np.frombuffer(raw, dtype=np.uint8).astype(np.float32) - 128.0 if normalize_embeddings: norm = np.linalg.norm(vec) if norm > 0: vec /= norm vecs.append(vec) return np.stack(vecs) async def test_capture_import_merges_profile_quotes_and_voice(tmp_path, monkeypatch): capture_root = tmp_path / "captures" capture_dir = capture_root / "fixture-capture" capture_dir.mkdir(parents=True) monkeypatch.setenv("MENTORCORE_CAPTURES_ROOT", str(capture_root)) _write_capture_tree(capture_dir) capture = load_capture("fixture-capture") build_dir = tmp_path / "build" style_dir = build_dir / "style" index_dir = build_dir / "index" style_dir.mkdir(parents=True) index_dir.mkdir(parents=True) (style_dir / "profile.md").write_text( "# Tone\nMachine tone.\n\n# Style Summary\nMachine summary.\n", encoding="utf-8", ) (style_dir / "exemplars.jsonl").write_text( json.dumps({"text": "Machine exemplar."}) + "\n", encoding="utf-8", ) build_index( {"machine": ["An unrelated passage about structure and routine."]}, index_dir, embedder=_HashEmbedder(), overwrite=True, ) fixture_wav = tmp_path / "fixture.wav" _write_wav(fixture_wav, seconds=2.0) def fake_downloader(reference, voice_dir: Path) -> Path: return fixture_wav provider = FakeProvider.single("I hold to this worldview myself, stated plainly.") result = await import_capture( build_dir, capture, provider=provider, embedder=_HashEmbedder(), reference_downloader=fake_downloader, ) profile = (style_dir / "profile.md").read_text(encoding="utf-8") assert profile.startswith("# Capture Philosophy") assert "Fixture worldview." in profile assert "# Machine Extraction" in profile assert result.style_summary == "I hold to this worldview myself, stated plainly." assert "Fixture worldview." not in result.style_summary assert result.verified_quotes == 1 assert result.skipped_model_memory_quotes == 1 exemplars = [ json.loads(line) for line in (style_dir / "exemplars.jsonl").read_text().splitlines() if line.strip() ] assert exemplars[0]["synthetic"] is True assert exemplars[0]["text"] == "Answer in the capture voice." quote_stats = json.loads((index_dir / "quote_stats.json").read_text()) assert quote_stats["verified_quotes"] == 1 assert quote_stats["skipped_model_memory_quotes"] == 1 verified_quotes = [ json.loads(line) for line in (index_dir / "verified_quotes.jsonl").read_text().splitlines() if line.strip() ] assert len(verified_quotes) == 1 assert verified_quotes[0]["chunk_id"].startswith("quote:") voice_config = yaml.safe_load((build_dir / "voice" / "voice.yaml").read_text()) assert voice_config["engine"] == "f5" assert voice_config["policy"] == "cloned-personal" reference_audio = Path(voice_config["reference_audio"]) assert reference_audio.exists() with wave.open(str(reference_audio), "rb") as wf: assert wf.getnframes() > 0 retrieved = retrieve( "Virtue is the only good.", index_dir, top_k=2, embedder=_HashEmbedder(), ) assert retrieved[0]["chunk_id"].startswith("quote:") assert retrieved[0]["text"] == "Virtue is the only good." def _write_capture_tree(capture_dir: Path) -> None: (capture_dir / "philosophy.md").write_text( ( "# Identity\n\nFixture worldview.\n\n## Core worldview\n\n" "Fixture worldview. Keep the answer short.\n" ), encoding="utf-8", ) (capture_dir / "style.md").write_text( "# Register\n\nMeasured and direct.\n", encoding="utf-8", ) (capture_dir / "sources.md").write_text("# Sources\n\n1. Fixture source.\n", encoding="utf-8") (capture_dir / "capture.yaml").write_text( yaml.safe_dump( { "schema_version": 1, "slug": "fixture-capture", "display_name": "Fixture Capture", "born": 1939, "died": None, "voice_policy": "cloned-personal", "captured_by": "gpt-5-codex", "captured_at": "2026-07-17", "capture_mode": "hybrid", "capture_focus": "fixture", "confidence": "medium", "notes": "Fixture capture for import tests.", }, sort_keys=False, ), encoding="utf-8", ) (capture_dir / "quotes.jsonl").write_text( "\n".join( [ json.dumps( { "id": "q001", "text": "Virtue is the only good.", "source": "Fixture source", "date": None, "topics": ["virtue"], "provenance": "verified", "verify_url": "https://example.com/verified", "context": None, } ), json.dumps( { "id": "q002", "text": "The model remembers this line.", "source": "Fixture source", "date": None, "topics": ["memory"], "provenance": "model-memory", "verify_url": None, "context": None, } ), ] ) + "\n", encoding="utf-8", ) (capture_dir / "qa-exemplars.jsonl").write_text( json.dumps( { "id": "x001", "question": "How should I respond?", "answer": "Answer in the capture voice.", "topics": ["fixture"], "synthetic": True, "register": "coaching-debrief", } ) + "\n", encoding="utf-8", ) (capture_dir / "voice.yaml").write_text( yaml.safe_dump( { "policy": "cloned-personal", "references": [ { "url": "https://example.com/reference", "kind": "interview", "title": "Fixture Reference", "clean_segments": ["00:00-00:01"], "quality_notes": "Fixture segment.", } ], "pacing_notes": "Measured and direct.", "stylized_fallback": { "description": "Fixture fallback.", "kokoro_voicepack_hint": None, }, }, sort_keys=False, ), encoding="utf-8", ) (capture_dir / "eval-questions.jsonl").write_text( json.dumps( { "id": "e001", "question": "What matters most?", "expect": "A fixture answer.", "kind": "core-belief", } ) + "\n", encoding="utf-8", ) def _write_wav(path: Path, *, seconds: float) -> None: sample_rate = 16000 n_frames = int(sample_rate * seconds) pcm = struct.pack(f"<{n_frames}h", *([0] * n_frames)) with wave.open(str(path), "wb") as wf: wf.setnchannels(1) wf.setsampwidth(2) wf.setframerate(sample_rate) wf.writeframes(pcm)