"""Gateway API for the AgentScope report-collaboration workbench.

Plan/run writes enqueue durable jobs and return immediately. AgentScope
execution is not started from this router. ``GET /runs/{id}/stream`` is a
read-only durable SSE tail: disconnect does not cancel the run.
"""

from __future__ import annotations

import asyncio
import json
import logging
import re
from collections.abc import AsyncIterator
from typing import Any

from fastapi import APIRouter, Header, HTTPException, Query, Request
from fastapi.responses import JSONResponse, Response, StreamingResponse
from pydantic import BaseModel, Field

from app.gateway.deps import get_current_user
from app.report_collaboration.agentscope_runtime.model_resolver import freeze_plan_models
from app.report_collaboration.contracts.requirements import ReportRequirementSnapshot
from app.report_collaboration.contracts.snapshots import SessionSnapshot
from app.report_collaboration.execution.live_hub import iter_run_sse, replay_cursor
from app.report_collaboration.interventions.intent_router import build_runtime_intent_router
from app.report_collaboration.interventions.service import RuntimeCommandService
from app.report_collaboration.planning import PlanController, PlanProposalService, project_agent_catalog, requirement_from_session
from app.report_collaboration.planning.planner_agent import PlannerAgent, _extract_json
from app.report_collaboration.reporting import ConservativeRewriteGenerator, ModelRewriteGenerator, ReportRewriteService, freeze_report_template
from app.report_collaboration.reporting.templates import requirement_snapshot_from_record
from app.report_collaboration.security import freeze_run_budget
from deerflow.config.app_config import get_app_config
from deerflow.persistence.report_collaboration import (
    ReportCollaborationConflictError,
    ReportCollaborationLimitError,
    ReportCollaborationNotFoundError,
    ReportCollaborationStore,
    ReportCollaborationValidationError,
)

router = APIRouter(prefix="/api/report-collaboration", tags=["report-collaboration"])
logger = logging.getLogger(__name__)


class CreateSessionBody(BaseModel):
    title: str | None = Field(default=None, max_length=255)


class UpdateSessionBody(BaseModel):
    title: str | None = Field(default=None, max_length=255)


class SendMessageBody(BaseModel):
    text: str = Field(..., min_length=1, max_length=20_000)
    target_node_id: str | None = None
    agent_run_id: str | None = None


class CompleteIntentTurnBody(BaseModel):
    """One completed turn from the real streaming intent-agent thread."""

    text: str = Field(..., min_length=1, max_length=20_000)
    assistant_content: str = Field(default="", max_length=100_000)
    reasoning_content: str | None = Field(default=None, max_length=100_000)
    intent_thread_id: str = Field(..., min_length=1, max_length=128)
    status: str = Field(..., pattern="^(asking|clarification|done)$")
    summary: dict[str, Any] | None = None
    clarifications: list[dict[str, Any]] = Field(default_factory=list, max_length=8)


class StreamPlanBody(BaseModel):
    adjustment_text: str | None = Field(default=None, max_length=20_000)


class StartRunBody(BaseModel):
    plan_id: str = Field(..., min_length=1, max_length=64)


class CreateCommandBody(BaseModel):
    text: str = Field(..., min_length=1, max_length=20_000)
    target_node_id: str | None = None
    agent_run_id: str | None = None


class CreateRewriteBody(BaseModel):
    instruction: str = Field(..., min_length=1, max_length=20_000)
    section: str | None = None


def _store(request: Request) -> ReportCollaborationStore:
    store = getattr(request.app.state, "report_collaboration_store", None)
    if store is None:
        raise HTTPException(status_code=503, detail="Report collaboration store is unavailable")
    return store


_THINK_BLOCK = re.compile(r"<think>(.*?)</think>", re.DOTALL | re.IGNORECASE)


def _intent_display_content(content: str) -> tuple[str, str | None]:
    """Separate visible answer/reasoning and hide the machine hand-off block."""

    reasoning = "\n\n".join(part.strip() for part in _THINK_BLOCK.findall(content) if part.strip())
    visible = _THINK_BLOCK.sub("", content)
    visible = visible.split("[INTENT_READY]", 1)[0].strip()
    return visible, reasoning or None


def _planner_chunk_parts(chunk: Any) -> tuple[str, str]:
    raw_content = getattr(chunk, "content", None)
    if isinstance(raw_content, str):
        content = raw_content
    elif isinstance(raw_content, list):
        content = "".join(
            str(block.get("text") or "")
            if isinstance(block, dict)
            else str(getattr(block, "text", "") or "")
            for block in raw_content
        )
    else:
        content = ""
    additional = getattr(chunk, "additional_kwargs", None) or {}
    reasoning = ""
    if isinstance(additional, dict):
        value = additional.get("reasoning_content") or additional.get("reasoning")
        if isinstance(value, str):
            reasoning = value
    return content, reasoning


def _planner_sse(event: str, data: dict[str, Any]) -> str:
    return f"event: {event}\ndata: {json.dumps(data, ensure_ascii=False, default=str)}\n\n"


def _planner_inline_reasoning(text: str) -> str:
    """Return reasoning emitted inside `<think>` blocks, including an open block."""

    parts = list(_THINK_BLOCK.findall(text))
    last_open = text.lower().rfind("<think>")
    last_close = text.lower().rfind("</think>")
    if last_open > last_close:
        parts.append(text[last_open + len("<think>") :])
    return "".join(parts)


def _report_requirement(summary: dict[str, Any], revision: int) -> ReportRequirementSnapshot:
    raw = summary.get("report_requirement") or summary.get("reportRequirement")
    if not isinstance(raw, dict):
        objective = str(summary.get("objective") or "").strip()
        raw = {
            "topic": objective,
            "objective": objective,
            "required_angles": [],
            "excluded_angles": [],
        }
    payload = dict(raw)
    payload["revision"] = revision
    payload.setdefault("required_angles", [])
    payload.setdefault("excluded_angles", [])
    return ReportRequirementSnapshot.model_validate(payload)


def _runtime_commands(request: Request) -> RuntimeCommandService:
    service = getattr(request.app.state, "report_collaboration_command_service", None)
    if service is None:
        app_config = get_app_config()
        service = RuntimeCommandService(
            _store(request),
            router=build_runtime_intent_router(app_config),
            config=app_config.report_collaboration,
        )
        request.app.state.report_collaboration_command_service = service
    return service


def _require_enabled() -> JSONResponse | None:
    if not get_app_config().report_collaboration.enabled:
        return _error(503, "REPORT_COLLABORATION_DISABLED", "报告协作工作台未启用")
    return None


async def _require_user(request: Request) -> str:
    user_id = await get_current_user(request)
    if not user_id:
        raise HTTPException(status_code=401, detail="Authentication required")
    return str(user_id)


def _error(status: int, code: str, message: str, detail: Any = None) -> JSONResponse:
    body: dict[str, Any] = {"code": code, "message": message}
    if detail is not None:
        body["detail"] = detail
    return JSONResponse(status_code=status, content=body)


def _require_idempotency(idempotency_key: str | None) -> str | JSONResponse:
    key = (idempotency_key or "").strip()
    if not key:
        return _error(422, "IDEMPOTENCY_KEY_REQUIRED", "写操作必须携带 X-Idempotency-Key")
    if len(key) > 128:
        return _error(422, "IDEMPOTENCY_KEY_INVALID", "X-Idempotency-Key 过长")
    return key


def _map_store_error(exc: Exception) -> JSONResponse:
    if isinstance(exc, ReportCollaborationNotFoundError):
        return _error(404, "NOT_FOUND", str(exc))
    if isinstance(exc, ReportCollaborationConflictError):
        return _error(409, exc.code, str(exc), detail={"current_revision": exc.current_revision})
    if isinstance(exc, ReportCollaborationValidationError):
        return _error(422, exc.code, str(exc))
    if isinstance(exc, ReportCollaborationLimitError):
        return _error(429, exc.code, str(exc), detail={"usage": exc.usage, "limits": exc.limits})
    raise exc


def _rewrite_service(store: ReportCollaborationStore) -> ReportRewriteService:
    cfg = get_app_config().report_collaboration
    generator = ModelRewriteGenerator(cfg.writer_model) if cfg.writer_model else ConservativeRewriteGenerator()
    return ReportRewriteService(store, generator=generator)


async def _frozen_run_template(request: Request, session: dict[str, Any], plan: dict[str, Any]) -> dict[str, Any]:
    requirement = requirement_snapshot_from_record(session)
    structure_id = plan.get("report_structure_id") or (requirement.report_structure_id if requirement else None)
    structure = None
    structure_store = getattr(request.app.state, "report_structure_store", None)
    if structure_id and structure_store is not None:
        structure = await structure_store.get_structure(str(structure_id))
    return freeze_report_template(structure, requirement=requirement).model_dump(mode="json")


async def _owned_session(store: ReportCollaborationStore, session_id: str, user_id: str) -> dict[str, Any] | JSONResponse:
    record = await store.get_session_record(session_id)
    if record is None or record.get("owner_id") != user_id:
        return _error(404, "NOT_FOUND", f"report collaboration session not found: {session_id}")
    return record


async def _owned_run(store: ReportCollaborationStore, run_id: str, user_id: str) -> dict[str, Any] | JSONResponse:
    run = await store.get_run(run_id)
    if run is None:
        return _error(404, "NOT_FOUND", f"report collaboration run not found: {run_id}")
    owned = await _owned_session(store, run["session_id"], user_id)
    if isinstance(owned, JSONResponse):
        return owned
    return run


@router.post("/sessions")
async def create_session(
    request: Request,
    body: CreateSessionBody | None = None,
    x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
):
    if disabled := _require_enabled():
        return disabled
    user_id = await _require_user(request)
    key = _require_idempotency(x_idempotency_key)
    if isinstance(key, JSONResponse):
        return key
    payload = body or CreateSessionBody()
    try:
        return await _store(request).create_session(owner_id=user_id, title=payload.title or "", idempotency_key=key)
    except Exception as exc:
        return _map_store_error(exc)


@router.get("/sessions")
async def list_sessions(request: Request):
    if disabled := _require_enabled():
        return disabled
    user_id = await _require_user(request)
    sessions = await _store(request).list_sessions(user_id)
    return {"sessions": sessions}


@router.get("/sessions/{session_id}")
async def get_session_snapshot(request: Request, session_id: str):
    if disabled := _require_enabled():
        return disabled
    user_id = await _require_user(request)
    store = _store(request)
    owned = await _owned_session(store, session_id, user_id)
    if isinstance(owned, JSONResponse):
        return owned
    snapshot = await store.get_snapshot(session_id)
    return SessionSnapshot.model_validate(snapshot).model_dump()


@router.patch("/sessions/{session_id}")
async def update_session(
    request: Request,
    session_id: str,
    body: UpdateSessionBody,
    x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
    x_expected_revision: int | None = Header(default=None, alias="X-Expected-Revision"),
):
    if disabled := _require_enabled():
        return disabled
    user_id = await _require_user(request)
    key = _require_idempotency(x_idempotency_key)
    if isinstance(key, JSONResponse):
        return key
    store = _store(request)
    owned = await _owned_session(store, session_id, user_id)
    if isinstance(owned, JSONResponse):
        return owned
    try:
        updated = await store.update_session(session_id, title=body.title, expected_revision=x_expected_revision)
        await store.create_command(session_id, idempotency_key=key, text=body.title or "", operation="update_session", status="completed")
        return updated
    except Exception as exc:
        return _map_store_error(exc)


@router.delete("/sessions/{session_id}")
async def delete_session(request: Request, session_id: str):
    if disabled := _require_enabled():
        return disabled
    user_id = await _require_user(request)
    store = _store(request)
    owned = await _owned_session(store, session_id, user_id)
    if isinstance(owned, JSONResponse):
        return owned
    await store.delete_session(session_id)
    return Response(status_code=204)


@router.get("/sessions/{session_id}/messages")
async def list_messages(request: Request, session_id: str):
    if disabled := _require_enabled():
        return disabled
    user_id = await _require_user(request)
    store = _store(request)
    owned = await _owned_session(store, session_id, user_id)
    if isinstance(owned, JSONResponse):
        return owned
    return {"messages": await store.list_messages(session_id)}


@router.post("/sessions/{session_id}/messages")
async def send_message(
    request: Request,
    session_id: str,
    body: SendMessageBody,
    x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
):
    if disabled := _require_enabled():
        return disabled
    user_id = await _require_user(request)
    key = _require_idempotency(x_idempotency_key)
    if isinstance(key, JSONResponse):
        return key
    store = _store(request)
    owned = await _owned_session(store, session_id, user_id)
    if isinstance(owned, JSONResponse):
        return owned
    metadata: dict[str, Any] = {"idempotency_key": key}
    if body.target_node_id:
        metadata["target_node_id"] = body.target_node_id
    if body.agent_run_id:
        metadata["agent_run_id"] = body.agent_run_id
    try:
        existing = await store.get_message_by_idempotency(key)
        if existing is not None:
            return existing
        snapshot = await store.get_snapshot(session_id)
        active = snapshot.get("run") or {}
        if str(active.get("status") or "") in {
            "queued",
            "planning",
            "running",
            "awaiting_input",
            "reviewing",
            "recovering",
        }:
            return _error(
                409,
                "RUN_ACTIVE",
                "运行进行中请使用 /runs/{run_id}/commands 发送干预，不要走规划前消息接口",
                detail={"run_id": active.get("id"), "status": active.get("status")},
            )
        if str(owned.get("status") or "") in {"empty", "clarifying"}:
            return _error(
                409,
                "USE_STREAMING_INTENT",
                "需求确认阶段必须调用真实意图智能体流，不能使用阻塞消息接口",
            )
        created = await store.create_message(
            session_id,
            role="human",
            content=body.text,
            idempotency_key=key,
            metadata=metadata,
            agent_run_id=body.agent_run_id,
            node_run_id=body.target_node_id,
        )
        return created
    except Exception as exc:
        return _map_store_error(exc)


@router.post("/sessions/{session_id}/intent-turns")
async def complete_intent_turn(
    request: Request,
    session_id: str,
    body: CompleteIntentTurnBody,
    x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
):
    """Persist one streamed agent turn and auto-plan after its READY frame.

    No model is invoked for intent recognition here: the browser has already
    consumed the real `/api/intent/stream` run incrementally.
    """

    if disabled := _require_enabled():
        return disabled
    user_id = await _require_user(request)
    key = _require_idempotency(x_idempotency_key)
    if isinstance(key, JSONResponse):
        return key
    store = _store(request)
    owned = await _owned_session(store, session_id, user_id)
    if isinstance(owned, JSONResponse):
        return owned
    try:
        existing_user = await store.get_message_by_idempotency(key)
        if existing_user is None:
            await store.create_message(
                session_id,
                role="human",
                content=body.text,
                idempotency_key=key,
                metadata={"intent_thread_id": body.intent_thread_id, "source": "intent_agent_stream"},
            )

        visible, inline_reasoning = _intent_display_content(body.assistant_content)
        reasoning = body.reasoning_content or inline_reasoning
        tool_calls = [
            {
                "id": str(item.get("id") or f"clarification-{index}"),
                "name": "ask_clarification",
                "args": item,
            }
            for index, item in enumerate(body.clarifications)
        ]
        await store.create_message(
            session_id,
            role="ai",
            content=visible,
            idempotency_key=f"{key}:intent-ai",
            name="协作意图规划师",
            metadata={
                "intent_thread_id": body.intent_thread_id,
                "intent_status": body.status,
                **({"reasoning_content": reasoning} if reasoning else {}),
            },
            tool_calls=tool_calls,
            completed=True,
        )
        # MessageList renders an actionable clarification card from a real
        # `tool` message, not from the assistant's tool-call trace alone.
        # Mirror every streamed ask_clarification result into the report
        # collaboration timeline so it survives refresh and is answerable.
        for index, item in enumerate(body.clarifications):
            call_id = str(item.get("id") or f"clarification-{index}")
            question = str(item.get("question") or "请补充报告需求。")
            clarification = {
                "question": question,
                "clarification_type": str(
                    item.get("clarification_type") or item.get("clarificationType") or ""
                ),
                "context": str(item.get("context") or item.get("clarification_context") or ""),
                "options": list(item.get("options") or []),
                "allow_custom": item.get("allow_custom", item.get("allowCustom", True)) is not False,
                "allow_multiple": item.get("allow_multiple", item.get("allowMultiple")),
            }
            await store.create_message(
                session_id,
                role="tool",
                name="ask_clarification",
                content=f"❓ {question}",
                tool_call_id=call_id,
                idempotency_key=f"{key}:intent-clarification:{index}",
                metadata={"clarification": clarification, "intent_thread_id": body.intent_thread_id},
                completed=True,
            )

        record = await store.get_session_record(session_id)
        if record is None:
            return _error(404, "NOT_FOUND", "会话不存在")
        raw_state = record.get("requirement_json")
        current_state = dict(raw_state) if isinstance(raw_state, dict) else {}
        current_revision = int(record.get("requirement_revision") or 0)
        turn_already_applied = current_state.get("last_intent_turn_key") == key
        if turn_already_applied:
            existing_plans = await store.list_plans(session_id)
            if body.status != "done" or existing_plans:
                return await store.get_snapshot(session_id)

        if body.status != "done" or not isinstance(body.summary, dict):
            await store.save_requirement_state(
                session_id,
                requirement_json={
                    **current_state,
                    "intent_thread_id": body.intent_thread_id,
                    "ready_to_plan": False,
                    "last_intent_turn_key": key,
                },
                requirement_revision=current_revision,
                status="clarifying",
            )
            return await store.get_snapshot(session_id)

        requirement = _report_requirement(
            body.summary,
            current_revision if turn_already_applied else current_revision + 1,
        )
        await store.save_requirement_state(
            session_id,
            requirement_json={
                "snapshot": requirement.model_dump(),
                "clarifications": list(current_state.get("clarifications") or []),
                "assumptions": list(body.summary.get("assumptions") or []),
                "clarification_round": int(current_state.get("clarification_round") or 0),
                "ready_to_plan": True,
                "intent_thread_id": body.intent_thread_id,
                "last_intent_turn_key": key,
            },
            requirement_revision=requirement.revision,
            status="planning",
            title=requirement.topic[:40],
        )
        return await store.get_snapshot(session_id)
    except Exception as exc:
        return _map_store_error(exc)


@router.post("/sessions/{session_id}/plan-stream")
async def stream_plans(
    request: Request,
    session_id: str,
    body: StreamPlanBody,
    x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
):
    """Run the workflow-planning agent as a real SSE stream, then persist plans."""

    if disabled := _require_enabled():
        return disabled
    user_id = await _require_user(request)
    key = _require_idempotency(x_idempotency_key)
    if isinstance(key, JSONResponse):
        return key
    store = _store(request)
    owned = await _owned_session(store, session_id, user_id)
    if isinstance(owned, JSONResponse):
        return owned
    if owned.get("selected_plan_id"):
        return _error(409, "PLAN_ALREADY_SELECTED", "会话已选定方案，无法重新规划")

    try:
        command = await store.enqueue_plan_request(
            session_id,
            idempotency_key=key,
            expected_revision=int(owned.get("requirement_revision") or 0),
        )
    except Exception as exc:
        return _map_store_error(exc)

    if command.get("status") == "completed":

        async def replay_completed() -> AsyncIterator[str]:
            yield _planner_sse("complete", {"snapshot": await store.get_snapshot(session_id)})

        return StreamingResponse(replay_completed(), media_type="text/event-stream")
    if command.get("status") != "pending":
        return _error(409, "PLAN_REQUEST_NOT_PENDING", "本次规划请求已结束，请重新发起")

    async def generate() -> AsyncIterator[str]:
        service = PlanProposalService(
            store,
            agent_store=getattr(request.app.state, "agent_store", None),
            config=get_app_config().report_collaboration,
            llm_factory=getattr(request.app.state, "report_collaboration_planner_llm", None),
        )
        planner_reasoning = ""
        planner_steps = [
            {
                "id": f"{command['command_id']}:catalog",
                "name": "workflow_planning_progress",
                "args": {"description": "正在读取可用智能体及能力描述"},
            },
            {
                "id": f"{command['command_id']}:design",
                "name": "workflow_planning_progress",
                "args": {"description": "规划智能体正在设计候选工作流"},
            },
            {
                "id": f"{command['command_id']}:validate",
                "name": "workflow_planning_progress",
                "args": {"description": "正在校验依赖关系并绑定智能体"},
            },
        ]
        try:
            yield _planner_sse("phase", {"id": "catalog", "label": "正在读取可用智能体及能力描述"})
            record = await store.get_session_record(session_id)
            if record is None:
                raise ReportCollaborationValidationError("NOT_FOUND", "会话不存在")
            requirement = requirement_from_session(record)
            cfg = get_app_config().report_collaboration
            catalog = await project_agent_catalog(
                user_id=user_id,
                config=cfg,
                agent_store=getattr(request.app.state, "agent_store", None),
            )
            history = [
                {"role": "human" if item.get("role") == "human" else "ai", "content": str(item.get("content") or "")}
                for item in await store.list_messages(session_id)
            ]
            current_plans = [
                item
                for item in await store.list_plans(session_id)
                if item.get("status") in {"proposed", "selected"}
            ]
            llm = service.planner_llm()
            binder = getattr(llm, "bind", None)
            if callable(binder):
                llm = binder(max_tokens=cfg.planner_max_tokens)
            planner = PlannerAgent(llm)
            messages = planner.workflow_messages(
                requirement=requirement,
                history=history,
                catalog=catalog,
                current_plans=current_plans,
                adjustment_text=body.adjustment_text,
                max_team_members=cfg.max_team_members,
            )
            yield _planner_sse("phase", {"id": "design", "label": "规划智能体正在设计候选工作流"})
            raw = ""
            inline_reasoning_length = 0
            planner_stream_error: Exception | None = None
            logger.info(
                "rc planner stream started session=%s timeout=%ss max_tokens=%s agents=%s",
                session_id,
                cfg.planner_timeout_seconds,
                cfg.planner_max_tokens,
                len(catalog.agents),
            )
            try:
                async with asyncio.timeout(cfg.planner_timeout_seconds):
                    async for chunk in llm.astream(messages):
                        content_delta, reasoning_delta = _planner_chunk_parts(chunk)
                        raw += content_delta
                        if reasoning_delta:
                            planner_reasoning += reasoning_delta
                            yield _planner_sse("reasoning", {"delta": reasoning_delta})
                        inline_reasoning = _planner_inline_reasoning(raw)
                        if len(inline_reasoning) > inline_reasoning_length:
                            inline_delta = inline_reasoning[inline_reasoning_length:]
                            planner_reasoning += inline_delta
                            yield _planner_sse(
                                "reasoning",
                                {"delta": inline_delta},
                            )
                            inline_reasoning_length = len(inline_reasoning)
            except TimeoutError as exc:
                planner_stream_error = exc
                logger.warning(
                    "rc planner stream timed out session=%s visible_chars=%s reasoning_chars=%s",
                    session_id,
                    len(raw),
                    len(planner_reasoning),
                )
            except Exception as exc:  # noqa: BLE001 - recover into executable plans below
                planner_stream_error = exc
                logger.warning(
                    "rc planner stream failed session=%s error=%s: %s",
                    session_id,
                    type(exc).__name__,
                    exc,
                )

            # Some reasoning providers place their final JSON in
            # reasoning_content instead of the visible content channel. Try
            # both channels, including truncated-output salvage.
            parsed = _extract_json(raw)
            if parsed is None:
                parsed = _extract_json(planner_reasoning)
            if parsed is None:
                parsed = _extract_json(f"{raw}\n{planner_reasoning}")
            designed = planner.parse_workflow_design(parsed)
            if designed is None:
                recovery_label = "规划输出较长，正在收敛为可执行候选工作流"
                planner_steps.append(
                    {
                        "id": f"{command['command_id']}:recover",
                        "name": "workflow_planning_progress",
                        "args": {"description": recovery_label},
                    }
                )
                yield _planner_sse("phase", {"id": "recover", "label": recovery_label})
                logger.warning(
                    "rc planner output could not be parsed; using constrained recovery session=%s visible_chars=%s reasoning_chars=%s stream_error=%s",
                    session_id,
                    len(raw),
                    len(planner_reasoning),
                    type(planner_stream_error).__name__ if planner_stream_error else "none",
                )
                strategies = PlanController().propose(
                    requirement,
                    max_parallel_tasks=cfg.max_parallel_tasks,
                    max_team_members=cfg.max_team_members,
                )
                designed = (
                    strategies,
                    "规划智能体已依据当前意图和可用智能体目录收敛出三套可执行候选工作流，可继续通过对话调整。",
                )
            else:
                logger.info(
                    "rc planner stream completed session=%s visible_chars=%s reasoning_chars=%s candidates=%s",
                    session_id,
                    len(raw),
                    len(planner_reasoning),
                    len(designed[0]),
                )
            yield _planner_sse("phase", {"id": "validate", "label": "正在校验依赖关系并绑定智能体"})
            await service.fulfill(
                session_id,
                user_id=user_id,
                command_id=command["command_id"],
                idempotency_key=key,
                adjustment_text=body.adjustment_text,
                precomputed_design=designed,
                planner_reasoning=planner_reasoning or None,
                planner_steps=planner_steps,
            )
            snapshot = await store.get_snapshot(session_id)
            yield _planner_sse("complete", {"snapshot": snapshot})
        except asyncio.CancelledError as exc:
            await service.fail_command(session_id, command["command_id"], exc)
            raise
        except Exception as exc:  # noqa: BLE001 - SSE errors must be returned in-band
            await service.fail_command(session_id, command["command_id"], exc)
            yield _planner_sse("error", {"message": str(exc)})

    return StreamingResponse(
        generate(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no", "Connection": "keep-alive"},
    )


@router.post("/sessions/{session_id}/plan-requests")
async def request_plans(
    request: Request,
    session_id: str,
    x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
    x_expected_revision: int | None = Header(default=None, alias="X-Expected-Revision"),
):
    if disabled := _require_enabled():
        return disabled
    user_id = await _require_user(request)
    key = _require_idempotency(x_idempotency_key)
    if isinstance(key, JSONResponse):
        return key
    store = _store(request)
    owned = await _owned_session(store, session_id, user_id)
    if isinstance(owned, JSONResponse):
        return owned
    if owned.get("selected_plan_id"):
        return _error(409, "PLAN_ALREADY_SELECTED", "会话已选定方案，无法重新规划")
    try:
        command = await store.enqueue_plan_request(session_id, idempotency_key=key, expected_revision=x_expected_revision)
        if command.get("status") == "pending":
            await PlanProposalService(
                store,
                agent_store=getattr(request.app.state, "agent_store", None),
                config=get_app_config().report_collaboration,
                llm_factory=getattr(request.app.state, "report_collaboration_planner_llm", None),
            ).fulfill(session_id, user_id=user_id, command_id=command["command_id"], idempotency_key=key)
        return Response(status_code=204)
    except Exception as exc:
        return _map_store_error(exc)


@router.get("/sessions/{session_id}/plans")
async def list_plans(request: Request, session_id: str):
    if disabled := _require_enabled():
        return disabled
    user_id = await _require_user(request)
    store = _store(request)
    owned = await _owned_session(store, session_id, user_id)
    if isinstance(owned, JSONResponse):
        return owned
    return {"plans": await store.list_plans(session_id)}


@router.post("/sessions/{session_id}/plans/{plan_id}/select")
async def select_plan(
    request: Request,
    session_id: str,
    plan_id: str,
    x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
    x_expected_revision: int | None = Header(default=None, alias="X-Expected-Revision"),
):
    if disabled := _require_enabled():
        return disabled
    user_id = await _require_user(request)
    key = _require_idempotency(x_idempotency_key)
    if isinstance(key, JSONResponse):
        return key
    store = _store(request)
    owned = await _owned_session(store, session_id, user_id)
    if isinstance(owned, JSONResponse):
        return owned
    try:
        result = await store.select_plan(session_id, plan_id, idempotency_key=key, expected_revision=x_expected_revision)
        return {"selected_plan_id": result["selected_plan_id"]}
    except Exception as exc:
        return _map_store_error(exc)


@router.post("/sessions/{session_id}/runs")
async def start_run(
    request: Request,
    session_id: str,
    body: StartRunBody,
    x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
    x_expected_revision: int | None = Header(default=None, alias="X-Expected-Revision"),
):
    if disabled := _require_enabled():
        return disabled
    user_id = await _require_user(request)
    key = _require_idempotency(x_idempotency_key)
    if isinstance(key, JSONResponse):
        return key
    store = _store(request)
    owned = await _owned_session(store, session_id, user_id)
    if isinstance(owned, JSONResponse):
        return owned
    try:
        plans = await store.list_plans(session_id)
        plan = next((item for item in plans if item.get("id") == body.plan_id), None)
        if plan is None:
            return _error(404, "NOT_FOUND", f"report collaboration plan not found: {body.plan_id}")
        bundle = freeze_plan_models(plan, get_app_config())
        cfg = get_app_config().report_collaboration
        run = await store.create_run(
            session_id,
            plan_id=body.plan_id,
            idempotency_key=key,
            expected_revision=x_expected_revision,
            role_snapshots=bundle.dump_by_role(),
            template_snapshot=await _frozen_run_template(request, owned, plan),
            budget_snapshot=freeze_run_budget(cfg),
            max_active_runs_per_owner=cfg.max_concurrent_runs_per_user,
        )
        existing_events = await store.list_events(run["id"], after_seq=0, limit=1)
        if not existing_events:
            await store.append_event(
                session_id=session_id,
                run_id=run["id"],
                event_type="heartbeat",
                data={
                    "audit": "model_snapshot",
                    "roles": [item.model_dump() for item in bundle.by_role.values()],
                    "fallbacks": [item.model_dump() for item in bundle.fallbacks()],
                },
            )
        return run
    except Exception as exc:
        return _map_store_error(exc)


@router.get("/sessions/{session_id}/reports")
async def list_reports(request: Request, session_id: str):
    if disabled := _require_enabled():
        return disabled
    user_id = await _require_user(request)
    store = _store(request)
    owned = await _owned_session(store, session_id, user_id)
    if isinstance(owned, JSONResponse):
        return owned
    return {"versions": await store.list_report_versions(session_id)}


@router.post("/sessions/{session_id}/report-rewrites")
async def create_rewrite(
    request: Request,
    session_id: str,
    body: CreateRewriteBody,
    x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
):
    if disabled := _require_enabled():
        return disabled
    user_id = await _require_user(request)
    key = _require_idempotency(x_idempotency_key)
    if isinstance(key, JSONResponse):
        return key
    store = _store(request)
    owned = await _owned_session(store, session_id, user_id)
    if isinstance(owned, JSONResponse):
        return owned
    try:
        return await _rewrite_service(store).create_candidate(
            session_id,
            idempotency_key=key,
            instruction=body.instruction,
            section=body.section,
        )
    except Exception as exc:
        return _map_store_error(exc)


@router.post("/sessions/{session_id}/report-rewrites/{rewrite_id}/apply")
async def apply_rewrite(
    request: Request,
    session_id: str,
    rewrite_id: str,
    x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
):
    if disabled := _require_enabled():
        return disabled
    user_id = await _require_user(request)
    key = _require_idempotency(x_idempotency_key)
    if isinstance(key, JSONResponse):
        return key
    store = _store(request)
    owned = await _owned_session(store, session_id, user_id)
    if isinstance(owned, JSONResponse):
        return owned
    try:
        version, _created = await _rewrite_service(store).apply_candidate(session_id, rewrite_id, idempotency_key=key)
        return version
    except Exception as exc:
        return _map_store_error(exc)


@router.post("/sessions/{session_id}/report-versions/{version_id}/restore")
async def restore_report(
    request: Request,
    session_id: str,
    version_id: str,
    x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
):
    if disabled := _require_enabled():
        return disabled
    user_id = await _require_user(request)
    key = _require_idempotency(x_idempotency_key)
    if isinstance(key, JSONResponse):
        return key
    store = _store(request)
    owned = await _owned_session(store, session_id, user_id)
    if isinstance(owned, JSONResponse):
        return owned
    try:
        version, _created = await _rewrite_service(store).restore(session_id, version_id, idempotency_key=key)
        return version
    except Exception as exc:
        return _map_store_error(exc)


@router.get("/runs/{run_id}")
async def get_run(request: Request, run_id: str):
    if disabled := _require_enabled():
        return disabled
    user_id = await _require_user(request)
    run = await _owned_run(_store(request), run_id, user_id)
    if isinstance(run, JSONResponse):
        return run
    return run


@router.get("/runs/{run_id}/events")
async def list_run_events(
    request: Request,
    run_id: str,
    after_seq: int = Query(default=0, ge=0),
    limit: int = Query(default=200, ge=1, le=1000),
):
    if disabled := _require_enabled():
        return disabled
    user_id = await _require_user(request)
    store = _store(request)
    run = await _owned_run(store, run_id, user_id)
    if isinstance(run, JSONResponse):
        return run
    return {"events": await store.list_events(run_id, after_seq=after_seq, limit=limit)}


@router.get("/runs/{run_id}/stream")
async def stream_run(
    request: Request,
    run_id: str,
    after_seq: int = Query(default=0, ge=0),
    last_event_id: str | None = Header(default=None, alias="Last-Event-ID"),
):
    if disabled := _require_enabled():
        return disabled
    user_id = await _require_user(request)
    store = _store(request)
    run = await _owned_run(store, run_id, user_id)
    if isinstance(run, JSONResponse):
        return run
    hub = getattr(request.app.state, "report_collaboration_live_hub", None)
    cursor = replay_cursor(last_event_id or request.headers.get("last-event-id"), after_seq)
    settings = getattr(request.app.state, "report_collaboration_sse_settings", None)

    async def generator():
        async def _disconnected() -> bool:
            return await request.is_disconnected()

        async for chunk in iter_run_sse(
            store=store,
            hub=hub,
            run_id=run_id,
            cursor=cursor,
            is_disconnected=_disconnected,
            settings=settings,
        ):
            yield chunk

    return StreamingResponse(
        generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache, no-transform",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",
            "X-Report-Collaboration-Run-Status": str(run.get("status") or ""),
        },
    )


@router.post("/runs/{run_id}/cancel")
async def cancel_run(
    request: Request,
    run_id: str,
    x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
):
    if disabled := _require_enabled():
        return disabled
    user_id = await _require_user(request)
    key = _require_idempotency(x_idempotency_key)
    if isinstance(key, JSONResponse):
        return key
    store = _store(request)
    run = await _owned_run(store, run_id, user_id)
    if isinstance(run, JSONResponse):
        return run
    try:
        return await store.request_cancel_run(run_id, idempotency_key=key)
    except Exception as exc:
        return _map_store_error(exc)


@router.post("/runs/{run_id}/commands")
async def create_command(
    request: Request,
    run_id: str,
    body: CreateCommandBody,
    x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
):
    if disabled := _require_enabled():
        return disabled
    user_id = await _require_user(request)
    key = _require_idempotency(x_idempotency_key)
    if isinstance(key, JSONResponse):
        return key
    store = _store(request)
    run = await _owned_run(store, run_id, user_id)
    if isinstance(run, JSONResponse):
        return run
    try:
        return await _runtime_commands(request).submit(
            run_id,
            text=body.text,
            idempotency_key=key,
            target_node_id=body.target_node_id,
            agent_run_id=body.agent_run_id,
        )
    except Exception as exc:
        return _map_store_error(exc)


@router.post("/runs/{run_id}/commands/{command_id}/confirm")
async def confirm_command(
    request: Request,
    run_id: str,
    command_id: str,
    x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
):
    if disabled := _require_enabled():
        return disabled
    user_id = await _require_user(request)
    key = _require_idempotency(x_idempotency_key)
    if isinstance(key, JSONResponse):
        return key
    store = _store(request)
    run = await _owned_run(store, run_id, user_id)
    if isinstance(run, JSONResponse):
        return run
    try:
        return await _runtime_commands(request).confirm(run_id, command_id, idempotency_key=key)
    except Exception as exc:
        return _map_store_error(exc)


@router.post("/runs/{run_id}/commands/{command_id}/cancel")
async def cancel_command(
    request: Request,
    run_id: str,
    command_id: str,
    x_idempotency_key: str | None = Header(default=None, alias="X-Idempotency-Key"),
):
    if disabled := _require_enabled():
        return disabled
    user_id = await _require_user(request)
    key = _require_idempotency(x_idempotency_key)
    if isinstance(key, JSONResponse):
        return key
    store = _store(request)
    run = await _owned_run(store, run_id, user_id)
    if isinstance(run, JSONResponse):
        return run
    try:
        return await _runtime_commands(request).cancel(run_id, command_id, idempotency_key=key)
    except Exception as exc:
        return _map_store_error(exc)


@router.get("/runs/{run_id}/nodes/{node_id}/contract")
async def get_node_contract(request: Request, run_id: str, node_id: str):
    if disabled := _require_enabled():
        return disabled
    user_id = await _require_user(request)
    store = _store(request)
    run = await _owned_run(store, run_id, user_id)
    if isinstance(run, JSONResponse):
        return run
    try:
        return await store.get_node_contract(run_id, node_id)
    except Exception as exc:
        return _map_store_error(exc)


@router.get("/runs/{run_id}/artifacts")
async def list_artifacts(request: Request, run_id: str):
    if disabled := _require_enabled():
        return disabled
    user_id = await _require_user(request)
    store = _store(request)
    run = await _owned_run(store, run_id, user_id)
    if isinstance(run, JSONResponse):
        return run
    return {"artifacts": await store.list_artifacts(run_id)}


@router.get("/runs/{run_id}/sources")
async def list_sources(request: Request, run_id: str):
    if disabled := _require_enabled():
        return disabled
    user_id = await _require_user(request)
    run = await _owned_run(_store(request), run_id, user_id)
    if isinstance(run, JSONResponse):
        return run
    return {"sources": []}