"""RC-BE-004 gateway tests for /api/report-collaboration."""

from __future__ import annotations

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient

from app.gateway.routers import report_collaboration
from app.report_collaboration.execution.live_hub import PublishingReportCollaborationStore, ReportCollaborationLiveHub, SseStreamSettings
from deerflow.config.model_config import ModelConfig
from deerflow.config.report_collaboration_config import ReportCollaborationConfig
from deerflow.persistence.report_collaboration import MemoryReportCollaborationStore


class _Cfg:
    report_collaboration = ReportCollaborationConfig(enabled=True)
    models = [ModelConfig(name="demo-chat", use="langchain_openai:ChatOpenAI", model="demo")]

    def get_model_config(self, name: str):
        return next((item for item in self.models if item.name == name), None)


class _DisabledCfg:
    report_collaboration = ReportCollaborationConfig(enabled=False)


def _client(monkeypatch: pytest.MonkeyPatch, *, enabled: bool = True, user: str | None = "user_a") -> tuple[TestClient, MemoryReportCollaborationStore]:
    monkeypatch.setattr(report_collaboration, "get_app_config", lambda: _Cfg() if enabled else _DisabledCfg())
    app = FastAPI()
    hub = ReportCollaborationLiveHub()
    inner = MemoryReportCollaborationStore()
    store = PublishingReportCollaborationStore(inner, hub)
    app.state.report_collaboration_store = store
    app.state.report_collaboration_live_hub = hub
    app.state.report_collaboration_sse_settings = SseStreamSettings(
        deadline_seconds=0.6,
        live_wait_seconds=0.05,
        db_poll_seconds=0.2,
        heartbeat_seconds=30,
        flush_padding="",
    )
    app.include_router(report_collaboration.router)

    async def _fake_user(_request=None):
        return user

    monkeypatch.setattr(report_collaboration, "get_current_user", _fake_user)
    return TestClient(app), store


def test_disabled_flag_returns_503(monkeypatch: pytest.MonkeyPatch) -> None:
    client, _ = _client(monkeypatch, enabled=False)
    res = client.get("/api/report-collaboration/sessions")
    assert res.status_code == 503
    assert res.json()["code"] == "REPORT_COLLABORATION_DISABLED"


def test_unauthenticated_returns_401(monkeypatch: pytest.MonkeyPatch) -> None:
    client, _ = _client(monkeypatch, user=None)
    res = client.get("/api/report-collaboration/sessions")
    assert res.status_code == 401


def test_session_message_plan_run_http(monkeypatch: pytest.MonkeyPatch) -> None:
    client, store = _client(monkeypatch)

    missing_key = client.post("/api/report-collaboration/sessions", json={"title": "新能源"})
    assert missing_key.status_code == 422

    created = client.post(
        "/api/report-collaboration/sessions",
        json={"title": "新能源"},
        headers={"X-Idempotency-Key": "k-create"},
    )
    assert created.status_code == 200, created.text
    session = created.json()
    session_id = session["id"]
    assert session["status"] == "empty"

    replay = client.post(
        "/api/report-collaboration/sessions",
        json={"title": "other"},
        headers={"X-Idempotency-Key": "k-create"},
    )
    assert replay.json()["id"] == session_id

    listed = client.get("/api/report-collaboration/sessions")
    assert listed.status_code == 200
    assert listed.json()["sessions"][0]["id"] == session_id

    sent = client.post(
        f"/api/report-collaboration/sessions/{session_id}/messages",
        json={"text": "写一份新能源汽车趋势报告"},
        headers={"X-Idempotency-Key": "k-msg"},
    )
    assert sent.status_code == 200, sent.text
    assert sent.json()["role"] == "human"
    assert sent.json()["metadata"]["idempotency_key"] == "k-msg"

    snapshot = client.get(f"/api/report-collaboration/sessions/{session_id}")
    assert snapshot.status_code == 200
    body = snapshot.json()
    assert body["session"]["status"] == "clarifying"
    assert body["session"]["requirement_revision"] == 1
    assert body["messages"][0]["content"].startswith("写一份")
    assert any(item["role"] == "ai" and "读者" in item["content"] for item in body["messages"])
    assert body["last_seq"] == 0
    revision = body["session"]["requirement_revision"]

    stale = client.post(
        f"/api/report-collaboration/sessions/{session_id}/plan-requests",
        headers={"X-Idempotency-Key": "k-plan-stale", "X-Expected-Revision": "9"},
    )
    assert stale.status_code == 409
    assert stale.json()["code"] == "STALE_REVISION"

    queued = client.post(
        f"/api/report-collaboration/sessions/{session_id}/plan-requests",
        headers={"X-Idempotency-Key": "k-plan", "X-Expected-Revision": str(revision)},
    )
    assert queued.status_code == 204

    proposed = client.get(f"/api/report-collaboration/sessions/{session_id}")
    assert proposed.status_code == 200
    proposed_body = proposed.json()
    assert proposed_body["session"]["status"] == "proposal_ready"
    plans = proposed_body["plans"]
    assert {item["strategy"] for item in plans} == {"balanced_review", "parallel_depth", "focused_fast"}
    plan = next(item for item in plans if item.get("recommended"))
    assert plan["validation"]["ok"] is True
    assert all(node.get("output_artifact_type") and node.get("acceptance_criteria") for node in plan["nodes"])

    listed = client.get(f"/api/report-collaboration/sessions/{session_id}/plans")
    assert listed.status_code == 200
    assert len(listed.json()["plans"]) == 3

    selected = client.post(
        f"/api/report-collaboration/sessions/{session_id}/plans/{plan['id']}/select",
        headers={"X-Idempotency-Key": "k-select", "X-Expected-Revision": str(revision)},
    )
    assert selected.status_code == 200, selected.text
    assert selected.json()["selected_plan_id"] == plan["id"]
    assert client.get(f"/api/report-collaboration/sessions/{session_id}").json()["run"] is None

    started = client.post(
        f"/api/report-collaboration/sessions/{session_id}/runs",
        json={"plan_id": plan["id"]},
        headers={"X-Idempotency-Key": "k-run", "X-Expected-Revision": str(revision)},
    )
    assert started.status_code == 200, started.text
    run = started.json()
    assert run["status"] == "queued"
    run_id = run["id"]

    conflict = client.post(
        f"/api/report-collaboration/sessions/{session_id}/runs",
        json={"plan_id": plan["id"]},
        headers={"X-Idempotency-Key": "k-run-2", "X-Expected-Revision": str(revision)},
    )
    assert conflict.status_code == 409

    got = client.get(f"/api/report-collaboration/runs/{run_id}")
    assert got.status_code == 200
    assert got.json()["id"] == run_id

    events = client.get(f"/api/report-collaboration/runs/{run_id}/events?after_seq=0")
    assert events.status_code == 200
    recorded = events.json()["events"]
    assert len(recorded) == 1
    assert recorded[0]["type"] == "heartbeat"
    assert recorded[0]["data"]["audit"] == "model_snapshot"
    assert "api_key" not in str(recorded[0]["data"])
    assert "sk-" not in str(recorded[0]["data"])

    stream = client.stream("GET", f"/api/report-collaboration/runs/{run_id}/stream")
    with stream as response:
        assert response.status_code == 200, response.text
        assert response.headers["content-type"].startswith("text/event-stream")
        first = next(response.iter_text())
        assert "stream-ready" in first or first.startswith(":") or "data:" in first

    still_queued = client.get(f"/api/report-collaboration/runs/{run_id}")
    assert still_queued.status_code == 200
    assert still_queued.json()["status"] == "queued"
    assert still_queued.json()["cancel_requested"] is False

    contract = client.get(f"/api/report-collaboration/runs/{run_id}/nodes/{plan['nodes'][0]['id']}/contract")
    assert contract.status_code == 200
    assert contract.json()["mission"]
    assert "content" not in contract.json()

    cancelled = client.post(
        f"/api/report-collaboration/runs/{run_id}/cancel",
        headers={"X-Idempotency-Key": "k-cancel"},
    )
    assert cancelled.status_code == 200
    assert cancelled.json()["cancel_requested"] is True
    assert cancelled.json()["status"] == "cancelled"

    async def _other_user(_request=None):
        return "user_b"

    monkeypatch.setattr(report_collaboration, "get_current_user", _other_user)
    hidden = client.get(f"/api/report-collaboration/sessions/{session_id}")
    assert hidden.status_code == 404

    async def _owner(_request=None):
        return "user_a"

    monkeypatch.setattr(report_collaboration, "get_current_user", _owner)
    deleted = client.delete(f"/api/report-collaboration/sessions/{session_id}")
    assert deleted.status_code == 204


def test_runtime_command_http_classify_confirm_cancel_and_idempotency(monkeypatch: pytest.MonkeyPatch) -> None:
    client, _store = _client(monkeypatch)
    session = client.post(
        "/api/report-collaboration/sessions",
        json={"title": "运行期命令"},
        headers={"X-Idempotency-Key": "cmd-session"},
    ).json()
    session_id = session["id"]
    sent = client.post(
        f"/api/report-collaboration/sessions/{session_id}/messages",
        json={"text": "写一份面向企业战略决策人员的新能源汽车市场份额趋势报告"},
        headers={"X-Idempotency-Key": "cmd-message"},
    )
    assert sent.status_code == 200
    revision = client.get(f"/api/report-collaboration/sessions/{session_id}").json()["session"]["requirement_revision"]
    planned = client.post(
        f"/api/report-collaboration/sessions/{session_id}/plan-requests",
        headers={"X-Idempotency-Key": "cmd-plan", "X-Expected-Revision": str(revision)},
    )
    assert planned.status_code == 204
    plan = next(item for item in client.get(f"/api/report-collaboration/sessions/{session_id}/plans").json()["plans"] if item["recommended"])
    selected = client.post(
        f"/api/report-collaboration/sessions/{session_id}/plans/{plan['id']}/select",
        headers={"X-Idempotency-Key": "cmd-select", "X-Expected-Revision": str(revision)},
    )
    assert selected.status_code == 200
    run = client.post(
        f"/api/report-collaboration/sessions/{session_id}/runs",
        json={"plan_id": plan["id"]},
        headers={"X-Idempotency-Key": "cmd-run", "X-Expected-Revision": str(revision)},
    ).json()
    run_id = run["id"]
    target = plan["nodes"][0]["id"]

    rerun = client.post(
        f"/api/report-collaboration/runs/{run_id}/commands",
        json={"text": "重新执行这个节点", "target_node_id": target},
        headers={"X-Idempotency-Key": "cmd-rerun"},
    )
    assert rerun.status_code == 200, rerun.text
    assert rerun.json()["intent"] == "rerun_node"
    assert rerun.json()["status"] == "completed"
    snapshot_after_command = client.get(f"/api/report-collaboration/sessions/{session_id}").json()
    runtime_message = next(item for item in snapshot_after_command["messages"] if item.get("command_id") == rerun.json()["command_id"])
    assert runtime_message["metadata"]["idempotency_key"] == "cmd-rerun"
    replay = client.post(
        f"/api/report-collaboration/runs/{run_id}/commands",
        json={"text": "这次文本应该被幂等忽略", "target_node_id": target},
        headers={"X-Idempotency-Key": "cmd-rerun"},
    )
    assert replay.json()["command_id"] == rerun.json()["command_id"]

    replan = client.post(
        f"/api/report-collaboration/runs/{run_id}/commands",
        json={"text": "重新规划，换一套方案"},
        headers={"X-Idempotency-Key": "cmd-replan"},
    )
    assert replan.status_code == 200
    assert replan.json()["status"] == "confirmation_required"
    confirmed = client.post(
        f"/api/report-collaboration/runs/{run_id}/commands/{replan.json()['command_id']}/confirm",
        headers={"X-Idempotency-Key": "cmd-confirm"},
    )
    assert confirmed.status_code == 200, confirmed.text
    assert confirmed.json()["status"] == "completed"

    ambiguous = client.post(
        f"/api/report-collaboration/runs/{run_id}/commands",
        json={"text": "这里不太对，改一下"},
        headers={"X-Idempotency-Key": "cmd-ambiguous"},
    )
    assert ambiguous.status_code == 200
    assert ambiguous.json()["status"] == "confirmation_required"
    cancelled = client.post(
        f"/api/report-collaboration/runs/{run_id}/commands/{ambiguous.json()['command_id']}/cancel",
        headers={"X-Idempotency-Key": "cmd-cancel"},
    )
    assert cancelled.status_code == 200
    assert cancelled.json()["status"] == "cancelled"

    blocked = client.post(
        f"/api/report-collaboration/sessions/{session_id}/messages",
        json={"text": "运行中不要走规划前接口"},
        headers={"X-Idempotency-Key": "cmd-blocked-message"},
    )
    assert blocked.status_code == 409
    assert blocked.json()["code"] == "RUN_ACTIVE"

    events = client.get(f"/api/report-collaboration/runs/{run_id}/events?after_seq=0&limit=500").json()["events"]
    rerun_classified = [
        item
        for item in events
        if item["type"] == "command.classified" and item["data"]["command"]["command_id"] == rerun.json()["command_id"]
    ]
    assert len(rerun_classified) == 1
    assert any(
        item["type"] == "command.cancelled" and item["data"]["command_id"] == cancelled.json()["command_id"]
        for item in events
    )


@pytest.mark.asyncio
async def test_rewrite_apply_restore_http(monkeypatch: pytest.MonkeyPatch) -> None:
    client, store = _client(monkeypatch)
    created = client.post(
        "/api/report-collaboration/sessions",
        json={"title": "版本联调"},
        headers={"X-Idempotency-Key": "http-rewrite-session"},
    )
    session_id = created.json()["id"]
    await store.create_report_version(
        session_id,
        idempotency_key="http-initial",
        markdown="# 标题\n\n# 市场分析\n\n市场变化，引用 c1。\n\n# 结论与局限\n\n证据有限。",
        source_index=[{"id": "c1", "title": "月报", "added_at": "2026-09-06T00:00:00Z"}],
        change_type="initial",
        change_summary="首稿",
    )
    rewrite = client.post(
        f"/api/report-collaboration/sessions/{session_id}/report-rewrites",
        json={"instruction": "语气更审慎", "section": "市场分析"},
        headers={"X-Idempotency-Key": "http-rewrite"},
    )
    assert rewrite.status_code == 200, rewrite.text
    applied = client.post(
        f"/api/report-collaboration/sessions/{session_id}/report-rewrites/{rewrite.json()['id']}/apply",
        headers={"X-Idempotency-Key": "http-apply"},
    )
    assert applied.status_code == 200, applied.text
    assert applied.json()["version"] == 2
    versions = client.get(f"/api/report-collaboration/sessions/{session_id}/reports").json()["versions"]
    initial_id = next(item["id"] for item in versions if item["version"] == 1)
    restored = client.post(
        f"/api/report-collaboration/sessions/{session_id}/report-versions/{initial_id}/restore",
        headers={"X-Idempotency-Key": "http-restore"},
    )
    assert restored.status_code == 200, restored.text
    assert restored.json()["version"] == 3
    assert restored.json()["change_type"] == "restore"