""" tests/track_b/test_story_8_03.py Story 8.03 (Track B): MetaArchitect — Scar-Driven Structural Analysis Black Box Tests (BB1–BB4): BB1 10 failed sagas in last 7 days → at least 1 bottleneck identified BB2 scope="epistemic" returned for prompt-fixable issues BB3 Analysis written to meta_architect_log.jsonl (tmp_path) BB4 Empty scars + empty sagas → ArchitectureAnalysis with empty bottlenecks, scope="epistemic" White Box Tests (WB1–WB4): WB1 Qdrant query groups similar scars (cluster, not per-scar) WB2 lookback_days parameter drives Postgres interval correctly WB3 _determine_scope returns "ontological" when any bottleneck affects .py files WB4 Log entry is valid JSON with required fields ALL tests use mocks — no real Qdrant or Postgres connections. tmp_path is used for all file I/O. """ from __future__ import annotations import json import sys from pathlib import Path from unittest.mock import MagicMock, patch, call import pytest # --------------------------------------------------------------------------- # Path bootstrap # --------------------------------------------------------------------------- GENESIS_ROOT = "/mnt/e/genesis-system" if GENESIS_ROOT not in sys.path: sys.path.insert(0, GENESIS_ROOT) # --------------------------------------------------------------------------- # Imports under test # --------------------------------------------------------------------------- from core.evolution.meta_architect import ( # noqa: E402 MetaArchitect, ArchitectureAnalysis, Bottleneck, FixProposal, ) from core.evolution import ( # noqa: E402 MetaArchitect as MetaArchitectFromInit, ArchitectureAnalysis as ArchitectureAnalysisFromInit, Bottleneck as BottleneckFromInit, FixProposal as FixProposalFromInit, ) # --------------------------------------------------------------------------- # Helpers / Factories # --------------------------------------------------------------------------- def _make_mock_qdrant(scars: list[dict]) -> MagicMock: """ Build a mock Qdrant client whose scroll() returns ``scars``. Each scar dict should have "id", "payload" (with "description"), and optionally "vector". The scroll() return value is a list of point-like dicts with the same shape. """ mock = MagicMock() # Convert plain scar dicts into the point-dict format _query_scars expects points = [ { "id": s["id"], "payload": {"description": s.get("description", "")}, "vector": s.get("vector", []), } for s in scars ] mock.scroll.return_value = points return mock def _make_mock_pg(sagas: list[dict]) -> MagicMock: """ Build a mock psycopg2-like connection whose cursor().fetchall() returns rows in the format (id, description, error_trace). """ mock_conn = MagicMock() mock_cursor = MagicMock() rows = [(s["id"], s.get("description", ""), s.get("error_trace", "")) for s in sagas] mock_cursor.fetchall.return_value = rows mock_conn.cursor.return_value = mock_cursor return mock_conn def _make_architect( scars: list[dict], sagas: list[dict], tmp_path: Path, ) -> MetaArchitect: """Factory: MetaArchitect with mocked clients and a tmp log path.""" qdrant = _make_mock_qdrant(scars) pg = _make_mock_pg(sagas) log = tmp_path / "meta_architect_log.jsonl" return MetaArchitect(qdrant_client=qdrant, pg_connection=pg, log_path=log) # =========================================================================== # BLACK BOX TESTS # =========================================================================== def test_bb1_ten_failed_sagas_produce_at_least_one_bottleneck(tmp_path): """BB1: 10 PARTIAL_FAIL sagas in last 7 days → at least 1 bottleneck identified.""" sagas = [ {"id": f"saga-{i:02d}", "description": "timeout connecting to redis", "error_trace": ""} for i in range(10) ] # Provide scars that match the saga theme scars = [ {"id": f"scar-{i:02d}", "description": "timeout connecting to redis"} for i in range(3) ] architect = _make_architect(scars, sagas, tmp_path) analysis = architect.analyze(lookback_days=7) assert isinstance(analysis, ArchitectureAnalysis) assert len(analysis.bottlenecks) >= 1, ( f"Expected ≥1 bottleneck but got {len(analysis.bottlenecks)}" ) def test_bb2_epistemic_scope_for_prompt_fixable_issues(tmp_path): """BB2: Bottleneck description containing 'prompt' → scope='epistemic'.""" scars = [ {"id": "scar-01", "description": "prompt instruction ambiguous"}, {"id": "scar-02", "description": "prompt instruction ambiguous"}, ] sagas = [ {"id": "saga-01", "description": "prompt instruction ambiguous", "error_trace": ""} ] architect = _make_architect(scars, sagas, tmp_path) analysis = architect.analyze(lookback_days=7) assert analysis.scope == "epistemic", ( f"Expected scope='epistemic' but got '{analysis.scope}'" ) def test_bb3_analysis_written_to_jsonl_log(tmp_path): """BB3: analyze() writes at least one JSON line to meta_architect_log.jsonl.""" scars = [{"id": "scar-01", "description": "network timeout"}] sagas = [{"id": "saga-01", "description": "network timeout failure", "error_trace": ""}] architect = _make_architect(scars, sagas, tmp_path) log_path = tmp_path / "meta_architect_log.jsonl" architect.log_path = log_path architect.analyze(lookback_days=7) assert log_path.exists(), "Log file was not created." lines = [line for line in log_path.read_text(encoding="utf-8").splitlines() if line.strip()] assert len(lines) >= 1, "Log file has no entries." # Each line must be parseable JSON for line in lines: parsed = json.loads(line) assert isinstance(parsed, dict) def test_bb4_empty_scars_and_sagas_returns_empty_analysis(tmp_path): """BB4: Empty scars + empty sagas → ArchitectureAnalysis(bottlenecks=[], scope='epistemic').""" architect = _make_architect([], [], tmp_path) analysis = architect.analyze(lookback_days=7) assert isinstance(analysis, ArchitectureAnalysis) assert analysis.bottlenecks == [], ( f"Expected empty bottlenecks, got: {analysis.bottlenecks}" ) assert analysis.scope == "epistemic", ( f"Expected scope='epistemic', got: '{analysis.scope}'" ) # =========================================================================== # WHITE BOX TESTS # =========================================================================== def test_wb1_similar_scars_are_grouped_into_one_cluster(tmp_path): """ WB1: _cluster_scars groups semantically similar scars into a single cluster rather than creating one cluster per scar. We use identical descriptions (no real vectors) to force text-based clustering. 3 scars with the same description → 1 cluster of size 3 (not 3 clusters of size 1). """ scars = [ {"id": "scar-01", "description": "redis connection refused"}, {"id": "scar-02", "description": "redis connection refused"}, {"id": "scar-03", "description": "redis connection refused"}, ] architect = _make_architect(scars, [], tmp_path) clusters = architect._cluster_scars(scars) # All three identical-description scars land in one cluster cluster_sizes = [len(c) for c in clusters] assert max(cluster_sizes) == 3, ( f"Expected at least 1 cluster of size 3, got cluster sizes: {cluster_sizes}" ) assert len(clusters) == 1, ( f"Expected 1 cluster but got {len(clusters)}: {cluster_sizes}" ) def test_wb2_lookback_days_drives_postgres_interval(tmp_path): """ WB2: _query_sagas uses the lookback_days parameter in the SQL INTERVAL clause. """ sagas = [{"id": "saga-01", "description": "test", "error_trace": ""}] mock_conn = _make_mock_pg(sagas) architect = MetaArchitect( qdrant_client=_make_mock_qdrant([]), pg_connection=mock_conn, log_path=tmp_path / "log.jsonl", ) # Run with a custom lookback_days value architect.analyze(lookback_days=14) # Verify cursor.execute was called and the SQL contains the correct interval mock_cursor = mock_conn.cursor.return_value assert mock_cursor.execute.called, "cursor.execute() was never called" executed_sql = mock_cursor.execute.call_args[0][0] assert "14 days" in executed_sql, ( f"Expected '14 days' in SQL but got: {executed_sql!r}" ) def test_wb3_determine_scope_returns_ontological_for_py_file_bottleneck(tmp_path): """ WB3: _determine_scope returns 'ontological' when a bottleneck description references a .py file or code-level artefact. """ architect = _make_architect([], [], tmp_path) # Bottleneck whose description clearly points to a code artefact bottlenecks_ontological = [ Bottleneck( description="failure in router.py causes dispatch loop", frequency=3, affected_saga_ids=["saga-01"], scar_ids=["scar-01"], ) ] scope = architect._determine_scope(bottlenecks_ontological) assert scope == "ontological", ( f"Expected 'ontological' for .py-referencing bottleneck, got '{scope}'" ) # Bottleneck that is purely a prompt issue → epistemic bottlenecks_epistemic = [ Bottleneck( description="prompt instruction unclear causing wrong response", frequency=2, affected_saga_ids=["saga-02"], scar_ids=["scar-02"], ) ] scope_ep = architect._determine_scope(bottlenecks_epistemic) assert scope_ep == "epistemic", ( f"Expected 'epistemic' for prompt-fixable bottleneck, got '{scope_ep}'" ) def test_wb4_log_entry_is_valid_json_with_required_fields(tmp_path): """ WB4: Each line in the log is valid JSON containing: timestamp, lookback_days, scope, bottleneck_count, fix_count, bottlenecks (list), recommended_fixes (list) """ scars = [{"id": "scar-01", "description": "timeout in memory lookup"}] sagas = [{"id": "saga-01", "description": "memory lookup timeout", "error_trace": ""}] log_path = tmp_path / "meta_architect_log.jsonl" architect = MetaArchitect( qdrant_client=_make_mock_qdrant(scars), pg_connection=_make_mock_pg(sagas), log_path=log_path, ) architect.analyze(lookback_days=5) assert log_path.exists(), "Log file was not written." lines = [l for l in log_path.read_text(encoding="utf-8").splitlines() if l.strip()] assert lines, "No log lines were written." required_fields = { "timestamp", "lookback_days", "scope", "bottleneck_count", "fix_count", "bottlenecks", "recommended_fixes", } for line in lines: entry = json.loads(line) missing = required_fields - set(entry.keys()) assert not missing, f"Log entry missing fields: {missing}. Entry: {entry}" # Type checks assert isinstance(entry["timestamp"], str) and entry["timestamp"] assert isinstance(entry["lookback_days"], int) assert entry["scope"] in ("epistemic", "ontological") assert isinstance(entry["bottleneck_count"], int) assert isinstance(entry["fix_count"], int) assert isinstance(entry["bottlenecks"], list) assert isinstance(entry["recommended_fixes"], list) # =========================================================================== # Package-level import test # =========================================================================== def test_pkg_exports_meta_architect_classes(): """core.evolution.__init__ exports MetaArchitect, ArchitectureAnalysis, Bottleneck, FixProposal.""" assert MetaArchitectFromInit is MetaArchitect assert ArchitectureAnalysisFromInit is ArchitectureAnalysis assert BottleneckFromInit is Bottleneck assert FixProposalFromInit is FixProposal # =========================================================================== # Dataclass structural tests # =========================================================================== def test_dataclass_fields_exist(): """All four dataclasses expose their required fields.""" import dataclasses # ArchitectureAnalysis assert dataclasses.is_dataclass(ArchitectureAnalysis) aa_fields = {f.name for f in dataclasses.fields(ArchitectureAnalysis)} assert "bottlenecks" in aa_fields assert "recommended_fixes" in aa_fields assert "scope" in aa_fields # Bottleneck assert dataclasses.is_dataclass(Bottleneck) b_fields = {f.name for f in dataclasses.fields(Bottleneck)} assert "description" in b_fields assert "frequency" in b_fields assert "affected_saga_ids" in b_fields assert "scar_ids" in b_fields # FixProposal assert dataclasses.is_dataclass(FixProposal) fp_fields = {f.name for f in dataclasses.fields(FixProposal)} assert "target_file" in fp_fields assert "change_type" in fp_fields assert "rationale" in fp_fields def test_architecture_analysis_defaults(): """ArchitectureAnalysis has sensible defaults.""" aa = ArchitectureAnalysis() assert aa.bottlenecks == [] assert aa.recommended_fixes == [] assert aa.scope == "epistemic" # =========================================================================== # Cosine-similarity clustering test (vector path) # =========================================================================== def test_cosine_clustering_identical_vectors_produce_one_cluster(tmp_path): """Scars with identical vectors cluster into one group (cosine sim = 1.0 ≥ 0.85).""" vec = [1.0, 0.0, 0.0] scars = [ {"id": "scar-01", "description": "oom error", "vector": vec}, {"id": "scar-02", "description": "oom error", "vector": vec}, {"id": "scar-03", "description": "oom error", "vector": vec}, ] architect = _make_architect([], [], tmp_path) clusters = architect._cluster_scars(scars) assert len(clusters) == 1 assert len(clusters[0]) == 3 def test_cosine_clustering_orthogonal_vectors_produce_separate_clusters(tmp_path): """Orthogonal vectors (cosine sim = 0.0) each get their own cluster.""" scars = [ {"id": "scar-01", "description": "err A", "vector": [1.0, 0.0, 0.0]}, {"id": "scar-02", "description": "err B", "vector": [0.0, 1.0, 0.0]}, {"id": "scar-03", "description": "err C", "vector": [0.0, 0.0, 1.0]}, ] architect = _make_architect([], [], tmp_path) clusters = architect._cluster_scars(scars) assert len(clusters) == 3, ( f"Expected 3 clusters for orthogonal vectors, got {len(clusters)}" ) # =========================================================================== # Dependency injection tests (no-client path) # =========================================================================== def test_no_clients_returns_empty_analysis(tmp_path): """MetaArchitect with no injected clients returns empty analysis, writes log.""" log_path = tmp_path / "log.jsonl" architect = MetaArchitect(log_path=log_path) analysis = architect.analyze(lookback_days=7) assert analysis.bottlenecks == [] assert analysis.scope == "epistemic" assert log_path.exists() # =========================================================================== # Standalone runner # =========================================================================== if __name__ == "__main__": import traceback class _TmpDir: """Minimal tmp_path stand-in for standalone runs.""" def __init__(self): import tempfile, os self._dir = tempfile.mkdtemp() def __truediv__(self, name: str) -> Path: return Path(self._dir) / name tests = [ ("BB1: 10 sagas → ≥1 bottleneck", test_bb1_ten_failed_sagas_produce_at_least_one_bottleneck), ("BB2: prompt scars → scope=epistemic", test_bb2_epistemic_scope_for_prompt_fixable_issues), ("BB3: log written to JSONL", test_bb3_analysis_written_to_jsonl_log), ("BB4: empty inputs → empty analysis", test_bb4_empty_scars_and_sagas_returns_empty_analysis), ("WB1: similar scars → 1 cluster", test_wb1_similar_scars_are_grouped_into_one_cluster), ("WB2: lookback_days in SQL", test_wb2_lookback_days_drives_postgres_interval), ("WB3: .py bottleneck → ontological", test_wb3_determine_scope_returns_ontological_for_py_file_bottleneck), ("WB4: log fields valid", test_wb4_log_entry_is_valid_json_with_required_fields), ] passed = 0 for name, fn in tests: tmp = _TmpDir() try: fn(tmp) print(f" [PASS] {name}") passed += 1 except Exception as exc: print(f" [FAIL] {name}: {exc}") traceback.print_exc() print(f"\n{passed}/{len(tests)} tests passed") if passed == len(tests): print("ALL TESTS PASSED — Story 8.03") else: sys.exit(1)