"""
Unit tests for the AI coach: feature engineering, prompt parsing, and the
briefing / trend-insight / Copilot endpoints.

Every model call is mocked. The suite never reaches the ai-gateway, so it is
neither slow nor dependent on that box being up.
"""
import json

import pytest

from services import ai as ai_svc
from services import analysis as analysis_svc
from services import coach
from services import insights


def day(date, **metrics):
    """One row in the shape `health.get_summary` returns."""
    sleep = metrics.pop("sleep", None)
    row = {"date": date, **metrics}
    row["sleep"] = sleep
    return row


def flat_days(n, start=1, **series):
    """`n` consecutive days from 2026-08-01, each metric a constant or list."""
    rows = []
    for i in range(n):
        values = {}
        for key, value in series.items():
            values[key] = value[i] if isinstance(value, list) else value
        rows.append(day(f"2026-08-{start + i:02d}", **values))
    return rows


# --- feature engineering ----------------------------------------------------
class TestFlatten:
    def test_sleep_stages_become_percentages_of_time_asleep(self):
        row = day("2026-08-01", sleep={
            "duration": 8.0, "quality": 80, "deepSeconds": 3600,
            "remSeconds": 7200, "lightSeconds": None, "awakeSeconds": None,
        })
        flat = insights._flatten(row)
        assert flat["sleepDeepPct"] == 12.5
        assert flat["sleepRemPct"] == 25.0

    def test_missing_stage_is_absent_not_zero(self):
        flat = insights._flatten(day("2026-08-01", sleep={"duration": 7.0}))
        assert "sleepDeepPct" not in flat

    def test_no_sleep_record_leaves_no_sleep_fields(self):
        flat = insights._flatten(day("2026-08-01", steps=100))
        assert "sleepDuration" not in flat

    def test_sedentary_seconds_become_hours(self):
        flat = insights._flatten(day("2026-08-01", sedentarySeconds=5400))
        assert flat["sedentaryHours"] == 1.5


class TestDeviations:
    def test_z_score_measures_departure_from_the_personal_baseline(self):
        # A baseline that varies, as real data does: mean 1000, sd 100.
        baseline = [900, 1000, 1100, 900, 1000, 1100, 900, 1000, 1100, 1000]
        history = [
            insights._flatten(r) for r in flat_days(11, steps=baseline + [1800])
        ]
        result = {d["metric"]: d for d in insights.deviations(history, history[-1])}
        assert result["steps"]["baselineMean"] == 1000
        assert result["steps"]["z"] > 3

    def test_today_is_excluded_from_its_own_baseline(self):
        rows = [insights._flatten(r) for r in flat_days(
            8, heartRate=[60, 60, 60, 60, 60, 60, 60, 70]
        )]
        result = {d["metric"]: d for d in insights.deviations(rows, rows[-1])}
        # Including today would pull the mean up to 61.25 and shrink the z.
        assert result["heartRate"]["baselineMean"] == 60

    def test_too_little_history_reports_insufficient_baseline(self):
        rows = [insights._flatten(r) for r in flat_days(3, steps=5000)]
        result = {d["metric"]: d for d in insights.deviations(rows, rows[-1])}
        assert result["steps"]["verdict"] == "基线不足"
        assert result["steps"]["z"] is None

    def test_a_flat_baseline_reports_no_z_rather_than_a_fabricated_zero(self):
        """Dividing by a zero standard deviation is undefined; calling the day
        'z = 0' would label a genuine departure as perfectly typical."""
        rows = [insights._flatten(r) for r in flat_days(8, steps=[5000] * 7 + [9000])]
        result = {d["metric"]: d for d in insights.deviations(rows, rows[-1])}
        assert result["steps"]["z"] is None
        assert result["steps"]["verdict"] == "基线无波动"

    def test_a_flat_baseline_matched_exactly_is_just_normal(self):
        rows = [insights._flatten(r) for r in flat_days(8, steps=5000)]
        result = {d["metric"]: d for d in insights.deviations(rows, rows[-1])}
        assert result["steps"]["verdict"] == "正常"

    def test_direction_is_judged_per_metric_not_by_sign(self):
        wobble = [58, 60, 62, 58, 60, 62, 60]
        rows = [insights._flatten(r) for r in flat_days(
            8,
            heartRate=wobble + [75],
            heartRateVariability=[38, 40, 42, 38, 40, 42, 40] + [55],
        )]
        result = {d["metric"]: d for d in insights.deviations(rows, rows[-1])}
        # Both moved up; only one of them is good news.
        assert result["heartRate"]["verdict"] == "偏差"
        assert result["heartRateVariability"]["verdict"] == "偏好"

    def test_largest_departure_comes_first(self):
        rows = [insights._flatten(r) for r in flat_days(
            8, steps=[5000] * 7 + [5100], heartRate=[60] * 7 + [80]
        )]
        result = insights.deviations(rows, rows[-1])
        assert result[0]["metric"] == "heartRate"

    def test_metrics_absent_today_are_omitted(self):
        rows = [insights._flatten(r) for r in flat_days(8, steps=5000)]
        assert all(d["metric"] == "steps" for d in insights.deviations(rows, rows[-1]))


class TestTrends:
    def test_slope_is_reported_per_thirty_days(self):
        rows = [insights._flatten(r) for r in flat_days(
            30, heartRateVariability=[40 + i for i in range(30)]
        )]
        entry = {t["metric"]: t for t in insights.trends(rows)}
        # One unit a day is 30 per 30 days.
        assert entry["heartRateVariability"]["slopePer30d"] == pytest.approx(30, abs=0.5)

    def test_endpoint_windows_do_not_overlap_on_short_histories(self):
        """A 30-day history must not report delta 0 for a line that moved."""
        rows = [insights._flatten(r) for r in flat_days(
            30, steps=[1000 + i * 100 for i in range(30)]
        )]
        entry = {t["metric"]: t for t in insights.trends(rows)}["steps"]
        assert entry["firstMean"] < entry["lastMean"]
        assert entry["delta"] > 0

    def test_direction_respects_which_way_is_better(self):
        rows = [insights._flatten(r) for r in flat_days(
            30, heartRate=[80 - i for i in range(30)]
        )]
        entry = {t["metric"]: t for t in insights.trends(rows)}["heartRate"]
        assert entry["direction"] == "改善"

    def test_metrics_with_too_few_samples_are_skipped(self):
        rows = [insights._flatten(r) for r in flat_days(5, steps=1000)]
        assert insights.trends(rows) == []

    def test_gaps_do_not_compress_the_x_axis(self):
        """Ordinal dates, not indices: the same rise spread over a longer span
        is a gentler slope, and indices would score the two identically."""
        values = [1000 + i * 100 for i in range(10)]
        dense = [(f"2026-08-{1 + i:02d}", v) for i, v in enumerate(values)]
        # Same ten readings, but the last five sit a month later.
        gapped = dense[:5] + [
            (f"2026-09-{6 + i:02d}", v) for i, v in enumerate(values[5:])
        ]
        assert insights._slope_per_30d(gapped) < insights._slope_per_30d(dense)


class TestActivityShift:
    def test_compares_the_last_week_with_the_weeks_before_it(self):
        rows = [insights._flatten(r) for r in flat_days(
            20, steps=[10000] * 13 + [5000] * 7
        )]
        shift = insights.activity_shift(rows)
        assert shift["steps"]["recentMean"] == 5000
        assert shift["steps"]["priorMean"] == 10000
        assert shift["steps"]["changePct"] == -50.0

    def test_absent_when_there_is_not_enough_history(self):
        rows = [insights._flatten(r) for r in flat_days(6, steps=8000)]
        assert "steps" not in insights.activity_shift(rows)


class TestWindowContext:
    def test_unknown_metric_returns_nothing(self, db, user):
        assert insights.window_context(
            user["id"], "notAMetric", "2026-08-01", "2026-08-30"
        ) is None

    def test_empty_span_returns_nothing(self, db, user):
        assert insights.window_context(
            user["id"], "steps", "2020-01-01", "2020-01-31"
        ) is None


class TestBuildContext:
    def test_returns_nothing_without_data(self, db, user):
        assert insights.build_context(user["id"]) is None

    def test_defaults_to_the_newest_recorded_day(self, db, user, seed_health):
        seed_health([
            {"date": "2026-08-01", "steps": 5000},
            {"date": "2026-08-02", "steps": 6000},
        ])
        context = insights.build_context(user["id"])
        assert context["snapshotDate"] == "2026-08-02"

    def test_an_unknown_date_is_not_silently_replaced(self, db, user, seed_health):
        seed_health([{"date": "2026-08-01", "steps": 5000}])
        assert insights.build_context(user["id"], "2026-08-09") is None

    def test_stays_small_enough_to_prompt_with(self, db, user, seed_health):
        seed_health([
            {"date": f"2026-08-{d:02d}", "steps": 8000 + d, "heart_rate": 60,
             "hrv": 45, "sleep_duration": 7, "stress": 30}
            for d in range(1, 31)
        ])
        context = insights.build_context(user["id"])
        blob = json.dumps(context, ensure_ascii=False)
        # A month of history has to cost thousands of characters, not tens of
        # thousands — the whole point of computing features server-side.
        assert len(blob) < 20_000


# --- reply parsing ----------------------------------------------------------
GOOD_BRIEFING = {
    "status": "中等偏上",
    "headline": "恢复尚可，睡眠偏短。",
    "diagnosis": [{"title": "睡眠结构", "detail": "睡眠 6 小时，低于目标。"}],
    "shortfall": "睡眠不足",
    "prescription": {
        "intensity": "中等", "hrZone": "Zone 2~Zone 3",
        "suggestion": "40 分钟慢跑", "durationMin": 40, "avoid": "高强度间歇",
    },
    "actions": ["提前 30 分钟入睡", "午后避免咖啡因"],
}


class TestParseBriefing:
    def test_plain_json(self):
        out = coach.parse_briefing(json.dumps(GOOD_BRIEFING, ensure_ascii=False))
        assert out["status"] == "中等偏上"
        assert out["prescription"]["durationMin"] == 40
        assert out["actions"] == ["提前 30 分钟入睡", "午后避免咖啡因"]

    def test_answer_is_taken_from_after_a_reasoning_preamble(self):
        """The gateway's primary upstream narrates its thinking first."""
        reply = (
            'The user wants a briefing. Let me consider {"draft": true} first.\n'
            "Actually I should output the final object now:\n"
            + json.dumps(GOOD_BRIEFING, ensure_ascii=False)
        )
        assert coach.parse_briefing(reply)["status"] == "中等偏上"

    def test_markdown_fences_are_tolerated(self):
        reply = "```json\n" + json.dumps(GOOD_BRIEFING, ensure_ascii=False) + "\n```"
        assert coach.parse_briefing(reply)["headline"] == "恢复尚可，睡眠偏短。"

    def test_missing_prescription_does_not_raise(self):
        payload = {k: v for k, v in GOOD_BRIEFING.items() if k != "prescription"}
        out = coach.parse_briefing(json.dumps(payload, ensure_ascii=False))
        assert out["prescription"]["suggestion"] is None

    def test_non_numeric_duration_becomes_none(self):
        payload = json.loads(json.dumps(GOOD_BRIEFING))
        payload["prescription"]["durationMin"] = "四十分钟"
        assert coach.parse_briefing(json.dumps(payload))["prescription"]["durationMin"] is None

    def test_diagnosis_written_as_plain_strings_is_accepted(self):
        payload = json.loads(json.dumps(GOOD_BRIEFING))
        payload["diagnosis"] = ["睡眠偏短。"]
        out = coach.parse_briefing(json.dumps(payload, ensure_ascii=False))
        assert out["diagnosis"][0]["detail"] == "睡眠偏短。"

    def test_an_empty_briefing_is_rejected_rather_than_rendered_blank(self):
        with pytest.raises(ai_svc.AIError):
            coach.parse_briefing(json.dumps({"status": "好"}))

    def test_prose_without_json_raises(self):
        with pytest.raises(ai_svc.AIError):
            coach.parse_briefing("今天状态不错，可以正常训练。")


class TestParseTrendInsight:
    def test_valid_reply(self):
        reply = json.dumps({
            "summary": "HRV 稳步上升。",
            "drivers": [{"factor": "有氧负荷", "detail": "区间内 8 次有氧。"}],
            "caution": None, "confidence": "high",
        }, ensure_ascii=False)
        out = coach.parse_trend_insight(reply)
        assert out["confidence"] == "high"
        assert out["drivers"][0]["factor"] == "有氧负荷"

    def test_unknown_confidence_falls_back_to_medium(self):
        reply = json.dumps({"summary": "上升。", "confidence": "很高"}, ensure_ascii=False)
        assert coach.parse_trend_insight(reply)["confidence"] == "medium"

    def test_empty_reply_raises(self):
        with pytest.raises(ai_svc.AIError):
            coach.parse_trend_insight(json.dumps({"confidence": "high"}))


class TestExtractJson:
    def test_last_object_wins_over_an_earlier_draft(self):
        assert ai_svc.extract_json('{"a": 1} then {"a": 2}') == {"a": 2}

    def test_braces_inside_strings_do_not_break_the_scan(self):
        assert ai_svc.extract_json('思考 } 中。{"t": "含 } 的文本"}')["t"] == "含 } 的文本"

    def test_escaped_quote_inside_a_string(self):
        assert ai_svc.extract_json(r'x {"t": "a \" b"}')["t"] == 'a " b'

    def test_arrays_are_extracted_too(self):
        assert ai_svc.extract_json("preamble [1, 2, 3]") == [1, 2, 3]

    def test_empty_reply_raises(self):
        with pytest.raises(ai_svc.AIError):
            ai_svc.extract_json("   ")


# --- prompt assembly --------------------------------------------------------
class TestPrompts:
    def test_system_prompt_forbids_inventing_numbers(self):
        assert "禁止编造" in coach.SYSTEM

    def test_system_prompt_disclaims_medical_diagnosis(self):
        assert "不做医疗诊断" in coach.SYSTEM

    def test_model_is_told_not_to_recompute_the_z_scores(self):
        assert "不要自行重算" in coach.SYSTEM

    def test_briefing_prompt_carries_the_context_as_json(self):
        messages = coach.briefing_messages({"snapshotDate": "2026-08-01", "x": 1})
        assert messages[0]["role"] == "system"
        assert '"snapshotDate":"2026-08-01"' in messages[1]["content"]

    def test_context_is_not_ascii_escaped(self):
        """Escaping Chinese to \\uXXXX roughly triples its token cost."""
        assert "睡眠" in coach._payload({"label": "睡眠"})

    def test_copilot_keeps_the_context_out_of_the_visible_transcript(self):
        messages = coach.copilot_messages(
            {"snapshotDate": "2026-08-01"}, [], "我今天能练吗？"
        )
        assert [m["role"] for m in messages] == ["system", "system", "user"]
        assert messages[-1]["content"] == "我今天能练吗？"

    def test_copilot_history_is_capped_and_role_filtered(self):
        history = [{"role": "user", "content": f"q{i}"} for i in range(20)]
        history.append({"role": "tool", "content": "ignored"})
        messages = coach.copilot_messages({}, history, "最后一问")
        turns = [m for m in messages if m["role"] != "system"]
        assert len(turns) == 9  # eight remembered turns plus the new question
        assert "ignored" not in json.dumps(messages, ensure_ascii=False)

    def test_copilot_asks_for_markdown_not_json(self):
        assert "Markdown" in coach.COPILOT_SYSTEM
        assert "最后出现的 JSON" not in coach.COPILOT_SYSTEM


# --- rule-based counterparts ------------------------------------------------
def context_with(**today):
    base = {
        "snapshotDate": "2026-08-30",
        "todayMetrics": {
            "sleep": None,
            "autonomicNervous": {},
            "recovery": {},
            "activityToday": {},
        },
        "deviations": [],
        "trends": [],
        "activityShift": {},
    }
    base["todayMetrics"].update(today)
    return base


class TestRuleBriefing:
    def test_readiness_drives_the_prescription(self):
        low = coach.rule_briefing(context_with(recovery={"trainingReadiness": 30}))
        high = coach.rule_briefing(context_with(recovery={"trainingReadiness": 85}))
        assert low["prescription"]["intensity"] == "低"
        assert high["prescription"]["intensity"] == "高"

    def test_short_sleep_is_named_as_the_shortfall(self):
        out = coach.rule_briefing(context_with(
            sleep={"durationHours": 5.0, "targetHours": 7.0}
        ))
        assert "睡眠" in out["shortfall"]

    def test_no_shortfall_is_stated_explicitly(self):
        out = coach.rule_briefing(context_with(
            sleep={"durationHours": 8.0, "targetHours": 7.0}
        ))
        assert out["shortfall"] == "无明显短板"

    def test_it_never_invents_a_metric_the_watch_did_not_record(self):
        out = coach.rule_briefing(context_with())
        assert out["diagnosis"] == []
        assert out["actions"]

    def test_the_headline_names_the_largest_departure(self):
        context = context_with(autonomicNervous={"restingHr": 80})
        context["deviations"] = [{
            "metric": "heartRate", "label": "静息心率", "unit": "bpm",
            "value": 80, "baselineMean": 60, "sd": 5, "baselineDays": 28,
            "z": 4.0, "verdict": "偏差",
        }]
        assert "静息心率" in coach.rule_briefing(context)["headline"]

    def test_a_sustained_drop_in_steps_becomes_an_action(self):
        context = context_with()
        context["activityShift"] = {
            "steps": {"label": "步数", "recentMean": 4000,
                      "priorMean": 10000, "changePct": -60.0}
        }
        assert any("60" in a for a in coach.rule_briefing(context)["actions"])


# --- orchestration ----------------------------------------------------------
@pytest.fixture
def gateway(monkeypatch):
    monkeypatch.setenv("AI_GATEWAY_TOKEN", "test-token")
    monkeypatch.setenv("AI_GATEWAY_BASE_URL", "http://gateway.test/v1")
    monkeypatch.setenv("AI_MODEL_CHAIN", "gateway")


@pytest.fixture
def month(db, user, seed_health):
    seed_health([
        {"date": f"2026-08-{d:02d}", "steps": 8000, "heart_rate": 60, "hrv": 45,
         "sleep_duration": 7, "sleep_quality": 80, "stress": 30}
        for d in range(1, 31)
    ])
    return user


def answer(monkeypatch, text):
    """Make every model reply with `text`, and count the calls."""
    calls = []

    def fake_chat(self, messages, timeout=None, max_tokens=None):
        calls.append(messages)
        return ai_svc.Completion(text, "nvidia")

    monkeypatch.setattr(ai_svc.Provider, "chat", fake_chat, raising=False)
    monkeypatch.setattr(ai_svc.OpenAICompatProvider, "chat", fake_chat)
    return calls


class TestGetBriefing:
    def test_no_data_says_so_rather_than_guessing(self, db, user, gateway):
        out = analysis_svc.get_briefing(user["id"])
        assert out["meta"]["source"] == "none"
        assert out["briefing"] is None

    def test_blocking_mode_returns_the_model_answer(self, month, gateway, monkeypatch):
        answer(monkeypatch, json.dumps(GOOD_BRIEFING, ensure_ascii=False))
        out = analysis_svc.get_briefing(month["id"], wait=True)
        assert out["meta"]["source"] == "ai"
        assert out["briefing"]["status"] == "中等偏上"

    def test_a_stored_answer_is_reused(self, month, gateway, monkeypatch):
        calls = answer(monkeypatch, json.dumps(GOOD_BRIEFING, ensure_ascii=False))
        analysis_svc.get_briefing(month["id"], wait=True)
        out = analysis_svc.get_briefing(month["id"])
        assert out["meta"]["cached"] is True
        assert len(calls) == 1, "the cached answer must not trigger a second call"

    def test_new_health_data_expires_the_stored_answer(
        self, month, gateway, monkeypatch, seed_health
    ):
        answer(monkeypatch, json.dumps(GOOD_BRIEFING, ensure_ascii=False))
        analysis_svc.get_briefing(month["id"], wait=True)
        seed_health([{"date": "2026-08-31", "steps": 12000}])
        out = analysis_svc.get_briefing(month["id"])
        assert out["meta"].get("cached") is not True

    def test_a_failing_model_degrades_to_the_rule_engine(
        self, month, gateway, monkeypatch
    ):
        def boom(self, messages, timeout=None, max_tokens=None):
            raise ai_svc.AIError("upstream down")

        monkeypatch.setattr(ai_svc.OpenAICompatProvider, "chat", boom)
        out = analysis_svc.get_briefing(month["id"], wait=True)
        assert out["meta"]["source"] == "rules"
        assert out["briefing"] is not None

    def test_the_non_blocking_path_answers_without_calling_a_model(
        self, month, gateway, monkeypatch
    ):
        calls = answer(monkeypatch, json.dumps(GOOD_BRIEFING, ensure_ascii=False))
        monkeypatch.setattr(analysis_svc, "_run_in_background", lambda key, fn: True)
        out = analysis_svc.get_briefing(month["id"])
        assert out["meta"]["pending"] is True
        assert out["briefing"]["status"]
        assert calls == []

    def test_one_generation_per_key_no_matter_how_often_it_is_polled(self):
        """The poll runs every few seconds; a generation takes minutes."""
        started = []
        # A job that never finishes, so the key stays claimed across polls.
        blocked = analysis_svc.threading.Event()
        analysis_svc._run_in_background("test-key", lambda: (
            started.append(1), blocked.wait(5)
        ))
        try:
            for _ in range(5):
                analysis_svc._run_in_background("test-key", lambda: started.append(1))
            assert len(started) == 1
        finally:
            blocked.set()


class TestGetTrendInsight:
    def test_empty_span_is_reported_not_analysed(self, month, gateway):
        out = analysis_svc.get_trend_insight(
            month["id"], "steps", "2020-01-01", "2020-01-31"
        )
        assert out["meta"]["source"] == "none"

    def test_model_answer_is_returned_and_cached(self, month, gateway, monkeypatch):
        reply = json.dumps({
            "summary": "步数稳定。", "drivers": [], "caution": None,
            "confidence": "medium",
        }, ensure_ascii=False)
        calls = answer(monkeypatch, reply)
        first = analysis_svc.get_trend_insight(
            month["id"], "steps", "2026-08-01", "2026-08-30"
        )
        second = analysis_svc.get_trend_insight(
            month["id"], "steps", "2026-08-01", "2026-08-30"
        )
        assert first["insight"]["summary"] == "步数稳定。"
        assert second["meta"]["cached"] is True
        assert len(calls) == 1

    def test_a_failing_model_degrades_to_the_rule_engine(
        self, month, gateway, monkeypatch
    ):
        def boom(self, messages, timeout=None, max_tokens=None):
            raise ai_svc.AIError("down")

        monkeypatch.setattr(ai_svc.OpenAICompatProvider, "chat", boom)
        out = analysis_svc.get_trend_insight(
            month["id"], "steps", "2026-08-01", "2026-08-30"
        )
        assert out["meta"]["source"] == "rules"
        assert out["insight"]["summary"]


class TestStreamChat:
    """The gateway's streaming path is measurably less reliable than its
    blocking one, so a stream that produces nothing must not end the attempt."""

    def test_deltas_are_forwarded(self, gateway, monkeypatch):
        def fake_stream(self, messages, timeout=None, max_tokens=None):
            yield ai_svc.Completion("你好", "nvidia")
            yield ai_svc.Completion("，世界", "nvidia")

        monkeypatch.setattr(ai_svc.OpenAICompatProvider, "stream", fake_stream)
        text = "".join(d.text for d in ai_svc.stream_chat([{"role": "user", "content": "hi"}]))
        assert text == "你好，世界"

    def test_a_failed_stream_retries_the_same_model_without_streaming(
        self, gateway, monkeypatch
    ):
        def fake_stream(self, messages, timeout=None, max_tokens=None):
            raise ai_svc.AIError("所有模型均不可用")
            yield  # pragma: no cover - generator marker

        monkeypatch.setattr(ai_svc.OpenAICompatProvider, "stream", fake_stream)
        answer(monkeypatch, "完整回答")
        deltas = list(ai_svc.stream_chat([{"role": "user", "content": "hi"}]))
        assert "".join(d.text for d in deltas) == "完整回答"

    def test_no_model_switch_once_text_has_been_sent(self, gateway, monkeypatch):
        def fake_stream(self, messages, timeout=None, max_tokens=None):
            yield ai_svc.Completion("半句", "nvidia")
            raise ai_svc.AIError("断流")

        monkeypatch.setattr(ai_svc.OpenAICompatProvider, "stream", fake_stream)
        with pytest.raises(ai_svc.AIError):
            list(ai_svc.stream_chat([{"role": "user", "content": "hi"}]))


class TestCopilotStream:
    def test_no_data_yields_an_error_event(self, db, user, gateway):
        events = list(analysis_svc.copilot_stream(user["id"], "我今天能练吗"))
        assert events[0][0] == "error"

    def test_a_successful_answer_is_framed_start_delta_done(
        self, month, gateway, monkeypatch
    ):
        def fake_stream(self, messages, timeout=None, max_tokens=None):
            yield ai_svc.Completion("可以。", "nvidia")

        monkeypatch.setattr(ai_svc.OpenAICompatProvider, "stream", fake_stream)
        events = list(analysis_svc.copilot_stream(month["id"], "我今天能练吗"))
        assert [e for e, _ in events] == ["start", "delta", "done"]
        assert events[-1][1]["upstream"] == "nvidia"


# --- endpoints --------------------------------------------------------------
class TestEndpoints:
    def test_briefing_requires_auth(self, client):
        assert client.get("/api/analysis/briefing").status_code == 401

    def test_trend_insight_requires_auth(self, client):
        assert client.get("/api/analysis/trend-insight").status_code == 401

    def test_copilot_requires_auth(self, client):
        assert client.post("/api/analysis/copilot", json={}).status_code == 401

    def test_briefing_answers_even_with_no_model_configured(self, client, auth, month):
        resp = client.get("/api/analysis/briefing", headers=auth)
        assert resp.status_code == 200
        assert resp.get_json()["briefing"] is not None

    def test_trend_insight_rejects_an_unknown_metric(self, client, auth, month):
        resp = client.get(
            "/api/analysis/trend-insight",
            query_string={"metric": "nope", "startDate": "2026-08-01",
                          "endDate": "2026-08-30"},
            headers=auth,
        )
        assert resp.status_code == 400
        assert "supported" in resp.get_json()

    def test_trend_insight_requires_a_range(self, client, auth, month):
        resp = client.get(
            "/api/analysis/trend-insight",
            query_string={"metric": "steps"}, headers=auth,
        )
        assert resp.status_code == 400

    def test_copilot_requires_a_question(self, client, auth, month):
        resp = client.post("/api/analysis/copilot", json={}, headers=auth)
        assert resp.status_code == 400

    def test_copilot_streams_server_sent_events(
        self, client, auth, month, gateway, monkeypatch
    ):
        def fake_stream(self, messages, timeout=None, max_tokens=None):
            yield ai_svc.Completion("可以，注意强度。", "nvidia")

        monkeypatch.setattr(ai_svc.OpenAICompatProvider, "stream", fake_stream)
        resp = client.post(
            "/api/analysis/copilot",
            json={"question": "我今天能练吗"}, headers=auth,
        )
        assert resp.status_code == 200
        assert resp.mimetype == "text/event-stream"
        body = resp.get_data(as_text=True)
        assert "event: delta" in body
        assert "可以，注意强度。" in body

    def test_briefing_never_leaks_the_gateway_token(
        self, client, auth, month, gateway, monkeypatch
    ):
        # No real background generation: this suite must not reach the network.
        monkeypatch.setattr(analysis_svc, "_run_in_background", lambda key, fn: True)
        body = client.get("/api/analysis/briefing", headers=auth).get_data(as_text=True)
        assert "test-token" not in body


class TestStreamRetryScope:
    """The blind non-streaming retry is only worth doing for endpoints that
    actually have a separate streaming transport."""

    def test_a_provider_without_streaming_is_not_called_twice(
        self, monkeypatch
    ):
        monkeypatch.setenv("GEMINI_API_KEY", "k")
        monkeypatch.setenv("AI_MODEL_CHAIN", "gemini-flash")
        calls = []

        def boom(self, messages, timeout=None, max_tokens=None):
            calls.append(1)
            raise ai_svc.AIError("down")

        monkeypatch.setattr(ai_svc.GeminiProvider, "chat", boom)
        with pytest.raises(ai_svc.AIError):
            list(ai_svc.stream_chat([{"role": "user", "content": "hi"}]))
        assert len(calls) == 1

    def test_the_gateway_declares_a_streaming_transport(self):
        assert ai_svc.CATALOG["gateway"].streaming is True
        assert ai_svc.CATALOG["gemini-flash"].streaming is False


class TestRegenerate:
    def test_refresh_evicts_the_stored_answer_so_the_poll_can_see_the_new_one(
        self, month, gateway, monkeypatch
    ):
        """Without eviction the poll after 重新生成 reads the row it was asked
        to replace, reports `cached`, and stops — leaving the old text on
        screen."""
        answer(monkeypatch, json.dumps(GOOD_BRIEFING, ensure_ascii=False))
        analysis_svc.get_briefing(month["id"], wait=True)
        assert analysis_svc.get_briefing(month["id"])["meta"]["cached"] is True

        monkeypatch.setattr(analysis_svc, "_run_in_background", lambda key, fn: True)
        analysis_svc.get_briefing(month["id"], refresh=True)

        after = analysis_svc.get_briefing(month["id"])
        assert after["meta"].get("cached") is not True
        assert after["meta"]["pending"] is True