chat/tests/test_delete_impact.py

"""Tests for Task 95 — delete-impact computation service (Phase 4).

`compute_delete_impact` walks event_log forward from a target event_id and
produces an :class:`ImpactReport` describing what would be removed if
rewind-to-target were invoked. It is a pure preview — no database mutation.
T98's drawer surgical-delete UI uses this to render an "are you sure?"
modal before invoking the actual rewind path.
"""

from __future__ import annotations

from chat.db.connection import open_db
from chat.db.migrate import apply_migrations
from chat.eventlog.log import append_event
from chat.services.delete_impact import compute_delete_impact


def _seed_chat(conn) -> tuple[int, int]:
    """Append minimal bot + chat events; return their event ids."""
    bot_id = append_event(
        conn,
        kind="bot_authored",
        payload={
            "id": "bot_a",
            "name": "BotA",
            "persona": "...",
            "voice_samples": [],
            "traits": [],
            "backstory": "",
            "initial_relationship_to_you": "",
            "kickoff_prose": "",
        },
    )
    chat_id = append_event(
        conn,
        kind="chat_created",
        payload={
            "id": "chat_bot_a",
            "host_bot_id": "bot_a",
            "initial_time": "2026-04-26T20:00:00+00:00",
            "narrative_anchor": "Day 1",
            "weather": "",
        },
    )
    return bot_id, chat_id


def test_impact_for_simple_turn_lists_memory_and_edges(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        _seed_chat(conn)
        user_id = append_event(
            conn,
            kind="user_turn",
            payload={
                "chat_id": "chat_bot_a",
                "prose": "hey there friend",
                "segments": [],
            },
        )
        append_event(
            conn,
            kind="assistant_turn",
            payload={
                "chat_id": "chat_bot_a",
                "speaker_id": "bot_a",
                "text": "Hi! Good to see you.",
                "truncated": False,
                "user_turn_id": user_id,
            },
        )
        append_event(
            conn,
            kind="memory_written",
            payload={
                "owner_id": "bot_a",
                "chat_id": "chat_bot_a",
                "pov_summary": "You greeted me warmly today.",
                "witness_you": 1,
                "witness_host": 1,
                "witness_guest": 0,
                "source": "turn",
                "reliability": 1.0,
                "significance": 1,
                "pinned": 0,
                "auto_pinned": 0,
            },
        )
        append_event(
            conn,
            kind="edge_update",
            payload={
                "source_id": "you",
                "target_id": "bot_a",
                "affinity_delta": 0.1,
            },
        )

        report = compute_delete_impact(conn, target_event_id=user_id)

    assert report.target_event_id == user_id
    kinds = [item.kind for item in report.cascading]
    # Walk from user_turn forward — user_turn, assistant_turn,
    # memory_written, edge_update should all be in scope, in order.
    assert kinds == [
        "user_turn",
        "assistant_turn",
        "memory_written",
        "edge_update",
    ]
    # Memory description includes the pov_summary excerpt.
    mem_item = report.cascading[2]
    assert "memory:" in mem_item.description
    assert "greeted" in mem_item.description
    # Edge description includes both endpoints.
    edge_item = report.cascading[3]
    assert "you" in edge_item.description
    assert "bot_a" in edge_item.description
    assert edge_item.target_id == "you->bot_a"
    # Notes mentions total count.
    assert any("4 events" in n for n in report.notes)


def test_impact_for_scene_opening_turn_warns_about_subsequent(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        _seed_chat(conn)
        early_id = append_event(
            conn,
            kind="user_turn",
            payload={"chat_id": "chat_bot_a", "prose": "the start", "segments": []},
        )
        append_event(
            conn,
            kind="assistant_turn",
            payload={
                "chat_id": "chat_bot_a",
                "speaker_id": "bot_a",
                "text": "ok",
                "truncated": False,
                "user_turn_id": early_id,
            },
        )
        append_event(
            conn,
            kind="scene_closed",
            payload={
                "scene_id": 1,
                "closed_at": "2026-04-26T21:00:00+00:00",
                "significance": 2,
            },
        )

        report = compute_delete_impact(conn, target_event_id=early_id)

    # Scene-close warning fires when one is in scope.
    assert any("scene close" in n.lower() for n in report.notes)
    # The scene_closed event also appears as a cascading item.
    assert any(item.kind == "scene_closed" for item in report.cascading)


def test_impact_for_missing_event_returns_empty_with_note(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        _seed_chat(conn)
        report = compute_delete_impact(conn, target_event_id=999_999)

    assert report.cascading == []
    assert any("not found" in n for n in report.notes)


def test_impact_does_not_mutate_database(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        _seed_chat(conn)
        user_id = append_event(
            conn,
            kind="user_turn",
            payload={"chat_id": "chat_bot_a", "prose": "hi", "segments": []},
        )
        append_event(
            conn,
            kind="assistant_turn",
            payload={
                "chat_id": "chat_bot_a",
                "speaker_id": "bot_a",
                "text": "hello",
                "truncated": False,
                "user_turn_id": user_id,
            },
        )

        # Snapshot all event_log rows as a tuple-of-tuples.
        before = conn.execute(
            "SELECT id, branch_id, ts, kind, payload_json, superseded_by, "
            "hidden FROM event_log ORDER BY id"
        ).fetchall()

        compute_delete_impact(conn, target_event_id=user_id)

        after = conn.execute(
            "SELECT id, branch_id, ts, kind, payload_json, superseded_by, "
            "hidden FROM event_log ORDER BY id"
        ).fetchall()

    # Byte-identical: nothing inserted, deleted, or updated.
    assert before == after


def test_impact_includes_regenerated_from_warning(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        _seed_chat(conn)
        original_id = append_event(
            conn,
            kind="assistant_turn",
            payload={
                "chat_id": "chat_bot_a",
                "speaker_id": "bot_a",
                "text": "first try",
                "truncated": False,
                "user_turn_id": 0,
            },
        )
        regen_id = append_event(
            conn,
            kind="assistant_turn",
            payload={
                "chat_id": "chat_bot_a",
                "speaker_id": "bot_a",
                "text": "second try",
                "truncated": False,
                "user_turn_id": 0,
                "regenerated_from": original_id,
            },
        )

        report = compute_delete_impact(conn, target_event_id=regen_id)

    # The regenerated_from note carries the original event id so the user
    # knows the original turn isn't lost.
    assert any("regenerated from" in n for n in report.notes)
    assert any(str(original_id) in n for n in report.notes)