6krrt/tests/test_outcome_attribution.py

"""Tests for which completion an outcome report attaches to.

`POST /outcome` is the only ground truth the router gets, so attributing one
to the wrong conversation is worse than losing it: a model gets penalized for
work it never did. The guard is a short window plus a refusal -- if more than
one conversation was served inside it, the report is refused rather than
guessed at.

The window was not a window. `observed_at` is written by
`datetime.now(timezone.utc).isoformat()`, which separates date from time with
'T', while `datetime('now', ...)` returns a space. Compared as strings, 'T'
sorts after ' ', so the time of day never participated once the dates matched
and a 120-second window admitted everything served that day. Measured on the
live DB: 14 rows across 2 sessions where the correct comparison matched 0.

These tests write timestamps through the same call the dispatcher uses, so
they stay honest if that format ever changes.
"""

import sqlite3
from datetime import datetime, timedelta, timezone
from pathlib import Path

import pytest

import dispatcher
from dispatcher import AMBIGUOUS, SEED_CATEGORY, _most_recent_if_unambiguous
from metrics import quota_burn

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

# Short, so "earlier today" is unambiguously outside it. The old row below sits
# at the first instant of the current UTC day -- the earliest time that still
# shares today's date, which is what the string comparison needed to go wrong.
WINDOW_SECONDS = 5


def _now() -> datetime:
    return datetime.now(timezone.utc)


def _start_of_today() -> datetime:
    return _now().replace(hour=0, minute=0, second=0, microsecond=1)


@pytest.fixture(autouse=True)
def short_window(monkeypatch):
    monkeypatch.setattr(
        dispatcher.cfg.verification,
        "outcome_attribution_window_seconds",
        WINDOW_SECONDS,
    )


@pytest.fixture
def db(tmp_path):
    conn = sqlite3.connect(tmp_path / "test.db")
    conn.row_factory = sqlite3.Row
    conn.executescript(SCHEMA_SQL)
    yield conn
    conn.close()


def _observe(conn, *, when: datetime, session_key: str, request_id: str,
             category: str = "coding_general"):
    conn.execute(
        """
        INSERT INTO energy_observations (
            model_id, provider, request_id, session_key, task_category, observed_at
        ) VALUES ('m', 'neuralwatt', ?, ?, ?, ?)
        """,
        # Written exactly the way log_observation writes it.
        (request_id, session_key, category, when.isoformat()),
    )
    conn.commit()


def test_a_session_from_earlier_today_is_outside_the_window(db):
    """The regression: same calendar date is not the same as inside 5 seconds."""
    _observe(db, when=_start_of_today(), session_key="morning", request_id="old")
    _observe(db, when=_now(), session_key="now", request_id="new")

    row = _most_recent_if_unambiguous(db)

    assert row is not AMBIGUOUS, "an 8-hours-stale session must not create ambiguity"
    assert row["request_id"] == "new"


def test_an_old_session_alone_is_not_recent_enough_to_attribute(db):
    """With nothing inside the window there is nothing to attach to -- 404, not a guess."""
    _observe(db, when=_start_of_today(), session_key="morning", request_id="old")

    assert _most_recent_if_unambiguous(db) is None


def test_two_live_conversations_are_refused(db):
    """The guard the window exists to serve, still working."""
    _observe(db, when=_now(), session_key="alice", request_id="a")
    _observe(db, when=_now(), session_key="bob", request_id="b")

    assert _most_recent_if_unambiguous(db) is AMBIGUOUS


def test_one_conversation_across_several_turns_is_not_ambiguous(db):
    """Several completions, one session -- the ordinary case must still attribute."""
    for i in range(3):
        _observe(db, when=_now(), session_key="alice", request_id=f"a{i}")

    row = _most_recent_if_unambiguous(db)

    assert row is not AMBIGUOUS
    assert row["request_id"] == "a2"


def test_the_reference_sweep_is_not_a_conversation(db):
    """seed_energy's traffic must never make a real report look ambiguous."""
    _observe(db, when=_now(), session_key=None, request_id="seed",
             category=SEED_CATEGORY)
    _observe(db, when=_now(), session_key="alice", request_id="a")

    row = _most_recent_if_unambiguous(db)

    assert row is not AMBIGUOUS
    assert row["request_id"] == "a"


def test_quota_burn_counts_only_the_last_thirty_days(db, tmp_path, monkeypatch):
    """Same comparison, same fix -- an old row must not inflate the burn figure.

    The row sits at the START of the day 30 days ago: the cutoff falls on that
    same date but later in it, which is exactly the case a string comparison
    gets wrong. A row from 45 days ago would be excluded either way and would
    pin nothing.
    """
    monkeypatch.setattr(dispatcher.cfg.objective, "plan_kwh_per_period", 6.25)
    just_outside = (_now() - timedelta(days=30)).replace(
        hour=0, minute=0, second=0, microsecond=1
    )
    db.execute(
        """
        INSERT INTO energy_observations (model_id, provider, energy_kwh, observed_at)
        VALUES ('m', 'neuralwatt', 1.0, ?)
        """,
        (just_outside.isoformat(),),
    )
    db.execute(
        """
        INSERT INTO energy_observations (model_id, provider, energy_kwh, observed_at)
        VALUES ('m', 'neuralwatt', 0.25, ?)
        """,
        (_now().isoformat(),),
    )
    db.commit()

    assert quota_burn(db, dispatcher.cfg)["metered_kwh_30d"] == pytest.approx(0.25)


# ---------------------------------------------------------------------------
# Degraded classifications must not train proficiency.
# ---------------------------------------------------------------------------
#
# A fallback-classified request is recorded with task_category=general_chat,
# and general_chat is a fully scored category AND the configured
# fallback_category. Without this guard, outcome reports on mislabelled outage
# traffic fold into proficiency(model, general_chat) via feedback.py and drag
# real scores toward whatever was flowing while the classifier was down.


def _decision_with_source(conn, request_id: str, source: str) -> None:
    conn.execute(
        """
        INSERT INTO route_decisions
            (kind, task_category, task_tier, required_context_tokens,
             selected_model, selected_provider, classification_source,
             request_id, observed_at)
        VALUES ('chat','general_chat',2,100,'m','neuralwatt',?,?,?)
        """,
        (source, request_id, _now().isoformat()),
    )
    conn.commit()


@pytest.mark.parametrize("source", ["fallback", "session_stale", "session_history"])
def test_degraded_source_outcome_is_not_attributable(db, source):
    """A guessed or borrowed category must not move proficiency."""
    _decision_with_source(db, "rid-degraded", source)
    assert dispatcher._outcome_is_attributable(db, "rid-degraded") is False


@pytest.mark.parametrize(
    "source", ["classifier", "cached", "override", "classifier_cloud"]
)
def test_trusted_source_outcome_is_attributable(db, source):
    """A measured category still trains proficiency, cloud included.

    classifier_cloud is a real classification by a real model — degrading its
    attribution too would throw away good ground truth.
    """
    _decision_with_source(db, "rid-trusted", source)
    assert dispatcher._outcome_is_attributable(db, "rid-trusted") is True


def test_unknown_decision_row_fails_open(db):
    """No matching decision -> attributable.

    Withholding ground truth needs positive knowledge the category was a
    guess. Over-applying an exclusion already starved feedback once (the
    client_capped lesson), so the default direction is to keep the signal.

    NOTE: this test passes on unmodified main by design — it pins the
    fail-open rule against a future over-correction, not a fixed defect.
    """
    assert dispatcher._outcome_is_attributable(db, "rid-does-not-exist") is True
    assert dispatcher._outcome_is_attributable(db, None) is True


def test_degraded_sources_match_the_cascade_steps():
    """The exclusion set and the cascade's degraded steps cannot drift apart.

    If a new degraded source is added to Classification without adding it
    here, its outcomes silently start training proficiency again.
    """
    literal_sources = set(
        dispatcher.Classification.model_fields["source"].annotation.__args__
    )
    trusted = {"classifier", "override", "cached", "classifier_cloud"}
    assert dispatcher.DEGRADED_CLASSIFICATION_SOURCES == literal_sources - trusted