chat/tests/test_edges.py

from chat.db.migrate import apply_migrations
from chat.db.connection import open_db
from chat.eventlog.log import append_event
from chat.eventlog.projector import project
from chat.state.edges import get_edge, list_edges_for
import chat.state.entities  # registers bot/you handlers
import chat.state.edges  # registers edge_update handler


def _seed_entities(conn) -> None:
    append_event(conn, kind="bot_authored", payload={
        "id": "bot_a", "name": "BotA", "persona": "p",
        "voice_samples": [], "traits": [],
        "backstory": "", "initial_relationship_to_you": "",
        "kickoff_prose": "",
    })
    append_event(conn, kind="you_authored", payload={
        "name": "Me", "pronouns": "", "persona": "",
    })


def test_edge_update_upsert_applies_first_delta(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        _seed_entities(conn)
        append_event(conn, kind="edge_update", payload={
            "source_id": "bot_a", "target_id": "you",
            "affinity_delta": 5,
        })
        project(conn)
        edge = get_edge(conn, "bot_a", "you")
        assert edge is not None
        assert edge["affinity"] == 55
        assert edge["trust"] == 50
        assert edge["knowledge"] == []
        assert edge["summary"] == ""


def test_edge_update_multiple_deltas_accumulate(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        _seed_entities(conn)
        append_event(conn, kind="edge_update", payload={
            "source_id": "bot_a", "target_id": "you",
            "affinity_delta": 5,
        })
        append_event(conn, kind="edge_update", payload={
            "source_id": "bot_a", "target_id": "you",
            "affinity_delta": -3,
            "trust_delta": 2,
            "knowledge_facts": ["she has a sister"],
        })
        project(conn)
        edge = get_edge(conn, "bot_a", "you")
        assert edge is not None
        assert edge["affinity"] == 52
        assert edge["trust"] == 52
        assert edge["knowledge"] == ["she has a sister"]


def test_edge_update_clamps_affinity_at_max(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        _seed_entities(conn)
        append_event(conn, kind="edge_update", payload={
            "source_id": "bot_a", "target_id": "you",
            "affinity_delta": 5,
        })
        append_event(conn, kind="edge_update", payload={
            "source_id": "bot_a", "target_id": "you",
            "affinity_delta": 100,
        })
        project(conn)
        edge = get_edge(conn, "bot_a", "you")
        assert edge is not None
        assert edge["affinity"] == 100


def test_edge_update_clamps_trust_at_min(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        _seed_entities(conn)
        append_event(conn, kind="edge_update", payload={
            "source_id": "bot_a", "target_id": "you",
            "trust_delta": -200,
        })
        project(conn)
        edge = get_edge(conn, "bot_a", "you")
        assert edge is not None
        assert edge["trust"] == 0


def test_edges_are_directed_and_asymmetric(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        _seed_entities(conn)
        append_event(conn, kind="edge_update", payload={
            "source_id": "bot_a", "target_id": "you",
            "affinity_delta": 5,
        })
        append_event(conn, kind="edge_update", payload={
            "source_id": "you", "target_id": "bot_a",
            "affinity_delta": 10,
        })
        project(conn)
        forward = get_edge(conn, "bot_a", "you")
        reverse = get_edge(conn, "you", "bot_a")
        assert forward is not None and reverse is not None
        assert forward["affinity"] == 55
        assert reverse["affinity"] == 60
        # Independent rows
        assert forward["affinity"] != reverse["affinity"]


def test_edge_update_bumps_last_interaction(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        _seed_entities(conn)
        append_event(conn, kind="edge_update", payload={
            "source_id": "bot_a", "target_id": "you",
            "affinity_delta": 1,
            "last_interaction_at": "2026-04-26T10:00:00",
            "last_interaction_chat_id": "chat_bot_a",
        })
        project(conn)
        edge = get_edge(conn, "bot_a", "you")
        assert edge is not None
        assert edge["last_interaction_at"] == "2026-04-26T10:00:00"
        assert edge["last_interaction_chat_id"] == "chat_bot_a"


def test_list_edges_for_returns_outgoing_only(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        _seed_entities(conn)
        append_event(conn, kind="bot_authored", payload={
            "id": "bot_b", "name": "BotB", "persona": "p",
            "voice_samples": [], "traits": [],
            "backstory": "", "initial_relationship_to_you": "",
            "kickoff_prose": "",
        })
        append_event(conn, kind="edge_update", payload={
            "source_id": "bot_a", "target_id": "you", "affinity_delta": 1,
        })
        append_event(conn, kind="edge_update", payload={
            "source_id": "bot_a", "target_id": "bot_b", "affinity_delta": 2,
        })
        append_event(conn, kind="edge_update", payload={
            "source_id": "bot_b", "target_id": "bot_a", "affinity_delta": 3,
        })
        project(conn)
        outgoing = list_edges_for(conn, "bot_a")
        targets = [e["target_id"] for e in outgoing]
        assert targets == sorted(targets)
        assert set(targets) == {"you", "bot_b"}