"""HTTP-side governance of the Intelligence Center recommendation lifecycle.

Two findings are pinned here:

* the four HTTP mutations wrote no platform audit row at all, while their
  MCP twins did (``app/mcp/runtime.py``) and 17+ peer domain routers do —
  so "who forced a billed brief" and "who fanned a model-written card out
  to every IT admin" were answerable only from the application log;
* ``transition_recommendation`` accepted any status from any state, so a
  booked outcome could be rewound and a dismissed suggestion could claim a
  saving.
"""
import uuid
from decimal import Decimal

import pytest
from sqlalchemy import select

from app.audit.models import AuditLog
from app.integrations.models import PlatformSetting
from app.intelligence.models import IntelligenceRecommendation, IntelligenceSnapshot
from app.scheduled_jobs.intelligence_handlers import run_intelligence_snapshot_daily


async def _enable(db_session) -> None:
    row = await db_session.get(PlatformSetting, "intelligence.enabled")
    if row is None:
        db_session.add(PlatformSetting(key="intelligence.enabled", value=True))
    else:
        row.value = True
    await db_session.commit()


async def _recommendation(db_session, *, status: str = "new", **overrides):
    await run_intelligence_snapshot_daily(db_session, {})
    snapshot = (await db_session.execute(
        select(IntelligenceSnapshot)
        .where(IntelligenceSnapshot.subject_type == "platform")
        .order_by(IntelligenceSnapshot.created_at.desc())
        .limit(1)
    )).scalars().first()
    row = IntelligenceRecommendation(
        snapshot_id=snapshot.id,
        subject_type=snapshot.subject_type,
        subject_id=snapshot.subject_id,
        audience_key=snapshot.audience_key,
        dedupe_key=f"governance-{uuid.uuid4().hex}",
        domain="reclaim",
        severity="high",
        priority=1,
        title_zh="回收闲置资源",
        title_en="Reclaim idle resource",
        summary_zh="",
        summary_en="",
        evidence_refs=[],
        action_contract={
            "contract_version": 2,
            "action_kind": "reclaim",
            "workbench_path": "/admin/assets",
            "requires_approval": False,
            "fact_refs": ["/facts/assets/idle_surplus_count"],
            "verification": {"baseline_facts_hash": snapshot.facts_hash},
        },
        status=status,
        **overrides,
    )
    db_session.add(row)
    await db_session.commit()
    await db_session.refresh(row)
    return row


async def _audit_rows(db_session, action: str) -> list[AuditLog]:
    return list((await db_session.execute(
        select(AuditLog).where(AuditLog.action == action)
    )).scalars().all())


@pytest.mark.asyncio
async def test_recommendation_update_writes_an_audit_row(
    client, db_session, admin_user,
):
    await _enable(db_session)
    row = await _recommendation(db_session)
    user, token = admin_user

    response = await client.patch(
        f"/api/v1/intelligence/recommendations/{row.id}",
        json={"status": "viewed"},
        headers={"Authorization": f"Bearer {token}"},
    )
    assert response.status_code == 200, response.text

    entries = await _audit_rows(db_session, "intelligence.recommendation.update")
    assert len(entries) == 1
    entry = entries[0]
    assert entry.actor_user_id == user.id
    assert entry.actor_type == "user"
    assert entry.status == "success"
    assert entry.resource_type == "intelligence.recommendation"
    assert entry.resource_id == str(row.id)
    assert entry.trace_id is not None
    assert entry.detail["from_status"] == "new"
    assert entry.detail["to_status"] == "viewed"


@pytest.mark.asyncio
async def test_recommendation_send_audits_who_fanned_the_card_out(
    client, db_session, admin_user,
):
    await _enable(db_session)
    row = await _recommendation(db_session)
    user, token = admin_user

    response = await client.post(
        f"/api/v1/intelligence/recommendations/{row.id}/send",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert response.status_code == 200, response.text
    queued = response.json()["data"]["queued"]

    entries = await _audit_rows(db_session, "intelligence.recommendation.send")
    assert len(entries) == 1
    assert entries[0].actor_user_id == user.id
    assert entries[0].resource_id == str(row.id)
    assert entries[0].detail["queued"] == queued
    assert entries[0].detail["channel_fanout"] == "admin_broadcast"


@pytest.mark.asyncio
async def test_recommendation_execute_audits_the_handoff(
    client, db_session, admin_user,
):
    await _enable(db_session)
    row = await _recommendation(db_session)
    user, token = admin_user

    response = await client.post(
        f"/api/v1/intelligence/recommendations/{row.id}/execute",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert response.status_code == 200, response.text

    entries = await _audit_rows(db_session, "intelligence.recommendation.execute")
    assert len(entries) == 1
    assert entries[0].actor_user_id == user.id
    assert entries[0].detail["action_kind"] == "reclaim"
    assert entries[0].detail["from_status"] == "new"
    assert entries[0].detail["to_status"] == "in_progress"
    assert entries[0].detail["handoff_only"] is True


@pytest.mark.asyncio
async def test_forced_brief_generation_is_audited_with_the_force_flag(
    client, db_session, admin_user, monkeypatch,
):
    """Forcing bypasses the cache and bills a model call — audit says who."""
    from app.intelligence import router as intelligence_router

    await _enable(db_session)
    await run_intelligence_snapshot_daily(db_session, {})
    user, token = admin_user

    class _Result:
        cache_reused = False

        def __init__(self, analysis):
            self.analysis = analysis

    captured = {}

    async def fake_generate(db, *, snapshot, analysis_kind, actor_id, force=False):
        captured["force"] = force
        from datetime import UTC, datetime

        from app.intelligence.models import IntelligenceAnalysis

        analysis = IntelligenceAnalysis(
            snapshot_id=snapshot.id,
            subject_type=snapshot.subject_type,
            subject_id=snapshot.subject_id,
            audience_key=snapshot.audience_key,
            analysis_kind=analysis_kind,
            facts_hash=snapshot.facts_hash,
            generation_key=uuid.uuid4().hex,
            status="succeeded",
            summary_zh="结论",
            summary_en="Conclusion",
            recommendations=[],
            generated_by_user_id=actor_id,
            data_as_of=snapshot.data_as_of,
            generated_at=datetime.now(UTC),
            forced=force,
            input_token_count=11,
            output_token_count=22,
        )
        db.add(analysis)
        await db.commit()
        await db.refresh(analysis)
        return _Result(analysis)

    monkeypatch.setattr(intelligence_router, "generate_analysis", fake_generate)
    response = await client.post(
        "/api/v1/intelligence/analyses/generate",
        json={"force": True, "confirm_force": True},
        headers={"Authorization": f"Bearer {token}"},
    )
    assert response.status_code == 200, response.text
    assert captured["force"] is True

    entries = await _audit_rows(db_session, "intelligence.analysis.generate")
    assert len(entries) == 1
    assert entries[0].actor_user_id == user.id
    assert entries[0].resource_type == "intelligence.analysis"
    assert entries[0].detail["force"] is True
    assert entries[0].detail["cache_reused"] is False
    assert entries[0].detail["model_calls"] == 1
    assert entries[0].detail["input_token_count"] == 11


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("current", "target"),
    [
        ("resolved", "new"),
        ("resolved", "viewed"),
        ("ignored", "resolved"),
        ("new", "resolved"),
    ],
)
async def test_illegal_lifecycle_moves_are_refused(
    client, db_session, admin_user, current, target,
):
    await _enable(db_session)
    extra = {"ignored_reason": "duplicate"} if current == "ignored" else {}
    row = await _recommendation(db_session, status=current, **extra)
    _, token = admin_user

    response = await client.patch(
        f"/api/v1/intelligence/recommendations/{row.id}",
        json={"status": target},
        headers={"Authorization": f"Bearer {token}"},
    )
    assert response.status_code == 422, response.text
    assert response.json()["errors"][0]["code"] == "INTELLIGENCE_RECOMMENDATION_TRANSITION"
    await db_session.refresh(row)
    assert row.status == current


@pytest.mark.asyncio
async def test_shipped_flow_still_works_and_resolved_at_is_written_once(
    client, db_session, admin_user,
):
    """new → viewed → in_progress → resolved, then a savings correction."""
    await _enable(db_session)
    row = await _recommendation(db_session)
    _, token = admin_user
    headers = {"Authorization": f"Bearer {token}"}

    for status in ("viewed", "in_progress", "resolved"):
        response = await client.patch(
            f"/api/v1/intelligence/recommendations/{row.id}",
            json={"status": status},
            headers=headers,
        )
        assert response.status_code == 200, response.text
    await db_session.refresh(row)
    first_resolved_at = row.resolved_at
    assert first_resolved_at is not None

    correction = await client.patch(
        f"/api/v1/intelligence/recommendations/{row.id}",
        json={"status": "resolved", "realized_savings": "120.50", "currency": "cny"},
        headers=headers,
    )
    assert correction.status_code == 200, correction.text
    await db_session.refresh(row)
    assert row.resolved_at == first_resolved_at
    assert row.realized_savings == Decimal("120.50")
    assert row.currency == "CNY"


def test_resolved_at_survives_a_later_savings_correction():
    """The audited outcome date is written once, not on every re-resolve.

    The HTTP round-trip above cannot see this on its own — both PATCHes
    land inside the same stored second — so pin the pure transition.
    """
    from datetime import UTC, datetime

    from app.intelligence.recommendations import transition_recommendation

    booked = datetime(2026, 1, 4, 9, 30, tzinfo=UTC)
    row = IntelligenceRecommendation(status="resolved", resolved_at=booked)
    transition_recommendation(
        row, status="resolved", ignored_reason=None, actor_id=uuid.uuid4(),
        confirmed_savings=Decimal("42"),
    )
    assert row.resolved_at == booked
    assert row.confirmed_savings == Decimal("42")


@pytest.mark.asyncio
async def test_leaving_ignored_clears_the_stale_dismissal_reason(
    client, db_session, admin_user,
):
    await _enable(db_session)
    row = await _recommendation(
        db_session, status="ignored", ignored_reason="handled elsewhere",
    )
    _, token = admin_user

    response = await client.patch(
        f"/api/v1/intelligence/recommendations/{row.id}",
        json={"status": "in_progress"},
        headers={"Authorization": f"Bearer {token}"},
    )
    assert response.status_code == 200, response.text
    await db_session.refresh(row)
    assert row.status == "in_progress"
    assert row.ignored_reason is None


def test_push_copy_does_not_instruct_a_transition_the_state_machine_rejects():
    """The card deep-links a ``new`` row; ``new -> resolved`` now 422s.

    The push told every IT admin to "mark it resolved or ignored", but the
    row the card opens is almost always ``new`` and the only moves offered
    there are view / start handling / ignore. Following the instruction
    literally produced an error the reader could not explain.
    """
    from app.intelligence.recommendations import _ALLOWED_TRANSITIONS
    from app.notifications.templates.platform import TEMPLATES

    assert "resolved" not in _ALLOWED_TRANSITIONS["new"]
    template = TEMPLATES["intelligence.recommendation"]
    for field, phrase in (
        ("body_text", "开始处理"),
        ("body_text_en", "start handling"),
        ("body_text_ja", "対応を開始"),
    ):
        body = template[field]
        assert phrase in body, field
        # Still names both terminal outcomes for the adoption-rate metric —
        # it just no longer presents "resolved" as the immediate next tap.
        assert "采纳率" in body or "adoption-rate" in body or "採用率" in body