"""Per-turn memory writes (T21).

After ``assistant_turn`` lands the turn flow records a ``memory_written``
event for each present POV owner. Phase 1 single-bot turns only have the
host bot as a memory-store owner — ``you`` doesn't have a memory store in
v1 — so we write exactly one row per turn with witness flags
``[you=1, host=1, guest=0]``. The ``pov_summary`` is the assistant's raw
narrative text; T27 rewrites at scene close into per-POV summary form.
"""

from __future__ import annotations

import json
from pathlib import Path

import pytest
from fastapi.testclient import TestClient

from chat.app import app
from chat.db.connection import open_db
from chat.db.migrate import apply_migrations
from chat.eventlog.log import append_event
from chat.eventlog.projector import project
from chat.llm.mock import MockLLMClient
from chat.services.memory_write import record_turn_memory, record_turn_memory_for_present
import chat.state.entities  # noqa: F401  - register handlers
import chat.state.memory  # noqa: F401
import chat.state.world  # noqa: F401


def _seed_minimal(db_path: Path) -> None:
    """Author a bot and create a chat — bare minimum for memory writes."""
    with open_db(db_path) as conn:
        append_event(
            conn,
            kind="bot_authored",
            payload={
                "id": "bot_a",
                "name": "BotA",
                "persona": "...",
                "voice_samples": [],
                "traits": [],
                "backstory": "",
                "initial_relationship_to_you": "",
                "kickoff_prose": "",
            },
        )
        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": "",
            },
        )
        project(conn)


def test_record_turn_memory_writes_event_and_projects(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    _seed_minimal(db)
    with open_db(db) as conn:
        eid, mid = record_turn_memory(
            conn,
            chat_id="chat_bot_a",
            host_bot_id="bot_a",
            narrative_text="BotA looks up. 'You're back late.'",
            scene_id=None,
            chat_clock_at="2026-04-26T20:00:00+00:00",
        )
        assert eid > 0
        assert mid is not None and mid > 0

        rows = conn.execute(
            "SELECT id, owner_id, chat_id, pov_summary, "
            "witness_you, witness_host, witness_guest, "
            "source, reliability, significance, pinned, auto_pinned, "
            "chat_clock_at "
            "FROM memories WHERE owner_id = ?",
            ("bot_a",),
        ).fetchall()
        assert len(rows) == 1
        m = rows[0]
        assert m[1] == "bot_a"
        assert m[2] == "chat_bot_a"
        assert "looks up" in m[3]
        assert m[4] == 1  # witness_you
        assert m[5] == 1  # witness_host
        assert m[6] == 0  # witness_guest
        assert m[7] == "direct"
        assert m[8] == 1.0  # reliability default
        assert m[9] == 1  # significance default
        assert m[10] == 0  # pinned default
        assert m[11] == 0  # auto_pinned default
        assert m[12] == "2026-04-26T20:00:00+00:00"

        # And the underlying event_log row is the canonical source.
        cur = conn.execute(
            "SELECT COUNT(*) FROM event_log WHERE kind = 'memory_written'"
        )
        assert cur.fetchone()[0] == 1


def test_record_turn_memory_omits_optional_fields(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    _seed_minimal(db)
    with open_db(db) as conn:
        # Call without scene_id/chat_clock_at — should default to None.
        eid, mid = record_turn_memory(
            conn,
            chat_id="chat_bot_a",
            host_bot_id="bot_a",
            narrative_text="A simple memory.",
        )
        assert eid > 0
        assert mid is not None and mid > 0

        row = conn.execute(
            "SELECT scene_id, chat_clock_at, source, reliability, "
            "significance, pinned, auto_pinned "
            "FROM memories WHERE owner_id = 'bot_a'"
        ).fetchone()
        assert row is not None
        scene_id, chat_clock_at, source, reliability, significance, pinned, auto_pinned = row
        assert scene_id is None
        assert chat_clock_at is None
        assert source == "direct"
        assert reliability == 1.0
        assert significance == 1
        assert pinned == 0
        assert auto_pinned == 0


# ---------------------------------------------------------------------------
# Integration: POST /chats/<id>/turns produces a memory_written event.
# ---------------------------------------------------------------------------


@pytest.fixture
def client(tmp_path, monkeypatch):
    cfg = tmp_path / "config.toml"
    cfg.write_text('featherless_api_key = "test"\n')
    monkeypatch.setenv("CHAT_CONFIG_PATH", str(cfg))
    db = tmp_path / "test.db"
    monkeypatch.setenv("CHAT_DB_PATH", str(db))

    canned_parse = json.dumps(
        {"segments": [{"kind": "dialogue", "text": "hello"}]}
    )
    canned_response = "BotA nods. 'Hi there.'"
    canned_state_update = json.dumps(
        {"affinity_delta": 0, "trust_delta": 0, "knowledge_facts": []}
    )
    # T26 scene-close detection runs after the state-update pass. ``_seed_full``
    # below doesn't open a scene so the classifier call is short-circuited in
    # turns.py — but the canned slot stays in place to document the order.
    canned_scene_close = json.dumps(
        {"should_close": False, "reason": "no signal"}
    )

    from chat.web.kickoff import get_llm_client

    mock = MockLLMClient(
        canned=[
            canned_parse,
            canned_response,
            canned_state_update,
            canned_state_update,
            canned_scene_close,
        ]
    )
    app.dependency_overrides[get_llm_client] = lambda: mock

    with TestClient(app) as c:
        # Disable the lifespan-managed background worker — it would try
        # to call Featherless with the test API key. The unit tests in
        # test_significance.py exercise the worker directly with a mock
        # factory; here we only care about the synchronous turn flow.
        app.state.background_worker.enabled = False
        c.mock_llm = mock  # type: ignore[attr-defined]
        yield c

    app.dependency_overrides.clear()


def _seed_full(db_path: Path) -> None:
    """Seed enough state for a full turn flow (matches test_turn_flow.py)."""
    with open_db(db_path) as conn:
        append_event(
            conn,
            kind="bot_authored",
            payload={
                "id": "bot_a",
                "name": "BotA",
                "persona": "thoughtful, observant",
                "voice_samples": [],
                "traits": [],
                "backstory": "",
                "initial_relationship_to_you": "",
                "kickoff_prose": "...",
            },
        )
        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": "",
            },
        )
        append_event(
            conn,
            kind="edge_update",
            payload={
                "source_id": "bot_a",
                "target_id": "you",
                "chat_id": "chat_bot_a",
                "knowledge_facts": ["coworker"],
            },
        )
        append_event(
            conn,
            kind="activity_change",
            payload={
                "entity_id": "you",
                "posture": "sitting",
                "action": {
                    "verb": "talking",
                    "interruptible": True,
                    "required_attention": "low",
                    "expected_duration": "ongoing",
                },
                "attention": "",
                "holding": [],
                "status": {},
            },
        )
        append_event(
            conn,
            kind="activity_change",
            payload={
                "entity_id": "bot_a",
                "posture": "sitting",
                "action": {
                    "verb": "listening",
                    "interruptible": True,
                    "required_attention": "low",
                    "expected_duration": "ongoing",
                },
                "attention": "",
                "holding": [],
                "status": {},
            },
        )
        project(conn)


def test_post_turn_writes_memory_for_host_bot(client, tmp_path):
    """After a POST turn, exactly one memory_written event is appended and
    a corresponding memory row is projected for the host bot's POV."""
    _seed_full(tmp_path / "test.db")
    response = client.post(
        "/chats/chat_bot_a/turns", data={"prose": "hello"}
    )
    assert response.status_code == 204

    with open_db(tmp_path / "test.db") as conn:
        cur = conn.execute(
            "SELECT COUNT(*) FROM event_log WHERE kind = 'memory_written'"
        )
        assert cur.fetchone()[0] == 1

        cur = conn.execute(
            "SELECT owner_id, chat_id, pov_summary, "
            "witness_you, witness_host, witness_guest, source, significance "
            "FROM memories"
        )
        rows = cur.fetchall()
        assert len(rows) == 1
        owner_id, chat_id, pov_summary, w_you, w_host, w_guest, source, sig = rows[0]
        assert owner_id == "bot_a"
        assert chat_id == "chat_bot_a"
        # pov_summary is the assistant's narrative text (Phase 1 simplification).
        assert pov_summary == "BotA nods. 'Hi there.'"
        assert w_you == 1
        assert w_host == 1
        assert w_guest == 0
        assert source == "direct"
        assert sig == 1


# ---------------------------------------------------------------------------
# T41: record_turn_memory_for_present — multi-witness helper.
# ---------------------------------------------------------------------------


def _seed_two_bots(db_path: Path) -> None:
    """Author host + guest bots and create a two-bot chat."""
    with open_db(db_path) as conn:
        for bot_id, name in (("bot_a", "BotA"), ("bot_b", "BotB")):
            append_event(
                conn,
                kind="bot_authored",
                payload={
                    "id": bot_id,
                    "name": name,
                    "persona": "...",
                    "voice_samples": [],
                    "traits": [],
                    "backstory": "",
                    "initial_relationship_to_you": "",
                    "kickoff_prose": "",
                },
            )
        append_event(
            conn,
            kind="chat_created",
            payload={
                "id": "chat_ab",
                "host_bot_id": "bot_a",
                "guest_bot_id": "bot_b",
                "initial_time": "2026-04-26T20:00:00+00:00",
                "narrative_anchor": "Day 1",
                "weather": "",
            },
        )
        project(conn)


def test_record_for_present_no_guest_writes_single_memory_with_witness_1_1_0(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    _seed_minimal(db)
    with open_db(db) as conn:
        result = record_turn_memory_for_present(
            conn,
            chat_id="chat_bot_a",
            host_bot_id="bot_a",
            guest_bot_id=None,
            narrative_text="BotA glances out the window.",
            scene_id=None,
            chat_clock_at="2026-04-26T20:00:00+00:00",
        )

        # Returned dict has only the host key.
        assert set(result.keys()) == {"bot_a"}
        eid_h, mid_h = result["bot_a"]
        assert eid_h > 0
        assert mid_h is not None and mid_h > 0

        rows = conn.execute(
            "SELECT owner_id, witness_you, witness_host, witness_guest "
            "FROM memories"
        ).fetchall()
        assert len(rows) == 1
        owner_id, w_you, w_host, w_guest = rows[0]
        assert owner_id == "bot_a"
        assert w_you == 1
        assert w_host == 1
        assert w_guest == 0

        # Exactly one memory_written event was appended.
        cur = conn.execute(
            "SELECT COUNT(*) FROM event_log WHERE kind = 'memory_written'"
        )
        assert cur.fetchone()[0] == 1


def test_record_for_present_with_guest_writes_two_memories_with_witness_1_1_1(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    _seed_two_bots(db)
    with open_db(db) as conn:
        result = record_turn_memory_for_present(
            conn,
            chat_id="chat_ab",
            host_bot_id="bot_a",
            guest_bot_id="bot_b",
            narrative_text="BotA and BotB share a glance.",
            scene_id=None,
            chat_clock_at="2026-04-26T20:00:00+00:00",
        )

        # Returned dict has both keys.
        assert set(result.keys()) == {"bot_a", "bot_b"}
        eid_h, mid_h = result["bot_a"]
        eid_g, mid_g = result["bot_b"]
        assert eid_h > 0 and eid_g > 0
        assert mid_h is not None and mid_h > 0
        assert mid_g is not None and mid_g > 0
        # Distinct event ids and memory ids.
        assert eid_h != eid_g
        assert mid_h != mid_g

        rows = conn.execute(
            "SELECT owner_id, witness_you, witness_host, witness_guest "
            "FROM memories ORDER BY owner_id"
        ).fetchall()
        assert len(rows) == 2
        owners = {r[0] for r in rows}
        assert owners == {"bot_a", "bot_b"}
        # All rows should have witness mask [1, 1, 1].
        for _owner, w_you, w_host, w_guest in rows:
            assert w_you == 1
            assert w_host == 1
            assert w_guest == 1

        # Two memory_written events were appended.
        cur = conn.execute(
            "SELECT COUNT(*) FROM event_log WHERE kind = 'memory_written'"
        )
        assert cur.fetchone()[0] == 2


def test_record_for_present_dict_keys_match(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    _seed_two_bots(db)
    with open_db(db) as conn:
        # No guest: keys == {host_bot_id}.
        result_no_guest = record_turn_memory_for_present(
            conn,
            chat_id="chat_ab",
            host_bot_id="bot_a",
            guest_bot_id=None,
            narrative_text="Just BotA's POV.",
        )
        assert set(result_no_guest.keys()) == {"bot_a"}

        # With guest: keys == {host_bot_id, guest_bot_id}.
        result_with_guest = record_turn_memory_for_present(
            conn,
            chat_id="chat_ab",
            host_bot_id="bot_a",
            guest_bot_id="bot_b",
            narrative_text="Both bots witness this.",
        )
        assert set(result_with_guest.keys()) == {"bot_a", "bot_b"}


# ---------------------------------------------------------------------------
# T84: unified record_turn_memory_for_present API with you_present kwarg.
# ---------------------------------------------------------------------------


def test_record_turn_memory_you_present_false_writes_meanwhile_witness_mask(tmp_path):
    """When ``you_present=False`` the witness mask should be
    ``[you=0, host=1, guest=1]`` for both bots — the meanwhile shape."""
    db = tmp_path / "t.db"
    apply_migrations(db)
    _seed_two_bots(db)
    with open_db(db) as conn:
        result = record_turn_memory_for_present(
            conn,
            chat_id="chat_ab",
            host_bot_id="bot_a",
            guest_bot_id="bot_b",
            narrative_text="BotA and BotB confer privately.",
            scene_id=None,
            chat_clock_at="2026-04-26T20:00:00+00:00",
            you_present=False,
        )

        assert set(result.keys()) == {"bot_a", "bot_b"}

        rows = conn.execute(
            "SELECT owner_id, witness_you, witness_host, witness_guest "
            "FROM memories ORDER BY owner_id"
        ).fetchall()
        assert len(rows) == 2
        for _owner, w_you, w_host, w_guest in rows:
            assert w_you == 0
            assert w_host == 1
            assert w_guest == 1

        # Two memory_written events were appended.
        cur = conn.execute(
            "SELECT COUNT(*) FROM event_log WHERE kind = 'memory_written'"
        )
        assert cur.fetchone()[0] == 2


def test_record_turn_memory_you_present_true_default_writes_normal_witness_mask(tmp_path):
    """Default ``you_present=True`` preserves Phase 2 behaviour:
    ``witness_you=1`` for the host POV row."""
    db = tmp_path / "t.db"
    apply_migrations(db)
    _seed_minimal(db)
    with open_db(db) as conn:
        # No explicit you_present arg — should default to True.
        result = record_turn_memory_for_present(
            conn,
            chat_id="chat_bot_a",
            host_bot_id="bot_a",
            guest_bot_id=None,
            narrative_text="BotA hums to herself.",
        )
        assert set(result.keys()) == {"bot_a"}

        row = conn.execute(
            "SELECT witness_you, witness_host, witness_guest "
            "FROM memories WHERE owner_id = 'bot_a'"
        ).fetchone()
        assert row is not None
        w_you, w_host, w_guest = row
        assert w_you == 1
        assert w_host == 1
        assert w_guest == 0


def test_record_turn_memory_you_present_false_requires_guest(tmp_path):
    """Calling with ``you_present=False`` and no ``guest_bot_id`` is a
    programming error — meanwhile scenes always have both bots."""
    db = tmp_path / "t.db"
    apply_migrations(db)
    _seed_minimal(db)
    with open_db(db) as conn:
        with pytest.raises(ValueError, match="you_present=False requires guest_bot_id"):
            record_turn_memory_for_present(
                conn,
                chat_id="chat_bot_a",
                host_bot_id="bot_a",
                guest_bot_id=None,
                narrative_text="invalid",
                you_present=False,
            )