"""Real-Postgres tests for the drift surface (milestone 399 step 5, #3895).

WHY THIS EXISTS

A preference is the one record kind the AGENT rewrites, mid-work, without
asking. That is what keeps it current and it is also the risk the milestone
named up front: an agent misreads one session, rewrites a preference, and
follows the rewritten version forever while the operator never sees the
moment it changed — a confident wrong answer wearing the operator's own
authority.

`rule_versions` already recorded every rewrite. What it did not do is arrive.
`recent_preference_drift` is the read that makes drift PUSHED rather than
pulled, and `restore_rule_version` is the veto.

WHY NOT MOCKS

Every claim here is about which rows come back and in what order — a grouped
subquery picking the newest version per rule, an ownership clause spanning two
paths, and a write that has to leave the history longer than it found it. A
stand-in session returns whatever the test handed it, so it could confirm none
of those. `test_the_restore_is_recorded_as_an_edit` is the one that matters
most: it is the whole basis for allowing a one-click revert here when
milestone 323 refused one for rules, and if it were ever to pass while the
restore silently overwrote history, the carve-out would be indefensible.
"""
import pytest
import pytest_asyncio
from sqlalchemy import select

from scribe.models import async_session
from scribe.models.note import Note
from scribe.models.rulebook import Rulebook
from scribe.models.rule_version import RuleVersion
from scribe.services import rulebooks as rulebooks_svc
from tests.helpers import ensure_user

pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]

OWNER_USERNAME = "drift_owner"
STRANGER_USERNAME = "drift_stranger"

# The moment a preference names, in the words a session actually produces.
# Required on a preference, so every fixture below carries one.
TRIGGER = "the operator pasted a stack trace and said it is still broken"


async def _purge(uid: int) -> None:
    """Clear this user's rulebooks. At SETUP, never teardown.

    `update_rule` fires a detached embedding refresh that opens its own
    connection and UPDATEs the rule row. A teardown delete would race it into
    a deadlock — the same reasoning test_integration_rule_versions records.
    """
    async with async_session() as s:
        for book in (await s.execute(
            select(Rulebook).where(Rulebook.owner_user_id == uid)
        )).scalars().all():
            await s.delete(book)
        await s.commit()


@pytest_asyncio.fixture
async def world():
    async with async_session() as s:
        owner = await ensure_user(s, OWNER_USERNAME)
        stranger = await ensure_user(s, STRANGER_USERNAME)
        await s.commit()
        uid, sid = owner.id, stranger.id

        # The record a preference points at as what taught it. A plain note:
        # `arose_from_id` is a FK to notes, and the drift row names it so the
        # operator reads a title rather than a bare number.
        taught = Note(user_id=uid, title="Paced the debugging one step at a time")
        s.add(taught)
        await s.commit()
        await s.refresh(taught)
        taught_id = taught.id

    await _purge(uid)
    await _purge(sid)

    book = await rulebooks_svc.create_rulebook(uid, "Drift fixtures")
    topic = await rulebooks_svc.create_topic(book.id, uid, "collaboration")

    pref = await rulebooks_svc.create_rule(
        topic.id, uid, "Pace hard debugging",
        "Change one thing, then look.",
        when_to_apply=TRIGGER, kind="preference", arose_from_id=taught_id,
    )
    rule = await rulebooks_svc.create_rule(
        topic.id, uid, "dev is home",
        "Ordinary work lands on dev.",
        when_to_apply="about to commit", kind="rule",
    )
    return {
        "uid": uid, "stranger_id": sid, "topic_id": topic.id,
        "pref_id": pref.id, "rule_id": rule.id, "taught_id": taught_id,
    }


async def test_a_preference_nobody_rewrote_is_not_drift(world):
    """The empty state is a REAL state, not a not-yet-loaded one.

    A preference that has never been touched has nothing for the operator to
    review, and listing it would bury the rows that do under the ones that
    don't — which is how a supervision surface stops being read.
    """
    rows = await rulebooks_svc.recent_preference_drift(world["uid"])
    assert rows == []


async def test_a_rewritten_preference_carries_both_wordings_and_what_taught_it(world):
    await rulebooks_svc.update_rule(
        world["pref_id"], world["uid"],
        statement="Change one thing, then look. Say what you expect first.",
    )

    [row] = await rulebooks_svc.recent_preference_drift(world["uid"])
    assert row["rule"]["id"] == world["pref_id"]
    # Both texts ride along, so the list can show the diff without a
    # follow-up call per row.
    assert row["previous"]["statement"] == "Change one thing, then look."
    assert row["current"]["statement"].endswith("Say what you expect first.")
    # The provenance is NAMED, not numbered: a bare id reads as complete to
    # the writer and as homework to the reader.
    assert row["taught_by"] == {
        "id": world["taught_id"],
        "title": "Paced the debugging one step at a time",
    }


async def test_a_rewritten_rule_is_not_listed(world):
    """Not a filter that could be relaxed — the question is what changed
    WITHOUT the operator, and a rule changes when they change it. Including
    rules would bury the few unreviewed rows under every edit they made."""
    await rulebooks_svc.update_rule(
        world["rule_id"], world["uid"], statement="Ordinary work lands on dev, always.",
    )
    assert await rulebooks_svc.recent_preference_drift(world["uid"]) == []


async def test_one_row_per_preference_carrying_its_LATEST_rewrite(world):
    """A preference rewritten three times this week is one answer to "what
    changed lately", not three. The full history stays in its editor."""
    for text in ("second wording.", "third wording.", "fourth wording."):
        await rulebooks_svc.update_rule(
            world["pref_id"], world["uid"], statement=text,
        )

    rows = await rulebooks_svc.recent_preference_drift(world["uid"])
    assert len(rows) == 1
    # The newest version holds what the LAST edit replaced.
    assert rows[0]["previous"]["statement"] == "third wording."
    assert rows[0]["current"]["statement"] == "fourth wording."


async def test_most_recently_changed_first(world):
    """The ORDER is the answer: the top of this list is what changed last."""
    second = await rulebooks_svc.create_rule(
        world["topic_id"], world["uid"], "Name the record",
        "Say the title beside the id.",
        when_to_apply="about to cite a record by number", kind="preference",
    )
    await rulebooks_svc.update_rule(
        world["pref_id"], world["uid"], statement="edited first.",
    )
    await rulebooks_svc.update_rule(
        second.id, world["uid"], statement="edited second.",
    )

    rows = await rulebooks_svc.recent_preference_drift(world["uid"])
    assert [r["rule"]["id"] for r in rows] == [second.id, world["pref_id"]]


async def test_another_users_preferences_are_not_listed(world):
    """The listing clause has to agree with the per-row fetch. A drift pane
    that reached across owners would leak the wording of someone else's
    preference — and the wording is the whole payload."""
    await rulebooks_svc.update_rule(
        world["pref_id"], world["uid"], statement="mine, rewritten.",
    )
    assert await rulebooks_svc.recent_preference_drift(world["stranger_id"]) == []


async def test_the_limit_bounds_the_list_and_keeps_the_newest(world):
    """A cap that dropped the wrong end would hide the change just made.

    Bounded because the payload is two full statements per row — so the list
    has to stay short — and the end it keeps has to be the recent one, which
    is the only end this surface is about.
    """
    second = await rulebooks_svc.create_rule(
        world["topic_id"], world["uid"], "Name the record",
        "Say the title beside the id.",
        when_to_apply="about to cite a record by number", kind="preference",
    )
    await rulebooks_svc.update_rule(
        world["pref_id"], world["uid"], statement="edited first.",
    )
    await rulebooks_svc.update_rule(
        second.id, world["uid"], statement="edited second.",
    )

    rows = await rulebooks_svc.recent_preference_drift(world["uid"], limit=1)
    assert [r["rule"]["id"] for r in rows] == [second.id]
    # A caller asking for everything gets a page, not the table — and the
    # clamp is silent rather than an error, because "too many" is a request
    # to serve, not a mistake to report.
    assert len(await rulebooks_svc.recent_preference_drift(
        world["uid"], limit=10_000,
    )) == 2


async def test_restoring_puts_the_old_wording_back(world):
    await rulebooks_svc.update_rule(
        world["pref_id"], world["uid"], statement="the agent's rewrite.",
    )
    [row] = await rulebooks_svc.recent_preference_drift(world["uid"])

    restored = await rulebooks_svc.restore_rule_version(
        world["pref_id"], row["previous"]["id"], world["uid"],
    )
    assert restored is not None
    assert restored.statement == "Change one thing, then look."


async def test_the_restore_is_recorded_as_an_edit(world):
    """THE ONE THAT MATTERS MOST — it is why this carve-out is defensible.

    Milestone 323 refused a one-click restore for rules because "a silent
    revert would erase the only record of why the rewrite happened". The
    preference carve-out survives that objection only while nothing is
    erased: the restore goes through `update_rule`, so it takes its own
    snapshot and the history GAINS an entry.

    A restore implemented as a direct write would satisfy the test above and
    fail this one, which is the point of having both.
    """
    await rulebooks_svc.update_rule(
        world["pref_id"], world["uid"], statement="the agent's rewrite.",
    )
    async with async_session() as s:
        before = len((await s.execute(
            select(RuleVersion).where(RuleVersion.rule_id == world["pref_id"])
        )).scalars().all())

    [row] = await rulebooks_svc.recent_preference_drift(world["uid"])
    await rulebooks_svc.restore_rule_version(
        world["pref_id"], row["previous"]["id"], world["uid"],
    )

    async with async_session() as s:
        versions = list((await s.execute(
            select(RuleVersion).where(RuleVersion.rule_id == world["pref_id"])
            .order_by(RuleVersion.id)
        )).scalars().all())
    assert len(versions) == before + 1
    # The rewrite is still there, and the newest entry is what the restore
    # replaced — the rewrite itself. Nothing was overwritten.
    assert versions[-1].statement == "the agent's rewrite."


async def test_a_rule_refuses_the_one_click_restore(world):
    """Milestone 323 stands for rules, and the kind check is in the service
    rather than only in the route — so a caller reaching this directly cannot
    forget it."""
    await rulebooks_svc.update_rule(
        world["rule_id"], world["uid"], statement="rewritten by hand.",
    )
    async with async_session() as s:
        version = (await s.execute(
            select(RuleVersion).where(RuleVersion.rule_id == world["rule_id"])
        )).scalars().first()

    with pytest.raises(ValueError, match="preference"):
        await rulebooks_svc.restore_rule_version(
            world["rule_id"], version.id, world["uid"],
        )


async def test_a_stranger_cannot_restore_a_preference_they_cannot_read(world):
    """None, not a raise: "not yours" and "not a preference" are different
    answers, and the route turns them into 404 and 409 respectively."""
    await rulebooks_svc.update_rule(
        world["pref_id"], world["uid"], statement="mine, rewritten.",
    )
    [row] = await rulebooks_svc.recent_preference_drift(world["uid"])
    assert await rulebooks_svc.restore_rule_version(
        world["pref_id"], row["previous"]["id"], world["stranger_id"],
    ) is None