""" tests/track_b/test_story_8_08.py Story 8.08: Tier1AutonomousUpdater — Epistemic Self-Updates Black Box Tests (BB1–BB4): BB1 Tier 1 update → KG entity file created in kg_base_path (tmp_path) BB2 No .py files modified by Tier 1 update (check all written files) BB3 Audit entry written to tier1_updates.jsonl (tmp_path) BB4 apply_tier1 returns Tier1Result with correct counts White Box Tests (WB1–WB4): WB1 Qdrant upsert uses "genesis_scars" collection name (mock verified) WB2 Rules file updates are append-only (existing content unchanged, new at end) WB3 KG entity file is valid JSONL (each line parseable as JSON) WB4 Prompt updates only create .md or .txt files (never .py) ALL tests use mocks for Qdrant. ALL file I/O uses tmp_path. """ from __future__ import annotations import json import sys from pathlib import Path from unittest.mock import MagicMock, 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.tier1_autonomous_updater import ( # noqa: E402 Tier1AutonomousUpdater, Tier1Result, ) from core.evolution.meta_architect import ( # noqa: E402 ArchitectureAnalysis, Bottleneck, FixProposal, ) # --------------------------------------------------------------------------- # Helpers — build a realistic ArchitectureAnalysis for tests # --------------------------------------------------------------------------- def _make_analysis( scope: str = "epistemic", num_bottlenecks: int = 2, fix_target: str = "config/prompts/system_prompt.md", ) -> ArchitectureAnalysis: """Return a populated ArchitectureAnalysis for use in tests.""" bottlenecks = [ Bottleneck( description=f"test bottleneck {i}", frequency=i + 1, affected_saga_ids=[f"saga_{i}"], scar_ids=[f"sc_{i}"], ) for i in range(num_bottlenecks) ] fixes = [ FixProposal( target_file=fix_target, change_type="epistemic", rationale=f"improve prompt quality iteration {i}", ) for i in range(num_bottlenecks) ] return ArchitectureAnalysis( bottlenecks=bottlenecks, recommended_fixes=fixes, scope=scope, ) def _make_updater(tmp_path: Path, qdrant_client=None) -> Tier1AutonomousUpdater: """Factory: Tier1AutonomousUpdater with all paths under tmp_path.""" return Tier1AutonomousUpdater( qdrant_client=qdrant_client, kg_base_path=str(tmp_path / "kg_entities"), prompts_dir=str(tmp_path / "prompts"), rules_file=str(tmp_path / "GLOBAL_GENESIS_RULES.md"), audit_log_path=str(tmp_path / "tier1_updates.jsonl"), ) # =========================================================================== # BLACK BOX TESTS # =========================================================================== def test_bb1_kg_entity_file_created(tmp_path): """ BB1: After apply_tier1(), a KG entity JSONL file exists at {kg_base_path}/epoch_{epoch_id}_learnings.jsonl. """ updater = _make_updater(tmp_path) analysis = _make_analysis(scope="epistemic") updater.apply_tier1(analysis, epoch_id="test_epoch_001") kg_dir = tmp_path / "kg_entities" expected_file = kg_dir / "epoch_test_epoch_001_learnings.jsonl" assert kg_dir.exists(), "KG base directory was not created" assert expected_file.exists(), ( f"Expected KG entity file not found: {expected_file}" ) def test_bb2_no_py_files_modified(tmp_path): """ BB2: None of the files written by apply_tier1() have a .py extension. We walk ALL files under tmp_path after the call and assert that no .py file was created or modified. """ updater = _make_updater(tmp_path) # Include a fix that references a .py file — Tier 1 must ignore it analysis = ArchitectureAnalysis( bottlenecks=[ Bottleneck( description="prompt instruction unclear", frequency=3, affected_saga_ids=["s1"], scar_ids=["sc1"], ) ], recommended_fixes=[ FixProposal( target_file="core/evolution/some_module.py", # must be IGNORED change_type="epistemic", rationale="should not create .py files", ), FixProposal( target_file="config/prompts/system_prompt.md", # allowed change_type="prompt_update", rationale="safe update", ), ], scope="epistemic", ) updater.apply_tier1(analysis, epoch_id="no_py_test") # Collect all files written under tmp_path all_files = list(tmp_path.rglob("*")) py_files = [f for f in all_files if f.is_file() and f.suffix == ".py"] assert py_files == [], ( f"Tier 1 must NOT create .py files. Found: {[str(f) for f in py_files]}" ) def test_bb3_audit_entry_written(tmp_path): """ BB3: apply_tier1() appends one JSON line to {audit_log_path}. The entry must contain: timestamp, epoch_id, kg_entities_added, scars_updated, prompts_updated, rules_updated. """ updater = _make_updater(tmp_path) analysis = _make_analysis(scope="epistemic") updater.apply_tier1(analysis, epoch_id="audit_test_epoch") audit_file = tmp_path / "tier1_updates.jsonl" assert audit_file.exists(), "Audit file was not created" lines = [l for l in audit_file.read_text(encoding="utf-8").splitlines() if l.strip()] assert len(lines) == 1, f"Expected 1 audit line, got {len(lines)}" entry = json.loads(lines[0]) required_keys = { "timestamp", "epoch_id", "kg_entities_added", "scars_updated", "prompts_updated", "rules_updated", } missing = required_keys - set(entry.keys()) assert not missing, f"Audit entry missing fields: {missing}" assert entry["epoch_id"] == "audit_test_epoch" assert isinstance(entry["kg_entities_added"], int) assert isinstance(entry["scars_updated"], int) assert isinstance(entry["prompts_updated"], int) assert isinstance(entry["rules_updated"], int) def test_bb4_apply_tier1_returns_correct_counts(tmp_path): """ BB4: apply_tier1 returns a Tier1Result with correct non-negative integer counts. We check that: - kg_entities_added > 0 (bottlenecks produce entities) - prompts_updated > 0 (epistemic fix with .md target) - rules_updated == 1 (epistemic scope with bottlenecks) - result is a Tier1Result instance """ updater = _make_updater(tmp_path) analysis = _make_analysis( scope="epistemic", num_bottlenecks=2, fix_target="config/prompts/system_prompt.md", ) result = updater.apply_tier1(analysis, epoch_id="counts_test") assert isinstance(result, Tier1Result), ( f"Expected Tier1Result instance, got {type(result)}" ) assert result.kg_entities_added > 0, ( "Expected at least one KG entity added for 2 bottlenecks" ) assert result.prompts_updated > 0, ( "Expected at least one prompt update for .md fix proposals" ) assert result.rules_updated == 1, ( "Expected exactly 1 rule added for epistemic bottlenecks" ) # All fields must be non-negative integers assert result.scars_updated >= 0 assert result.kg_entities_added >= 0 assert result.prompts_updated >= 0 assert result.rules_updated >= 0 # =========================================================================== # WHITE BOX TESTS # =========================================================================== def test_wb1_qdrant_upsert_uses_genesis_scars_collection(tmp_path): """ WB1: _update_scars() calls qdrant_client.upsert(collection_name="genesis_scars", ...). We verify the collection name is exactly "genesis_scars" (not "scars" or anything else). """ mock_qdrant = MagicMock() updater = _make_updater(tmp_path, qdrant_client=mock_qdrant) analysis = _make_analysis(scope="epistemic", num_bottlenecks=1) updater.apply_tier1(analysis, epoch_id="qdrant_test") # Confirm upsert was called assert mock_qdrant.upsert.called, "qdrant_client.upsert() was never called" # Confirm the collection_name argument call_kwargs = mock_qdrant.upsert.call_args # Supports both positional and keyword call styles if call_kwargs.kwargs: collection_name = call_kwargs.kwargs.get("collection_name") else: collection_name = call_kwargs.args[0] if call_kwargs.args else None assert collection_name == "genesis_scars", ( f"Expected collection_name='genesis_scars', got '{collection_name}'" ) def test_wb2_rules_file_append_only(tmp_path): """ WB2: _append_rules() APPENDS new content — existing content is preserved unchanged. Steps: 1. Write known sentinel text to rules_file. 2. Run apply_tier1() with epistemic analysis. 3. Assert sentinel text is still present at the START of the file. 4. Assert new content exists AFTER the sentinel text (not replacing it). """ rules_file = tmp_path / "GLOBAL_GENESIS_RULES.md" sentinel = "## ORIGINAL RULE\nThis content must not be modified by Tier 1 updates.\n" rules_file.write_text(sentinel, encoding="utf-8") updater = _make_updater(tmp_path) analysis = _make_analysis(scope="epistemic", num_bottlenecks=1) updater.apply_tier1(analysis, epoch_id="append_test") content = rules_file.read_text(encoding="utf-8") # Original content must be at the beginning assert content.startswith(sentinel), ( "Original rules file content was modified — Tier 1 must never overwrite existing rules" ) # New content must be appended AFTER the sentinel new_content = content[len(sentinel):] assert len(new_content) > 0, "No new content was appended to rules file" assert "AUTO-APPENDED RULE" in new_content, ( "Expected 'AUTO-APPENDED RULE' marker in appended content" ) def test_wb3_kg_entity_file_is_valid_jsonl(tmp_path): """ WB3: Every line in the epoch_{epoch_id}_learnings.jsonl file is parseable as JSON. Additional checks: - Each entry has a "type" field ("bottleneck" or "fix_proposal") - Each entry has an "epoch_id" field matching the provided epoch - Each entry has a "timestamp" field """ updater = _make_updater(tmp_path) analysis = ArchitectureAnalysis( bottlenecks=[ Bottleneck( description="memory leak in context window", frequency=4, affected_saga_ids=["s1", "s2"], scar_ids=["sc1"], ), Bottleneck( description="prompt truncation on long inputs", frequency=2, affected_saga_ids=["s3"], scar_ids=["sc2"], ), ], recommended_fixes=[ FixProposal( target_file="config/prompts/context.md", change_type="prompt_update", rationale="reduce context window usage", ), ], scope="epistemic", ) updater.apply_tier1(analysis, epoch_id="jsonl_valid_test") kg_file = tmp_path / "kg_entities" / "epoch_jsonl_valid_test_learnings.jsonl" assert kg_file.exists(), "KG entity file was not created" lines = [l for l in kg_file.read_text(encoding="utf-8").splitlines() if l.strip()] assert len(lines) > 0, "KG entity file is empty" for i, line in enumerate(lines): try: entry = json.loads(line) except json.JSONDecodeError as exc: pytest.fail(f"Line {i + 1} is not valid JSON: {exc}\nContent: {line!r}") assert "type" in entry, f"Line {i + 1} missing 'type' field: {entry}" assert entry["type"] in ("bottleneck", "fix_proposal"), ( f"Line {i + 1} has unexpected type: {entry['type']}" ) assert "epoch_id" in entry, f"Line {i + 1} missing 'epoch_id': {entry}" assert entry["epoch_id"] == "jsonl_valid_test" assert "timestamp" in entry, f"Line {i + 1} missing 'timestamp': {entry}" def test_wb4_prompt_updates_only_create_md_or_txt_files(tmp_path): """ WB4: _update_prompts() creates only .md or .txt files — never .py or any other extension. We provide fixes with various target extensions and assert only the safe ones produce output files in prompts_dir. """ updater = _make_updater(tmp_path) analysis = ArchitectureAnalysis( bottlenecks=[ Bottleneck( description="prompt unclear", frequency=2, affected_saga_ids=["s1"], scar_ids=["sc1"], ), ], recommended_fixes=[ FixProposal( target_file="config/prompts/safe_template.md", # ALLOWED change_type="prompt_update", rationale="safe md", ), FixProposal( target_file="config/prompts/safe_notes.txt", # ALLOWED change_type="prompt_update", rationale="safe txt", ), FixProposal( target_file="core/evolution/dangerous.py", # FORBIDDEN change_type="refactor", rationale="should not be created", ), FixProposal( target_file="data/output/result.json", # NOT allowed (not .md/.txt) change_type="data", rationale="not a prompt format", ), ], scope="epistemic", ) updater.apply_tier1(analysis, epoch_id="prompt_extension_test") prompts_dir = tmp_path / "prompts" assert prompts_dir.exists(), "Prompts directory was not created" created_files = list(prompts_dir.iterdir()) created_names = {f.name for f in created_files} created_suffixes = {f.suffix for f in created_files} # Safe files must exist assert "safe_template.md" in created_names, ( "Expected safe_template.md to be created in prompts_dir" ) assert "safe_notes.txt" in created_names, ( "Expected safe_notes.txt to be created in prompts_dir" ) # Forbidden extensions must NOT appear anywhere under tmp_path all_files = list(tmp_path.rglob("*")) forbidden = [f for f in all_files if f.is_file() and f.suffix == ".py"] assert forbidden == [], ( f"Found forbidden .py files: {[str(f) for f in forbidden]}" ) # Non-prompt formats should not appear in prompts_dir assert ".json" not in created_suffixes, ( "Prompts dir should not contain .json files" ) # Only .md and .txt in prompts_dir for f in created_files: if f.is_file(): assert f.suffix in (".md", ".txt"), ( f"Unexpected file extension in prompts_dir: {f.name}" ) # =========================================================================== # Additional contract tests # =========================================================================== def test_tier1result_is_dataclass(): """Tier1Result is a proper dataclass with required fields.""" import dataclasses assert dataclasses.is_dataclass(Tier1Result) field_names = {f.name for f in dataclasses.fields(Tier1Result)} assert "kg_entities_added" in field_names assert "scars_updated" in field_names assert "prompts_updated" in field_names assert "rules_updated" in field_names def test_no_qdrant_client_skips_scar_update_gracefully(tmp_path): """When qdrant_client is None, _update_scars returns 0 without raising.""" updater = _make_updater(tmp_path, qdrant_client=None) analysis = _make_analysis(scope="epistemic") result = updater.apply_tier1(analysis, epoch_id="no_qdrant_test") # Should not raise; scars_updated should be 0 assert result.scars_updated == 0, ( f"Expected scars_updated=0 when no Qdrant client, got {result.scars_updated}" ) def test_empty_analysis_produces_zero_counts(tmp_path): """Empty ArchitectureAnalysis → all Tier1Result fields are 0.""" updater = _make_updater(tmp_path) empty_analysis = ArchitectureAnalysis(bottlenecks=[], recommended_fixes=[], scope="epistemic") result = updater.apply_tier1(empty_analysis, epoch_id="empty_test") assert result.kg_entities_added == 0 assert result.scars_updated == 0 assert result.prompts_updated == 0 assert result.rules_updated == 0 def test_multiple_calls_append_multiple_audit_lines(tmp_path): """Each call to apply_tier1() appends a NEW line — not overwrites.""" updater = _make_updater(tmp_path) analysis = _make_analysis(scope="epistemic") updater.apply_tier1(analysis, epoch_id="run_1") updater.apply_tier1(analysis, epoch_id="run_2") audit_file = tmp_path / "tier1_updates.jsonl" lines = [l for l in audit_file.read_text(encoding="utf-8").splitlines() if l.strip()] assert len(lines) == 2, f"Expected 2 audit lines for 2 calls, got {len(lines)}" epochs = [json.loads(l)["epoch_id"] for l in lines] assert "run_1" in epochs assert "run_2" in epochs def test_ontological_scope_skips_rules_update(tmp_path): """Ontological scope analysis should NOT append rules (that's for PR scope).""" updater = _make_updater(tmp_path) analysis = _make_analysis(scope="ontological", num_bottlenecks=1) result = updater.apply_tier1(analysis, epoch_id="ontological_test") assert result.rules_updated == 0, ( "Tier 1 must not append rules for ontological (code-change) scope" ) def test_package_init_exports_tier1_classes(): """core.evolution.__init__.py exports Tier1AutonomousUpdater and Tier1Result.""" from core.evolution import ( # noqa: F401 Tier1AutonomousUpdater as T1U, Tier1Result as T1R, ) assert T1U is Tier1AutonomousUpdater assert T1R is Tier1Result def test_py_fix_proposals_excluded_from_kg_entities(tmp_path): """Fix proposals targeting .py files must not appear in KG entity files.""" updater = _make_updater(tmp_path) analysis = ArchitectureAnalysis( bottlenecks=[ Bottleneck( description="prompt issue", frequency=1, affected_saga_ids=["s1"], scar_ids=["sc1"], ), ], recommended_fixes=[ FixProposal( target_file="core/evolution/example.py", # must be excluded change_type="refactor", rationale="code fix", ), ], scope="epistemic", ) updater.apply_tier1(analysis, epoch_id="py_exclusion_test") kg_file = tmp_path / "kg_entities" / "epoch_py_exclusion_test_learnings.jsonl" if kg_file.exists(): lines = [l for l in kg_file.read_text(encoding="utf-8").splitlines() if l.strip()] for line in lines: entry = json.loads(line) if entry.get("type") == "fix_proposal": target = entry.get("target_file", "") assert not target.endswith(".py"), ( f"KG entity must not reference .py fix proposals: {entry}" ) # =========================================================================== # Standalone runner # =========================================================================== if __name__ == "__main__": import tempfile import traceback class _Tmp: def __init__(self): self._dir = Path(tempfile.mkdtemp()) def __truediv__(self, name: str) -> Path: return self._dir / name def rglob(self, pattern: str): return self._dir.rglob(pattern) tests = [ ("BB1: KG entity file created", test_bb1_kg_entity_file_created), ("BB2: No .py files modified", test_bb2_no_py_files_modified), ("BB3: Audit entry written", test_bb3_audit_entry_written), ("BB4: Returns correct counts", test_bb4_apply_tier1_returns_correct_counts), ("WB1: Qdrant uses genesis_scars collection", test_wb1_qdrant_upsert_uses_genesis_scars_collection), ("WB2: Rules file append-only", test_wb2_rules_file_append_only), ("WB3: KG JSONL is valid", test_wb3_kg_entity_file_is_valid_jsonl), ("WB4: Prompt only .md/.txt", test_wb4_prompt_updates_only_create_md_or_txt_files), ] passed = 0 for name, fn in tests: tmp = _Tmp() 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.08") else: sys.exit(1)