202 lines
8.2 KiB
Python
202 lines
8.2 KiB
Python
"""
|
|
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 "<recollection>" in block
|
|
assert "michigan:" in block
|
|
assert "[geography] usa" in block
|
|
assert "[type] state" in block
|
|
# No pending markers
|
|
assert "?" not in block
|