"""Tests for the feedback fold's dry run, and for the units that will run it.

The fold is irreversible: ``add_outcome`` accumulates into a running mean and
``recompute_category`` re-derives every row in the category from the new peer
rate, so there is no pre-fold value left anywhere afterwards. The whole point
of ``feedback_preview`` is to let an operator look before taking that step,
which makes ONE property load-bearing above all others -- a dry run must leave
the database it previewed byte-for-byte identical. That is the first test here
and it asserts on the file's sha256, not just on the rows, because a preview
that dirtied a page would still be a preview that wrote.

The second load-bearing property is honesty: the projection must be what the
real fold actually does, not a second implementation of the arithmetic that can
drift. ``test_the_projection_is_what_the_real_fold_does`` pins that by folding
for real onto an identical copy and comparing every score.

Everything here is offline: a temp SQLite DB seeded from ``config/schema.sql``,
no network, no provider call, no dispatcher import.
"""

from __future__ import annotations

import configparser
import csv
import hashlib
import io
import sqlite3
import sys
from pathlib import Path

import pytest

import feedback
import feedback_preview as fp
from config import load_config
from proficiency_store import add_outcome

ROOT = Path(__file__).resolve().parent.parent
SCHEMA_SQL = (ROOT / "config" / "schema.sql").read_text()
CFG = load_config(str(ROOT / "config" / "config.yaml"))

CATEGORY = "coding_general"


def _seed(path: Path, *, failures: int = 8, successes: int = 0) -> Path:
    """A two-model database with one model carrying an unapplied backlog.

    ``m1`` gets the new outcomes, so it moves directly. ``m2`` gets none and
    moves anyway, by ripple, because ``recompute_category`` re-derives the whole
    category against the peer rate m1's evidence just changed. Both halves of
    the projection therefore have something to report.
    """
    conn = sqlite3.connect(str(path))
    conn.row_factory = sqlite3.Row
    conn.executescript(SCHEMA_SQL)
    for model_id in ("m1", "m2"):
        conn.execute(
            """
            INSERT INTO models (model_id, provider, base_model_id, availability,
                                last_updated)
            VALUES (?, 'nw', ?, 'active', '2026-09-01T00:00:00+00:00')
            """,
            (model_id, model_id),
        )
    conn.commit()
    add_outcome(conn, CFG, "m1", "nw", CATEGORY, [1.0] * 10)
    add_outcome(conn, CFG, "m2", "nw", CATEGORY, [1.0] * 10)
    for verdict, n in (("failed", failures), ("succeeded", successes)):
        for _ in range(n):
            conn.execute(
                """
                INSERT INTO verifications (model_id, provider, task_category,
                                           kind, verdict, observed_at)
                VALUES ('m1', 'nw', ?, 'client_outcome', ?,
                        '2026-09-01T00:00:00+00:00')
                """,
                (CATEGORY, verdict),
            )
    conn.commit()
    conn.close()
    return path


@pytest.fixture()
def seeded(tmp_path: Path) -> Path:
    return _seed(tmp_path / "router.db")


def _digest(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def _rows(path: Path, sql: str) -> list[tuple]:
    conn = sqlite3.connect(str(path))
    try:
        return conn.execute(sql).fetchall()
    finally:
        conn.close()


PROFICIENCY_DUMP = """
    SELECT model_id, provider, category, leaderboard_score, self_eval_score,
           self_eval_samples, outcome_score, outcome_samples, blended_score,
           source, inherited_from, last_updated
      FROM proficiency ORDER BY model_id, category
"""

APPLIED_DUMP = "SELECT id, applied_at FROM verifications ORDER BY id"


# --- the one that matters: a dry run writes nothing -------------------------


def test_a_dry_run_leaves_the_database_byte_identical(seeded: Path) -> None:
    """The load-bearing guarantee. Rows, applied_at stamps, and the file itself.

    Checked three ways on purpose. The row dumps say what an operator would
    notice; the sha256 says the preview did not so much as dirty a page, which
    is the claim the docstring and the `--db` flag both make.
    """
    before_prof = _rows(seeded, PROFICIENCY_DUMP)
    before_applied = _rows(seeded, APPLIED_DUMP)
    before_digest = _digest(seeded)

    changes, suppressed, grouped = fp.project(seeded, CFG)

    assert _rows(seeded, PROFICIENCY_DUMP) == before_prof
    assert _rows(seeded, APPLIED_DUMP) == before_applied
    assert all(applied_at is None for _, applied_at in before_applied)
    assert _digest(seeded) == before_digest

    # ...and it was not a no-op that trivially wrote nothing.
    assert sum(len(v) for v in grouped.values()) == 8
    assert [c.model_id for c in changes if c.kind == "direct"] == ["m1"]


def test_the_projection_reports_the_delta_it_would_apply(seeded: Path) -> None:
    """A known backlog produces a known, signed, non-trivial movement."""
    changes, _suppressed, grouped = fp.project(seeded, CFG)
    direct = {(c.model_id, c.category): c for c in changes if c.kind == "direct"}

    m1 = direct[("m1", CATEGORY)]
    assert (m1.successes, m1.failures) == (0, 8)
    assert m1.before_score == pytest.approx(1.0)
    # 10 samples at 1.0 plus 8 at 0.0 -- the evidence count is exact even where
    # the empirical-Bayes score depends on the shipped prior.
    assert (m1.before_samples, m1.after_samples) == (10, 18)
    assert m1.delta < -0.1
    assert m1.after_score == pytest.approx(m1.before_score + m1.delta)
    assert grouped[("m1", "nw", CATEGORY)] and len(grouped) == 1


def test_the_projection_is_what_the_real_fold_does(tmp_path: Path) -> None:
    """Project one copy, fold another for real, and demand the same numbers.

    This is what stops the preview from becoming a second implementation of the
    empirical-Bayes conversion. If ``proficiency_outcome`` changes and the
    preview stops tracking it, a confidently wrong forecast of an irreversible
    action is the worst failure available here, so it fails loudly instead.
    """
    projected_db = _seed(tmp_path / "projected.db")
    applied_db = _seed(tmp_path / "applied.db")
    assert _digest(projected_db) != "" and _digest(applied_db) != ""

    changes, _suppressed, _grouped = fp.project(projected_db, CFG)
    forecast = {(c.model_id, c.category): c.after_score for c in changes}

    conn = sqlite3.connect(str(applied_db))
    conn.row_factory = sqlite3.Row
    try:
        grouped = feedback.summarize(feedback.unapplied_failures(conn))
        feedback.apply_failures(conn, CFG, grouped, dry_run=False)
        actual = {
            (r["model_id"], r["category"]): r["blended_score"]
            for r in conn.execute(
                "SELECT model_id, category, blended_score FROM proficiency"
            )
        }
    finally:
        conn.close()

    assert forecast
    for key, after in forecast.items():
        assert actual[key] == pytest.approx(after), key


def test_the_source_is_opened_read_only(seeded: Path) -> None:
    """Even a caller that tries to write through the returned handle cannot.

    The copy is writable -- the fold has to run somewhere -- but it is the copy,
    and the source is closed again before anything touches it.
    """
    before = _digest(seeded)
    conn = fp.copy_database(seeded)
    try:
        assert conn.execute("SELECT COUNT(*) FROM proficiency").fetchone()[0] == 2
        conn.execute("DELETE FROM proficiency")
        conn.commit()
        assert conn.execute("SELECT COUNT(*) FROM proficiency").fetchone()[0] == 0
    finally:
        conn.close()
    assert _digest(seeded) == before
    assert _rows(seeded, "SELECT COUNT(*) FROM proficiency")[0][0] == 2


def test_a_missing_database_fails_instead_of_being_created(tmp_path: Path) -> None:
    """``mode=ro`` refuses rather than conjuring an empty DB and reporting 'none'.

    Without this, a typo in ``--db`` would produce a clean, confident, entirely
    fictional 'nothing would change'.
    """
    missing = tmp_path / "not-here.db"
    with pytest.raises(sqlite3.OperationalError):
        fp.copy_database(missing)
    assert not missing.exists()


# --- direct vs ripple, and what the summary is allowed to hide --------------


def test_ripple_rows_move_without_evidence_of_their_own(seeded: Path) -> None:
    """m2 was never reported on and its score still moves. That has to be visible."""
    changes, _suppressed, _grouped = fp.project(seeded, CFG)
    ripple = [c for c in changes if c.kind == "ripple"]
    assert [c.model_id for c in ripple] == ["m2"]
    assert (ripple[0].successes, ripple[0].failures) == (0, 0)
    assert ripple[0].before_samples == ripple[0].after_samples == 10
    assert ripple[0].delta < 0


def test_rounding_sized_moves_are_counted_not_dropped(seeded: Path) -> None:
    """Suppression below MOVE_EPSILON is reported as a number, never silently."""
    before = {
        ("m1", "nw", CATEGORY): {"blended_score": 0.5, "source": "s",
                                 "outcome_score": 0.5, "outcome_samples": 1},
        ("m2", "nw", CATEGORY): {"blended_score": 0.5, "source": "s",
                                 "outcome_score": 0.5, "outcome_samples": 1},
    }
    after = {
        ("m1", "nw", CATEGORY): dict(before[("m1", "nw", CATEGORY)],
                                     blended_score=0.5 + fp.MOVE_EPSILON / 10),
        ("m2", "nw", CATEGORY): dict(before[("m2", "nw", CATEGORY)],
                                     blended_score=0.5 + fp.MOVE_EPSILON * 10),
    }
    changes, suppressed = fp.diff(before, after, grouped={})
    assert [c.model_id for c in changes] == ["m2"]
    assert suppressed == 1
    rendered = fp.format_projection(changes, suppressed, {}, 0)
    assert "1 further row(s) moved by less than" in rendered


def test_a_row_carrying_evidence_is_shown_even_when_nothing_moves() -> None:
    """"You spent 42 samples and the score did not budge" is a real answer."""
    row = {"blended_score": 0.5, "source": "s", "outcome_score": 0.5,
           "outcome_samples": 1}
    key = ("m1", "nw", CATEGORY)
    changes, suppressed = fp.diff(
        {key: row}, {key: dict(row)}, grouped={key: [(1, 0.0), (2, 1.0)]}
    )
    assert suppressed == 0
    assert len(changes) == 1 and changes[0].kind == "direct"
    assert (changes[0].successes, changes[0].failures) == (1, 1)
    assert changes[0].delta == pytest.approx(0.0)


def test_a_brand_new_row_renders_as_new_rather_than_a_delta() -> None:
    """There is no prior belief to revise, so a signed delta would be a lie."""
    key = ("m3", "nw", CATEGORY)
    after = {key: {"blended_score": 0.4, "source": "outcome_blended",
                   "outcome_score": 0.4, "outcome_samples": 2}}
    changes, _suppressed = fp.diff({}, after, grouped={key: [(1, 0.0)]})
    assert len(changes) == 1
    assert changes[0].is_new and changes[0].delta is None
    assert changes[0].magnitude == 0.0
    assert "new" in fp.format_projection(changes, 0, {key: [(1, 0.0)]}, 1)


# --- CSV ---------------------------------------------------------------------


def test_csv_carries_every_projected_row(seeded: Path) -> None:
    """The table digests ripple per category; --csv is where per-row lives."""
    changes, _suppressed, _grouped = fp.project(seeded, CFG)
    out = io.StringIO()
    fp.write_csv(changes, out)
    parsed = list(csv.DictReader(io.StringIO(out.getvalue())))
    assert list(parsed[0]) == fp.CSV_COLUMNS
    assert len(parsed) == len(changes)
    assert {r["kind"] for r in parsed} == {"direct", "ripple"}
    for row, change in zip(parsed, changes):
        assert row["model_id"] == change.model_id
        assert float(row["delta"]) == pytest.approx(change.delta)


# --- the CLI -----------------------------------------------------------------


def _run_main(monkeypatch, argv: list[str], db_path: Path) -> int:
    cfg = load_config(str(ROOT / "config" / "config.yaml"))
    monkeypatch.setattr(cfg.database, "path", str(db_path))
    monkeypatch.setattr(feedback, "load_config", lambda _path: cfg)
    monkeypatch.setattr(sys, "argv", ["feedback", *argv])
    return feedback.main()


def test_dry_run_against_a_copy_writes_nothing_and_says_so(
    monkeypatch, capsys, seeded: Path
) -> None:
    """End to end through the CLI, the way the operator will actually run it."""
    before = _digest(seeded)
    assert _run_main(monkeypatch, ["--dry-run", "--db", str(seeded)], seeded) == 0
    out = capsys.readouterr().out

    assert _digest(seeded) == before
    assert "would apply 8 sample(s)" in out
    assert "nothing was written" in out
    assert "IRREVERSIBLE" in out
    assert "m1" in out


def test_dry_run_csv_is_only_the_csv(monkeypatch, capsys, seeded: Path) -> None:
    before = _digest(seeded)
    assert _run_main(
        monkeypatch, ["--dry-run", "--csv", "--db", str(seeded)], seeded
    ) == 0
    out = capsys.readouterr().out
    assert _digest(seeded) == before
    assert out.splitlines()[0].split(",") == fp.CSV_COLUMNS
    assert "nothing was written" not in out


def test_csv_without_dry_run_is_refused(monkeypatch, seeded: Path) -> None:
    """Accepting and discarding it would drop a flag from the irreversible run."""
    with pytest.raises(SystemExit) as excinfo:
        _run_main(monkeypatch, ["--csv", "--db", str(seeded)], seeded)
    assert excinfo.value.code == 2


def test_an_empty_backlog_reports_nothing_to_do(
    monkeypatch, capsys, tmp_path: Path
) -> None:
    empty = _seed(tmp_path / "empty.db", failures=0)
    before = _digest(empty)
    assert _run_main(monkeypatch, ["--dry-run", "--db", str(empty)], empty) == 0
    assert "nothing to fold in" in capsys.readouterr().out
    assert _digest(empty) == before


def test_empty_backlog_csv_emits_header(
    monkeypatch, capsys, tmp_path: Path
) -> None:
    """--dry-run --csv with no unapplied signals still writes the CSV header."""
    empty = _seed(tmp_path / "empty-csv.db", failures=0)
    before = _digest(empty)
    assert (
        _run_main(monkeypatch, ["--dry-run", "--csv", "--db", str(empty)], empty)
        == 0
    )
    out = capsys.readouterr().out
    assert out.splitlines()[0].split(",") == fp.CSV_COLUMNS
    assert len(out.splitlines()) == 1
    assert _digest(empty) == before


def test_the_dry_run_default_target_is_the_configured_database(
    monkeypatch, capsys, seeded: Path
) -> None:
    """No --db: the preview still reads the live path read-only and writes nothing."""
    before = _digest(seeded)
    assert _run_main(monkeypatch, ["--dry-run"], seeded) == 0
    assert "would apply 8 sample(s)" in capsys.readouterr().out
    assert _digest(seeded) == before


# --- the units ship unstarted -------------------------------------------------

DEPLOY = ROOT / "deploy"


def _unit(name: str) -> configparser.RawConfigParser:
    """Parse a systemd unit.

    ``RawConfigParser`` because unit files are full of ``%h`` and configparser's
    default interpolation would choke on it, and ``strict=False`` because
    systemd permits a directive to repeat (the poller's two ``ExecStart`` lines).
    """
    parser = configparser.RawConfigParser(strict=False, allow_no_value=True)
    parser.optionxform = str
    parser.read_string((DEPLOY / name).read_text())
    return parser


def test_the_feedback_service_cannot_be_enabled_on_its_own() -> None:
    """The first fold is an operator decision, so the pair ships unstarted.

    A unit with no ``[Install]`` has no install target to enable, which is what
    makes "not enabled by default" structural rather than a README promise. The
    timer keeps its ``[Install]`` -- enabling the timer is exactly the deliberate
    act this arrangement is asking for.
    """
    service = _unit("llm-router-feedback.service")
    assert not service.has_section("Install")
    assert service["Service"]["Type"] == "oneshot"
    assert service["Service"]["ExecStart"].endswith("-m feedback")
    assert "--dry-run" not in service["Service"]["ExecStart"]

    timer = _unit("llm-router-feedback.timer")
    assert timer["Install"]["WantedBy"] == "timers.target"


def test_the_feedback_service_is_hardened_like_its_siblings() -> None:
    """Same sandbox as the poller. A weaker one would be a silent downgrade."""
    hardening = ("NoNewPrivileges", "PrivateTmp", "ProtectSystem", "ProtectHome",
                 "ReadWritePaths", "WorkingDirectory")
    feedback_svc = _unit("llm-router-feedback.service")["Service"]
    poller_svc = _unit("llm-router-poller.service")["Service"]
    for key in hardening:
        assert feedback_svc[key] == poller_svc[key], key

    # Environment= repeats, and configparser keeps only the last, so read the
    # raw text: src/ has to be importable for `-m feedback` to resolve at all.
    raw = (DEPLOY / "llm-router-feedback.service").read_text()
    assert "Environment=PYTHONPATH=%h/llm-router/src" in raw
    # Unbuffered where the siblings are not, on purpose: this run is the
    # irreversible one and the journal is the only record of what it folded.
    assert "Environment=PYTHONUNBUFFERED=1" in raw


def test_the_feedback_service_needs_no_provider_key() -> None:
    """It reads `verifications` and writes `proficiency`; both are local.

    An ``EnvironmentFile`` would make the fold fail on a missing ``.env`` for a
    key it never uses, and pull a unit that makes no network call into the
    blast radius of the provider credential.
    """
    service = _unit("llm-router-feedback.service")
    assert "EnvironmentFile" not in service["Service"]
    assert not service.has_option("Unit", "After")
    assert not service.has_option("Unit", "Wants")


def test_the_feedback_timer_does_not_carry_an_inert_persistent() -> None:
    """systemd.timer(5): Persistent= only has an effect with OnCalendar=.

    This timer is monotonic, so the line would read as catch-up insurance while
    buying nothing; ``OnBootSec`` is the real catch-up. Scoped to this unit --
    the poller and seed timers carry the same inert line and are not in this
    change's remit.
    """
    timer = _unit("llm-router-feedback.timer")["Timer"]
    assert "OnCalendar" not in timer
    assert "Persistent" not in timer
    assert timer["OnUnitActiveSec"] == "12h"
    assert timer["OnBootSec"] == "30min"