""" Scenario A — full integration walk-through (in-memory only, no DB required). Demonstrates the degenerate parent_id=dim_id pattern and the misclassification collision pipeline from initial seed through recollection rendering. Story: 1. World model receives coarse early knowledge: "michigan is in usa" → stored as michigan ISPART usa in dim:usa (parent_id = dim_id) 2. Agent sends prompt: "michigan is a state of USA" → cue scanner extracts michigan ISA state in dim:usa 3. Collision detected: existing ISPART vs incoming ISA in dim:usa → misclassification 4. Recollection rendered with [usa?] marker while conflict is pending 5. After resolution (simulated): fact moves to correct dimension → michigan ISA state in dim:type → michigan ISPART usa in dim:geography """ from __future__ import annotations import pytest from unittest.mock import AsyncMock, MagicMock, patch from festinger import cache from festinger.cache import SoasRow, UrdEdge from festinger.cue_scanner import scan_cues from festinger.recollection import ( build_recollection_block, inject_recollection, render_hit, query_edges ) from festinger.urd_writer import InsertRequest, insert_urd_edge, CollisionInfo from tests.helpers import reset_cache, add_soas, add_urd def make_mock_pool(): conn = AsyncMock() conn.execute = AsyncMock(return_value="INSERT 0 1") conn.fetchrow = AsyncMock(return_value=None) conn.__aenter__ = AsyncMock(return_value=conn) conn.__aexit__ = AsyncMock(return_value=False) pool = MagicMock() pool.acquire = MagicMock(return_value=conn) return pool, conn class TestScenarioAIntegration: def setup_method(self): reset_cache() # SOAS vocabulary self.michigan = add_soas(101, "michigan", saliency=1.5, novelty=1.0) self.usa = add_soas(102, "usa", saliency=0.8, novelty=0.8) self.state = add_soas(103, "state", saliency=0.6, novelty=0.5) self.type_dim = add_soas(1, "type", saliency=0.0) self.geo_dim = add_soas(5, "geography", saliency=0.0) # Degenerate seed edge: michigan ISPART usa, dim=usa (parent_id = dim_id = 102) self.seed_edge = add_urd( concept_id=101, parent_id=102, # usa dim_id=102, # usa — same as parent_id is_isa=False, confidence=0.85, source="test", ) # ------------------------------------------------------------------ # Step 1: Verify the degenerate seed state # ------------------------------------------------------------------ def test_seed_edge_has_parent_id_equals_dim_id(self): edge = cache.urd_by_concept_dim.get((101, 102)) assert edge is not None, "seed edge must be in cache" assert edge.parent_id == edge.dim_id, ( f"degenerate edge requires parent_id == dim_id, " f"got parent_id={edge.parent_id}, dim_id={edge.dim_id}" ) assert edge.is_isa is False def test_seed_recollection_renders_correctly(self): edges = query_edges(101, confidence_floor=0.5, recency_days=90) assert len(edges) == 1 line = render_hit("michigan", edges, concept_id=101) # dim token = "usa", parent token = "usa" assert "[usa] usa" in line # ------------------------------------------------------------------ # Step 2: Cue scanner extracts the incoming ISA triple # ------------------------------------------------------------------ def test_cue_scanner_extracts_michigan_isa_state(self): prompt = "michigan is a state of USA" triples = scan_cues(prompt) match = next( (t for t in triples if t.subject == "michigan" and t.parent == "state"), None, ) assert match is not None, "cue scanner must extract michigan ISA state" assert match.is_isa is True assert match.dimension == "usa" # from "of USA" modifier # ------------------------------------------------------------------ # Step 3: Incoming ISA in dim:usa collides with existing ISPART in dim:usa # ------------------------------------------------------------------ @pytest.mark.asyncio async def test_collision_classified_as_misclassification(self): pool, _ = make_mock_pool() captured: list[CollisionInfo] = [] async def fake_queue(pool, col, priority): captured.append(col) cache.pending_conflicts.add(col.concept_id) with patch("festinger.urd_writer._queue_collision", side_effect=fake_queue): req = InsertRequest( concept_id=101, # michigan parent_id=103, # state dim_id=102, # usa — same dim as existing ISPART is_isa=True, confidence=0.85, source="gutask", ) collision = await insert_urd_edge(pool, req) assert collision is not None assert collision.collision_type == "misclassification", ( f"expected misclassification, got {collision.collision_type!r}. " f"existing is_isa={self.seed_edge.is_isa}, incoming is_isa=True" ) # ------------------------------------------------------------------ # Step 4: Recollection renders the [usa?] pending-conflict marker # ------------------------------------------------------------------ @pytest.mark.asyncio async def test_recollection_shows_pending_marker_after_collision(self): pool, _ = make_mock_pool() async def fake_queue(pool, col, priority): cache.pending_conflicts.add(col.concept_id) with patch("festinger.urd_writer._queue_collision", side_effect=fake_queue): req = InsertRequest( concept_id=101, parent_id=103, dim_id=102, is_isa=True, confidence=0.85, source="gutask", ) await insert_urd_edge(pool, req) assert 101 in cache.pending_conflicts block = build_recollection_block([101], confidence_floor=0.5, recency_days=90) assert block is not None assert "[usa?]" in block, ( f"recollection must show [usa?] pending marker. Got:\n{block}" ) # ------------------------------------------------------------------ # Step 5: Simulate resolution — move facts to correct dimensions # After nightly job: # michigan ISA state in dim:type (the ISA fact) # michigan ISPART usa in dim:geography (the ISPART fact, moved off degenerate dim) # ------------------------------------------------------------------ def test_recollection_clean_after_simulated_resolution(self): # Remove the degenerate edge del cache.urd_by_concept_dim[(101, 102)] cache.urd_by_concept[101] = [] # Insert two correctly-dimensioned edges geo_edge = add_urd( concept_id=101, parent_id=102, dim_id=5, # geography dim is_isa=False, confidence=0.9, source="festinger", ) type_edge = add_urd( concept_id=101, parent_id=103, dim_id=1, # type dim is_isa=True, confidence=0.9, source="festinger", ) cache.pending_conflicts.discard(101) edges = query_edges(101, confidence_floor=0.5, recency_days=90) assert len(edges) == 2 line = render_hit("michigan", edges, concept_id=101) assert "[geography] usa" in line assert "[type] state" in line assert "?" not in line, "no pending marker after resolution" def test_full_recollection_block_after_resolution(self): # Simulate post-resolution state del cache.urd_by_concept_dim[(101, 102)] cache.urd_by_concept[101] = [] add_urd(concept_id=101, parent_id=102, dim_id=5, is_isa=False, confidence=0.9, source="festinger") add_urd(concept_id=101, parent_id=103, dim_id=1, is_isa=True, confidence=0.9, source="festinger") cache.pending_conflicts.discard(101) block = build_recollection_block([101], confidence_floor=0.5, recency_days=90) assert block is not None assert "" in block assert "michigan:" in block assert "[geography] usa" in block assert "[type] state" in block # No pending markers assert "?" not in block