"""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 intent, report_collaboration
from app.gateway.routers._roundtable_seed import REPORT_COLLABORATION_INTENT_AGENT_ID
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)


class _PlannerFakeLLM:
    """规划师假模型：需求确认回「读者」追问；工作流设计回单份定制方案。"""

    _converse = (
        '{"reply": "好的。先确认：这份报告的主要读者是谁？", "action": "clarify",'
        '"questions": ["这份报告的主要读者是谁？"], "impact": ["读者影响研究线划分"],'
        '"requirement": {"topic": "新能源汽车趋势报告"}}'
    )
    _ready = (
        '{"reply": "明白了，读者为企业战略决策人员。我来设计工作流。", "action": "ready",'
        '"assumptions": [], "requirement": {"topic": "新能源汽车趋势报告", "audience": "企业战略决策人员"}}'
    )
    _design = (
        '{"reply": "已按需求设计定制工作流。", "strategy": "custom", "title": "定制方案",'
        '"summary": "研究后写作", "rationale": "贴合需求", "estimated_duration_seconds": 900,'
        '"estimated_cost_level": "low", "quality_gate_ids": ["cite_required"],'
        '"nodes": ['
        '{"slot_id": "research-1", "role_key": "researcher", "display_name": "研究员", "mission": "检索证据", "depends_on": [], "output_artifact_type": "EvidenceBundle"},'
        '{"slot_id": "write-1", "role_key": "writer", "display_name": "撰写员", "mission": "成稿", "depends_on": ["research-1"], "output_artifact_type": "ReportSectionDraft"}]}'
    )

    async def ainvoke(self, messages):
        from types import SimpleNamespace

        prompt = str(messages[-1][1]) if messages else ""
        if "请设计工作流" in prompt:
            payload = self._design
        elif "企业战略决策人员" in prompt or "读者" in prompt and "用户最新消息" in prompt:
            payload = self._ready
        else:
            payload = self._converse
        return SimpleNamespace(content=payload)

    async def astream(self, messages):
        from types import SimpleNamespace

        yield SimpleNamespace(
            content="",
            additional_kwargs={"reasoning_content": "正在比较可用智能体的能力与依赖关系。"},
        )
        yield SimpleNamespace(content=self._design, additional_kwargs={})


class _BrokenPlannerStream:
    async def astream(self, messages):
        if False:
            yield None
        raise RuntimeError("provider stream closed early")


class _ReasoningOnlyPlannerStream:
    async def astream(self, messages):
        from types import SimpleNamespace

        yield SimpleNamespace(
            content="",
            additional_kwargs={"reasoning_content": _PlannerFakeLLM._design},
        )


def test_report_intent_agent_contract_supports_nested_requirement() -> None:
    assert (
        intent._requested_intent_agent_id(
            {"intent_agent_id": REPORT_COLLABORATION_INTENT_AGENT_ID}
        )
        == REPORT_COLLABORATION_INTENT_AGENT_ID
    )
    content = (
        "意图已经明确。\n[INTENT_READY]\n```json\n"
        '{"objective":"写报告","constraints":[],"assumptions":[],'
        '"report_requirement":{"topic":"新能源汽车","required_angles":["市场"]}}\n```'
    )
    parsed = intent._extract_intent_ready(content)
    assert parsed is not None
    assert parsed[0]["report_requirement"]["topic"] == "新能源汽车"
    assert content[: parsed[1]].endswith("```")


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_planner_llm = _PlannerFakeLLM
    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 _stream_plans(
    client: TestClient,
    session_id: str,
    key: str,
    adjustment_text: str | None = None,
) -> str:
    with client.stream(
        "POST",
        f"/api/report-collaboration/sessions/{session_id}/plan-stream",
        json={"adjustment_text": adjustment_text},
        headers={"X-Idempotency-Key": key},
    ) as response:
        assert response.status_code == 200, response.text
        assert response.headers["content-type"].startswith("text/event-stream")
        return "".join(response.iter_text())


def _ready_session(client: TestClient, *, key_prefix: str) -> str:
    session = client.post(
        "/api/report-collaboration/sessions",
        json={"title": "规划恢复测试"},
        headers={"X-Idempotency-Key": f"{key_prefix}-session"},
    ).json()
    session_id = session["id"]
    response = client.post(
        f"/api/report-collaboration/sessions/{session_id}/intent-turns",
        json={
            "text": "规划一份新能源汽车市场报告",
            "assistant_content": "意图已经确认。",
            "intent_thread_id": f"{key_prefix}-thread",
            "status": "done",
            "summary": {
                "objective": "形成新能源汽车市场判断",
                "constraints": [],
                "assumptions": [],
                "report_requirement": {
                    "topic": "新能源汽车市场报告",
                    "required_angles": ["市场规模", "竞争格局"],
                    "excluded_angles": [],
                },
            },
            "clarifications": [],
        },
        headers={"X-Idempotency-Key": f"{key_prefix}-intent"},
    )
    assert response.status_code == 200
    return session_id


def test_plan_stream_recovers_to_candidates_when_provider_stream_fails(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    client, _store = _client(monkeypatch)
    client.app.state.report_collaboration_planner_llm = _BrokenPlannerStream
    session_id = _ready_session(client, key_prefix="recover")

    stream = _stream_plans(client, session_id, "recover-plan")

    assert "event: phase" in stream
    assert "正在收敛为可执行候选工作流" in stream
    assert "event: complete" in stream
    snapshot = client.get(f"/api/report-collaboration/sessions/{session_id}").json()
    assert snapshot["session"]["status"] == "proposal_ready"
    assert len(snapshot["plans"]) == 3


def test_plan_stream_accepts_final_json_from_reasoning_channel(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    client, _store = _client(monkeypatch)
    client.app.state.report_collaboration_planner_llm = _ReasoningOnlyPlannerStream
    session_id = _ready_session(client, key_prefix="reasoning-json")

    stream = _stream_plans(client, session_id, "reasoning-json-plan")

    assert "event: reasoning" in stream
    assert "event: complete" in stream
    assert "正在收敛为可执行候选工作流" not in stream
    snapshot = client.get(f"/api/report-collaboration/sessions/{session_id}").json()
    assert snapshot["session"]["status"] == "proposal_ready"
    assert any(plan["strategy"] == "custom" for plan in snapshot["plans"])


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 == 409
    assert sent.json()["code"] == "USE_STREAMING_INTENT"

    streamed_turn = client.post(
        f"/api/report-collaboration/sessions/{session_id}/intent-turns",
        json={
            "text": "写一份新能源汽车趋势报告",
            "assistant_content": "<think>需要确认目标读者</think>请问这份报告主要给谁看？",
            "intent_thread_id": "intent-thread-1",
            "status": "clarification",
            "summary": None,
            "clarifications": [
                {
                    "id": "clar-1",
                    "question": "这份报告主要给谁看？",
                    "clarification_type": "missing_info",
                    "options": ["企业战略决策人员", "投资机构"],
                }
            ],
        },
        headers={"X-Idempotency-Key": "k-intent-turn-1"},
    )
    assert streamed_turn.status_code == 200, streamed_turn.text

    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"] == 0
    assert body["messages"][0]["content"].startswith("写一份")
    assert any(item["role"] == "ai" and "给谁看" in item["content"] for item in body["messages"])
    intent_ai = next(item for item in body["messages"] if item["role"] == "ai")
    assert intent_ai["metadata"]["reasoning_content"] == "需要确认目标读者"
    assert intent_ai["tool_calls"][0]["name"] == "ask_clarification"
    clarification_tool = next(item for item in body["messages"] if item["role"] == "tool")
    assert clarification_tool["name"] == "ask_clarification"
    assert clarification_tool["tool_call_id"] == "clar-1"
    assert clarification_tool["metadata"]["clarification"]["question"]
    assert body["last_seq"] == 0
    assert body["plans"] == []

    # 即使绕过前端直接调用兼容接口，也不能在规划师确认意图前抢跑工作流规划。
    premature = client.post(
        f"/api/report-collaboration/sessions/{session_id}/plan-requests",
        headers={"X-Idempotency-Key": "k-plan-premature", "X-Expected-Revision": "0"},
    )
    assert premature.status_code == 422
    assert premature.json()["code"] == "REQUIREMENT_NOT_READY"

    # 用户回答完意图规划师的澄清问题后，接口只落意图快照并切到 planning。
    # 工作流必须由独立规划智能体通过 SSE 运行，让客户端能实时显示过程。
    clarified = client.post(
        f"/api/report-collaboration/sessions/{session_id}/intent-turns",
        json={
            "text": "主要给企业战略决策人员看，重点关注市场格局和价格策略",
            "assistant_content": "明白，我会按这个意图规划报告。\n[INTENT_READY]\n```json\n{}\n```",
            "intent_thread_id": "intent-thread-1",
            "status": "done",
            "summary": {
                "objective": "撰写新能源汽车趋势报告",
                "constraints": [],
                "assumptions": ["中文输出"],
                "report_requirement": {
                    "topic": "新能源汽车趋势报告",
                    "objective": "支持企业战略决策",
                    "audience": "企业战略决策人员",
                    "language": "中文",
                    "required_angles": ["市场格局", "价格策略"],
                    "excluded_angles": [],
                },
            },
            "clarifications": [],
        },
        headers={"X-Idempotency-Key": "k-intent-turn-2"},
    )
    assert clarified.status_code == 200, clarified.text

    planning_body = clarified.json()
    assert planning_body["session"]["status"] == "planning"
    assert planning_body["plans"] == []

    planning_stream = _stream_plans(client, session_id, "k-plan-stream")
    assert "event: phase" in planning_stream
    assert "正在读取可用智能体及能力描述" in planning_stream
    assert "event: reasoning" in planning_stream
    assert "正在比较可用智能体的能力与依赖关系" in planning_stream
    assert "event: complete" in planning_stream

    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"
    revision = proposed_body["session"]["requirement_revision"]
    assert revision == 1
    assert proposed_body["messages"][-2]["content"] == "明白，我会按这个意图规划报告。"
    planner_message = next(
        item for item in proposed_body["messages"] if item.get("name") == "工作流规划师"
    )
    assert "正在比较可用智能体" in planner_message["metadata"]["reasoning_content"]
    assert [item["name"] for item in planner_message["tool_calls"]] == [
        "workflow_planning_progress",
        "workflow_planning_progress",
        "workflow_planning_progress",
    ]
    plans = proposed_body["plans"]
    assert {item["strategy"] for item in plans} == {"custom", "balanced_review", "parallel_depth"}
    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"])

    adjusted = client.post(
        f"/api/report-collaboration/sessions/{session_id}/messages",
        json={"text": "以第二个方案为基础，去掉审核智能体，再增加行业分析智能体"},
        headers={"X-Idempotency-Key": "k-adjust-plan"},
    )
    assert adjusted.status_code == 200, adjusted.text
    adjustment_stream = _stream_plans(
        client,
        session_id,
        "k-adjust-plan-stream",
        "以第二个方案为基础，去掉审核智能体，再增加行业分析智能体",
    )
    assert "event: complete" in adjustment_stream
    adjusted_body = client.get(f"/api/report-collaboration/sessions/{session_id}").json()
    active_plans = [item for item in adjusted_body["plans"] if item["status"] == "proposed"]
    assert len(active_plans) == 3
    assert active_plans[0]["proposal_group_id"] != plan["proposal_group_id"]
    assert any("第二个方案" in item["content"] for item in adjusted_body["messages"])
    plan = next(item for item in active_plans if item.get("recommended"))

    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"

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

    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}/intent-turns",
        json={
            "text": "写一份面向企业战略决策人员的新能源汽车市场份额趋势报告",
            "assistant_content": "意图已明确。",
            "intent_thread_id": "intent-thread-cmd",
            "status": "done",
            "summary": {
                "objective": "撰写新能源汽车市场份额趋势报告",
                "constraints": [],
                "assumptions": [],
                "report_requirement": {
                    "topic": "新能源汽车市场份额趋势报告",
                    "objective": "支持企业战略决策",
                    "audience": "企业战略决策人员",
                    "required_angles": ["市场份额"],
                    "excluded_angles": [],
                },
            },
            "clarifications": [],
        },
        headers={"X-Idempotency-Key": "cmd-message"},
    )
    assert sent.status_code == 200
    assert sent.json()["session"]["status"] == "planning"
    command_planning_stream = _stream_plans(client, session_id, "cmd-plan-stream")
    assert "event: complete" in command_planning_stream
    ready_snapshot = client.get(f"/api/report-collaboration/sessions/{session_id}").json()
    revision = ready_snapshot["session"]["requirement_revision"]
    assert ready_snapshot["session"]["status"] == "proposal_ready"
    plan = next(item for item in ready_snapshot["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"