"""Moving a floor, and the reason that has to come with it (#4102).

WHY THIS EXISTS

The operator handed the dial to the model: *"the user should be able to touch it
but the model should be the thing handling it 9 times out of 10."* Everything
here guards the half of that sentence people skip — the operator still has to be
able to see what was done on their behalf and disagree with it.

WHAT THIS PINS

  1. **A reason is required, and "" does not count.** The column is NOT NULL,
     which an empty string satisfies; the service is where the requirement is
     real. This is the whole guardrail: a caller who must write down why has to
     have looked, and a caller who writes down something wrong leaves the
     operator a sentence to argue with. A number that moved silently leaves
     nothing.
  2. **The change is recorded, with what it was before.** Without `old_value` a
     history cannot answer "was this always like that", which is the first
     question anyone asks of a surface behaving oddly.
  3. **A clamp is reported, never swallowed.** A caller that believes it set 1.4
     will read the next telemetry as evidence about a bar that was never in
     force — the same class of error as #3739, one layer up.
  4. **An unknown surface is refused before anything is written.** Settings keys
     are free-form strings, so a typo'd surface would write a key nothing reads:
     a change that reports success and alters nothing.
  5. **The tool teaches the procedure that works**, not the one the numbers
     suggest. Asserted on the docstring because the docstring IS the contract an
     agent reads, and the failure it prevents is a model tuning from a
     percentile — which has been measured pointing the wrong way.
"""
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from scribe.services import retrieval_tuning as rt
from tests.helpers import make_mock_session


def _patches(floor=0.72, budget=3):
    """Patch everything `set_dial` touches except the thing under test."""
    session = make_mock_session()
    return session, (
        patch.object(rt, "async_session", MagicMock(return_value=session)),
        patch.object(rt, "set_setting", AsyncMock()),
        patch.object(rt, "floor_for", AsyncMock(return_value=floor)),
        patch.object(rt, "budget_for", AsyncMock(return_value=budget)),
    )


@pytest.mark.asyncio
@pytest.mark.parametrize("reason", ["", "   ", "noisy", "too high"])
async def test_a_change_without_a_real_reason_is_refused(reason):
    """The guardrail. "too high" is a restatement of the change, not a basis."""
    session, ctx = _patches()
    with ctx[0], ctx[1], ctx[2], ctx[3], pytest.raises(ValueError) as e:
        await rt.set_dial(1, "prompt_rule", "floor", 0.66, reason=reason)
    # The message has to name the tool that produces a real basis, or the
    # caller's next move is a longer sentence rather than a look at the records.
    assert "near_miss_samples" in str(e.value)
    session.add.assert_not_called()


@pytest.mark.asyncio
async def test_nothing_is_written_when_the_reason_is_refused():
    """Refused BEFORE the setting is touched, not after.

    Writing the value and then raising would leave the number moved and the
    history empty — the exact state this table exists to make impossible.
    """
    session, ctx = _patches()
    with ctx[0], patch.object(rt, "set_setting", AsyncMock()) as setter, \
         ctx[2], ctx[3]:
        with pytest.raises(ValueError):
            await rt.set_dial(1, "prompt_rule", "floor", 0.66, reason="x")
        setter.assert_not_called()


@pytest.mark.asyncio
async def test_a_good_change_writes_the_setting_and_the_event():
    session, ctx = _patches(floor=0.72)
    reason = ("read prompt_rule's 5 highest declines: 3 were project rules for "
              "another repo, so the bar is doing its job here")
    with ctx[0], patch.object(rt, "set_setting", AsyncMock()) as setter, \
         ctx[2], ctx[3]:
        out = await rt.set_dial(1, "prompt_rule", "floor", 0.66, reason=reason)

    setter.assert_awaited_once()
    _uid, key, stored = setter.await_args.args
    assert key == "kb_promptrule_threshold" and stored == "0.66"

    event = session.add.call_args.args[0]
    assert event.surface == "prompt_rule" and event.dial == "floor"
    # The before-value is what makes the history answerable.
    assert event.old_value == 0.72 and event.new_value == 0.66
    assert event.reason == reason and event.actor == "model"
    assert out["previous"] == 0.72 and out["applied"] == 0.66
    assert out["clamped"] is False


@pytest.mark.asyncio
@pytest.mark.parametrize("dial, sent, applied", [
    ("floor", 1.4, 1.0),
    ("floor", -0.2, 0.0),
    ("budget", 99, 10),
    ("budget", 0, 1),
])
async def test_a_clamp_is_reported_rather_than_swallowed(dial, sent, applied):
    """Said out loud, because silence here poisons the next reading.

    A caller that believes it set 1.4 treats the following week's telemetry as
    evidence about a bar that never existed, and then moves the dial again to
    fix a problem it invented.
    """
    session, ctx = _patches()
    reason = "checked the refused records for this surface and they were fine"
    with ctx[0], ctx[1], ctx[2], ctx[3]:
        out = await rt.set_dial(1, "auto_inject", dial, sent, reason=reason)
    assert out["applied"] == applied
    assert out["clamped"] is True


@pytest.mark.asyncio
async def test_an_unknown_surface_is_refused_before_anything_is_written():
    session, ctx = _patches()
    with ctx[0], patch.object(rt, "set_setting", AsyncMock()) as setter, \
         ctx[2], ctx[3]:
        with pytest.raises(ValueError):
            await rt.set_dial(1, "pretool_rule", "floor", 0.6,
                              reason="a perfectly good reason that is long enough")
        setter.assert_not_called()
    session.add.assert_not_called()


@pytest.mark.asyncio
@pytest.mark.parametrize("dial", ["threshold", "limit", "", "FLOOR"])
async def test_an_unknown_dial_is_refused(dial):
    """The near-misses are the old vocabulary — `threshold` and `limit` are what
    these were called before this step, so they are exactly what a stale caller
    will send, and a silent no-op there would be indistinguishable from a
    change that did not take."""
    session, ctx = _patches()
    with ctx[0], ctx[1], ctx[2], ctx[3], pytest.raises(ValueError):
        await rt.set_dial(1, "auto_inject", dial, 0.6,
                          reason="a perfectly good reason that is long enough")
    session.add.assert_not_called()


@pytest.mark.asyncio
async def test_a_human_change_is_distinguishable_from_the_models():
    """Both act as the same user, so the id cannot tell them apart — and "did I
    do this, or did the session?" is the first question the history is asked."""
    session, ctx = _patches()
    with ctx[0], ctx[1], ctx[2], ctx[3]:
        await rt.set_dial(1, "auto_inject", "floor", 0.6, actor="human",
                          reason="operator set this themselves in Settings")
    assert session.add.call_args.args[0].actor == "human"


@pytest.mark.asyncio
async def test_an_unknown_actor_is_refused():
    """Free text here would make the column unreadable within a month."""
    session, ctx = _patches()
    with ctx[0], ctx[1], ctx[2], ctx[3], pytest.raises(ValueError):
        await rt.set_dial(1, "auto_inject", "floor", 0.6, actor="agent",
                          reason="a perfectly good reason that is long enough")


def test_the_tool_teaches_reading_the_records_not_the_percentile():
    """The contract an agent actually reads (rule 167).

    The failure this prevents is a model moving a bar because `near_misses.p90`
    sat close to it. That has been measured pointing the wrong way — 69 declines
    where every percentile said "lower it" and the refused record was a false
    positive — so the docstring has to carry the method, not just the warning.
    """
    from scribe.mcp.tools import retrieval_tuning as tool

    doc = tool.tune_retrieval.__doc__
    assert "near_miss_samples" in doc, "the tool does not name how to get records"
    assert "PERCENTILE ALONE" in doc.upper()
    # And the worked example, because an abstract warning loses to a number.
    assert "69" in doc


def test_every_tool_in_the_module_is_registered():
    from scribe.mcp.tools import retrieval_tuning as tool

    from tests.helpers import FakeMCP

    mcp = FakeMCP()
    tool.register(mcp)
    # Order is the module's, and asserted rather than sorted: an unregistered
    # tool is invisible to every caller, so the list is worth reading literally.
    assert mcp.names == [
        "retrieval_surfaces", "migrate_retrieval_floor",
        "tune_retrieval", "retrieval_tuning_history",
    ]


# ── record_release_defaults / floor_moves_since (#4225) ───────────────────
#
# THE HOLE THESE FILL. This table records dial turns — a person or a model
# choosing a number. It was silent about the other way a floor moves: somebody
# edits `floor_default` in the registry and ships it. That change is invisible
# to every consumer of this table, and one consumer is `band_hugs_floor`, which
# compares a band against a floor and could not notice they came from different
# regimes.
#
# Measured on the instance that found it: `write_path_rule` went 0.68 -> 0.72
# on 2026-09-02 as a shipped default, so a 30-day window opening 2026-08-22
# held six days of calls made under the old bar. The readout reported the band
# sitting "-0.0216 above" its floor — which is what a two-population comparison
# looks like when it finally says so out loud.


def _release_rows(rows):
    """Patch `async_session` so the recorder sees `rows` as what is on record.

    `rows` maps (surface, dial) -> new_value already recorded by a release.
    """
    session = make_mock_session()
    seen = []

    async def execute(stmt):
        seen.append(stmt)
        result = MagicMock()
        # The recorder asks one question at a time, in registry order, so the
        # answers are handed back in the order it asks them.
        key = seen_keys.pop(0) if seen_keys else None
        row = None
        if key is not None and key in rows:
            row = MagicMock()
            row.new_value = rows[key]
        result.scalars.return_value.first.return_value = row
        return result

    seen_keys = [(n, d) for n in rt.surface_names() for d in ("floor", "budget")]
    session.execute = AsyncMock(side_effect=execute)
    return session


@pytest.mark.asyncio
async def test_a_first_boot_records_a_baseline_and_calls_it_one():
    """Nothing moved. The row exists so the NEXT release has a predecessor.

    Load-bearing downstream: the band check suspends itself on a genuine move
    and must NOT suspend itself on a fresh install's baseline. It tells them
    apart by `old_value` being null, so a baseline that claimed a change would
    silently retire the check on every new install.
    """
    session = _release_rows({})
    with patch.object(rt, "async_session", MagicMock(return_value=session)):
        written = await rt.record_release_defaults()
    assert written, "a first boot records every dial"
    assert all(w["baseline"] for w in written)
    assert all(w["old_value"] is None for w in written)


@pytest.mark.asyncio
async def test_a_default_that_did_not_move_writes_nothing():
    """Called on every boot, so it has to be idempotent — otherwise the
    history fills with rows saying the release shipped the same number again,
    and a history nobody can skim is one nobody reads."""
    current = {(n, d): (rt.get_surface(n).floor_default if d == "floor"
                        else float(rt.get_surface(n).budget_default))
               for n in rt.surface_names() for d in ("floor", "budget")}
    session = _release_rows(current)
    with patch.object(rt, "async_session", MagicMock(return_value=session)):
        written = await rt.record_release_defaults()
    assert written == []
    session.add.assert_not_called()
    session.commit.assert_not_called()


@pytest.mark.asyncio
async def test_a_moved_default_is_recorded_with_both_values():
    """The event the whole task is about, and it carries what it moved FROM —
    without that a reader knows a change happened and nothing about whether
    the old sample can be pooled with the new one."""
    name = rt.surface_names()[0]
    s = rt.get_surface(name)
    current = {(n, d): (rt.get_surface(n).floor_default if d == "floor"
                        else float(rt.get_surface(n).budget_default))
               for n in rt.surface_names() for d in ("floor", "budget")}
    current[(name, "floor")] = s.floor_default - 0.04    # what the last release shipped
    session = _release_rows(current)
    with patch.object(rt, "async_session", MagicMock(return_value=session)):
        written = await rt.record_release_defaults()
    moved = [w for w in written if w["surface"] == name and w["dial"] == "floor"]
    assert len(moved) == 1
    assert moved[0]["baseline"] is False
    assert moved[0]["old_value"] == pytest.approx(s.floor_default - 0.04)
    assert moved[0]["new_value"] == pytest.approx(s.floor_default)


@pytest.mark.asyncio
async def test_a_release_row_belongs_to_no_account():
    """`user_id IS NULL`, because no user did this.

    A release acts on every account that has not overridden the dial. Writing
    one row per user would both multiply the row and misattribute it, and the
    readers take the newest of (this user's change, the release's) — which only
    works if the release's is distinguishable.
    """
    session = _release_rows({})
    with patch.object(rt, "async_session", MagicMock(return_value=session)):
        await rt.record_release_defaults()
    added = [c.args[0] for c in session.add.call_args_list]
    assert added
    assert all(e.user_id is None for e in added)
    assert all(e.actor == rt.RELEASE_ACTOR for e in added)


@pytest.mark.asyncio
async def test_every_release_row_states_why_it_exists():
    """`reason` is the guardrail on this table, and a row written by machinery
    is the one most likely to arrive blank."""
    session = _release_rows({})
    with patch.object(rt, "async_session", MagicMock(return_value=session)):
        await rt.record_release_defaults()
    added = [c.args[0] for c in session.add.call_args_list]
    assert all(e.reason and e.reason.strip() for e in added)
    assert all(e.surface in e.reason for e in added)


@pytest.mark.asyncio
async def test_a_baseline_is_not_reported_as_a_floor_move():
    """`floor_moves_since` is what suspends the band check, so it must ask for
    a genuine move — `old_value IS NOT NULL` — rather than for any row."""
    from datetime import datetime, timezone
    session = make_mock_session()
    result = MagicMock()
    result.scalars.return_value.all.return_value = []
    session.execute = AsyncMock(return_value=result)
    with patch.object(rt, "async_session", MagicMock(return_value=session)):
        out = await rt.floor_moves_since(datetime.now(timezone.utc))
    assert out == {}
    # The filter is the whole correctness argument; assert it is in the query.
    stmt = str(session.execute.call_args.args[0])
    assert "old_value IS NOT NULL" in stmt
    assert "dial" in stmt


# ── floor_history_gaps: "no move recorded" is not "no move" ───────────────
#
# THE HOLE ITS COMPANION LEFT. `floor_moves_since` answers "did this arm's
# floor move inside the window", and the band check read an empty answer as
# "no, it held steady". Those are the same answer only where the ledger was
# WATCHING. Before an arm's first row there is nothing to move, nothing to
# report, and no way to tell a steady floor from an unrecorded change.
#
# Measured the day the ledger was seeded on this instance: its earliest event
# of any kind is 2026-09-17, its `write_path_rule` floor rows begin at
# 2026-09-21, and a 30-day window opening 2026-08-22 reported that arm's band
# sitting "-0.0208 above" its floor. That negative distance is the documented
# tell of a raise inside the window — announced by the check the suspension
# was built to stop, because the suspension had nothing to read.


def _gap_rows(rows):
    """Patch `async_session` so the reader sees `rows` as the whole ledger.

    `rows` maps surface -> the datetime of its earliest floor event. A surface
    left out has no floor history at all, which is the harder of the two
    cases: absent from the ledger AND present in the registry.
    """
    session = make_mock_session()
    result = MagicMock()
    result.all.return_value = list(rows.items())
    session.execute = AsyncMock(return_value=result)
    return session


@pytest.mark.asyncio
async def test_an_arm_recorded_before_the_window_opens_has_no_gap():
    """The ordinary case, and the one that must stay silent: a ledger that
    covers the window means silence from `floor_moves_since` is trustworthy,
    and the band check should go ahead and judge."""
    from datetime import datetime, timedelta, timezone
    since = datetime(2026, 8, 22, tzinfo=timezone.utc)
    covered = {n: since - timedelta(days=1) for n in rt.surface_names()}
    session = _gap_rows(covered)
    with patch.object(rt, "async_session", MagicMock(return_value=session)):
        out = await rt.floor_history_gaps(since)
    assert out == {}, "a covered arm is absent from the result, not present-and-false"


@pytest.mark.asyncio
async def test_an_arm_first_recorded_inside_the_window_reports_when_it_clears():
    """The symptom case. A finding with no remedy is a complaint, so the value
    IS the date from which the floor becomes knowable."""
    from datetime import datetime, timezone
    since = datetime(2026, 8, 22, tzinfo=timezone.utc)
    first = datetime(2026, 9, 21, 11, 58, tzinfo=timezone.utc)
    name = rt.surface_names()[0]
    rows = {n: since for n in rt.surface_names()}
    rows[name] = first
    session = _gap_rows(rows)
    with patch.object(rt, "async_session", MagicMock(return_value=session)):
        out = await rt.floor_history_gaps(since)
    assert set(out) == {name}
    assert out[name] == first.isoformat()


@pytest.mark.asyncio
async def test_an_arm_with_no_floor_history_at_all_is_reported_as_unknowable():
    """Distinct from "recorded too late", and it has a different remedy: there
    is no date to wait for, so the caller is told to start the record. Mapping
    it to None rather than leaving it out is the point — leaving it out is what
    the band check was already doing wrong."""
    from datetime import datetime, timezone
    since = datetime(2026, 8, 22, tzinfo=timezone.utc)
    names = rt.surface_names()
    session = _gap_rows({n: since for n in names[1:]})
    with patch.object(rt, "async_session", MagicMock(return_value=session)):
        out = await rt.floor_history_gaps(since)
    assert out == {names[0]: None}


@pytest.mark.asyncio
async def test_a_baseline_counts_here_though_it_is_not_a_move():
    """The one place this and `floor_moves_since` deliberately disagree.

    A baseline records a default without changing it, so it is not a move and
    that function filters it out. It IS the ledger beginning to watch the arm,
    and from that moment silence genuinely means the floor held — so filtering
    it out here would suspend the band check forever on an install whose only
    floor rows are baselines, which is every install that has never tuned.
    """
    from datetime import datetime, timezone
    session = _gap_rows({n: datetime(2026, 1, 1, tzinfo=timezone.utc)
                         for n in rt.surface_names()})
    with patch.object(rt, "async_session", MagicMock(return_value=session)):
        out = await rt.floor_history_gaps(datetime(2026, 8, 22, tzinfo=timezone.utc))
    assert out == {}
    stmt = str(session.execute.call_args.args[0])
    assert "old_value IS NOT NULL" not in stmt, (
        "a baseline is not a move, but it is the ledger observing the arm"
    )
    assert "dial" in stmt and "min" in stmt.lower()


@pytest.mark.asyncio
async def test_the_registry_supplies_the_arms_not_the_ledger():
    """An arm the ledger has never heard of is exactly the one at risk, so it
    cannot be the ledger that decides which arms get asked about."""
    from datetime import datetime, timezone
    session = _gap_rows({})
    with patch.object(rt, "async_session", MagicMock(return_value=session)):
        out = await rt.floor_history_gaps(datetime(2026, 8, 22, tzinfo=timezone.utc))
    assert set(out) == set(rt.surface_names())
    assert all(v is None for v in out.values())


@pytest.mark.asyncio
async def test_a_row_written_exactly_as_the_window_opens_covers_it():
    """The boundary, from both sides. `since` is the first instant the window
    contains, so a floor recorded AT it leaves nothing earlier unaccounted
    for — and one microsecond later does. Asserted because an off-by-one here
    fails in the expensive direction: a covered arm reported as unknowable
    retires a working check on the strength of a rounding."""
    from datetime import datetime, timedelta, timezone
    since = datetime(2026, 8, 22, tzinfo=timezone.utc)
    names = rt.surface_names()

    session = _gap_rows({n: since for n in names})
    with patch.object(rt, "async_session", MagicMock(return_value=session)):
        assert await rt.floor_history_gaps(since) == {}

    later = since + timedelta(microseconds=1)
    session = _gap_rows({n: later for n in names})
    with patch.object(rt, "async_session", MagicMock(return_value=session)):
        assert set(await rt.floor_history_gaps(since)) == set(names)