deerflow-code/offline-backend-20260512/backend/app/gateway/routers/roundtable_jobs.py

"""圆桌后台作业控制 API（Phase 3a）。

把 Step 2「后台挂起」的任务交给后端编排引擎跑，并暴露进度订阅 / 续跑 / 取消。

Routes（prefix ``/api/roundtable-jobs``）::
    POST   ""                 启动一个后台作业（创建 job 行 + 拉起执行器任务）
    GET    ""?draft_id=       查某草稿最近一个作业
    GET    "/{job_id}"        作业快照
    GET    "/{job_id}/stream" SSE 进度流（轮询 job 表，多 worker 安全）
    POST   "/{job_id}/resume" 回答澄清后续跑
    POST   "/{job_id}/cancel" 取消

进度数据源是 ``roundtable_jobs`` 表（Phase 1），SSE 走 DB 轮询而非内存 pub/sub —— 这样
刷新 / 换 worker / 重连都能继续收到进度，规避 ai_writing 踩过的「per-worker 内存 dict
丢会话」坑。响应统一 camelCase（对齐前端 §2.5）。
"""

from __future__ import annotations

import asyncio
import json
import logging
import time
from datetime import UTC, datetime, timedelta
from typing import Any
from uuid import uuid4

from fastapi import APIRouter, HTTPException, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field

from app.gateway.roundtable_job_executor import StartParams
from deerflow.agents.roundtable_orchestrator import SeatRef
from deerflow.runtime.user_context import get_effective_user_id

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

_TERMINAL = {"done", "error", "cancelled"}
# SSE 轮询间隔与最长存活（对齐 ai_writing：长连接最多挂 30 分钟）。
_POLL_INTERVAL_S = 0.7
_MAX_STREAM_TICKS = int(30 * 60 / _POLL_INTERVAL_S)

# 续跑澄清时拼给 leader 的 prompt 前缀。
_RESUME_PREFIX = "用户已回复你刚才的澄清问题，回答如下，请据此继续派活或给出结论：\n"


# ── schemas ──────────────────────────────────────────────────────────────────


class JobAgent(BaseModel):
    agentId: str
    name: str = ""


class StartJobRequest(BaseModel):
    draftId: str | None = Field(default=None, max_length=64)
    # 外部任务 id（taskId 深链会商）。非空 = task 作业：按 task_id 共享、不按 user 分权，
    # 其 draft_id 指向独立的 roundtable_task_drafts 表（最终报告写回那里）。
    taskId: str | None = Field(default=None, max_length=128)
    intent: str = ""
    agents: list[JobAgent] = Field(default_factory=list)
    model: str | None = None
    orchestrationMode: str = "recommend"
    chain: dict[str, Any] | None = None
    # dag 模式分层编排计划（前端 OrchestrationPlan：{mode,stages,finalSynthesis,coordinatorPrompt?}）。
    # 仅 orchestrationMode=="dag" 时携带；驱动后端 _run_dag。缺失 → 引擎回退 recommend。
    orchestrationPlan: dict[str, Any] | None = None
    # 第一轮 leader prompt；缺省用 intent。
    seedMessage: str | None = None
    # 预建好的 thread 映射（Phase 3b 真网关用；模拟网关忽略）。
    threadIds: dict[str, str] | None = None
    coordinatorName: str = "roundtable-coordinator"
    # 席位执行模式（快速/思考/专业问答/多智能体）派生参数，由前端 toStartBody 下发。
    # 仅 special 席位用：覆盖 roundtable_run_policy("seat") 的 thinking/reasoning；
    # None/缺省 → 后端按 policy 默认(≈pro)，向后兼容旧前端。subagent 仅 ultra 档为真。
    thinking_enabled: bool | None = None
    reasoning_effort: str | None = None
    subagent_enabled: bool = False
    # 每席位「技能识别强化」开关映射 {agent_id: bool}（前端读所选业务链条里**每个席位**的
    # 同名配置后下发）。后台席位 run 据此（与各席位 agent 自身 config.yaml 的开关取 OR）决定
    # 是否注入。缺省空 dict（全部按 False/关，仅各 agent 自身开关生效）。
    seat_skill_directives: dict[str, bool] = Field(default_factory=dict)
    # 每席位「推理深度覆盖」映射 {agent_id: SeatMode}（前端读所选业务链条里**每个席位**单独配的
    # 推理深度后下发；值为 flash/thinking/pro/ultra）。后台席位 run 按 agent_id 查表：命中则用该档
    # 派生的 thinking/reasoning/subagent 覆盖作业级 seatMode；未命中 → 跟随作业级。缺省空 dict。
    seat_modes: dict[str, str] = Field(default_factory=dict)
    # 是否研讨前先并行『取数』（业务链条逐条 opt-in；前端读所选链条的开关后下发）。默认 False。
    gather_first: bool = False
    # 深链接业务码（rwfx→6BF 等）。透传到后台总结报告的「业务链完整性核对」。普通会商为空。
    business_code: str | None = None


class ResumeJobRequest(BaseModel):
    answer: str = ""


class CleanupRequest(BaseModel):
    retention_days: int = Field(default=7, ge=0, description="保留多少天内的作业；0 = 不保留")
    only_finished: bool = Field(default=True, description="仅清终态（done/error/cancelled），保护在跑作业")
    batch_size: int = Field(default=500, ge=1, le=5000)


class CleanupResponse(BaseModel):
    deleted: int
    elapsed_seconds: float
    params: dict[str, Any]


class RoundtableJobResponse(BaseModel):
    id: str
    draftId: str | None = None
    status: str = "queued"
    phase: str = "initializing"
    cycle: int = 0
    dispatchChain: list[dict[str, Any]] = Field(default_factory=list)
    activeAgentId: str | None = None
    consensusPercentage: int = 0
    pendingClarification: str | None = None
    orchestrationMode: str = "recommend"
    chain: dict[str, Any] | None = None
    orchestrationPlan: dict[str, Any] | None = None
    error: str | None = None
    dialogues: list[dict[str, Any]] = Field(default_factory=list)
    step3: dict[str, Any] | None = None
    createdAt: datetime | str | None = None
    updatedAt: datetime | str | None = None


# ── helpers ──────────────────────────────────────────────────────────────────


def _current_user_id(request: Request) -> str:
    user = getattr(request.state, "user", None)
    if user is not None:
        return str(user.id)
    return get_effective_user_id()


def _get_store(request: Request):
    store = getattr(request.app.state, "roundtable_job_store", None)
    if store is None:
        raise HTTPException(status_code=503, detail="Roundtable job store not available")
    return store


def _get_executor(request: Request):
    executor = getattr(request.app.state, "roundtable_job_executor", None)
    if executor is None:
        raise HTTPException(status_code=503, detail="Roundtable job executor not available")
    return executor


def _pending_nodes(agents: list[JobAgent]) -> list[dict[str, Any]]:
    """启动即落一份「全 pending」的派活链路，使 GET / 首帧 SSE 立刻能渲染流程图。"""
    return [
        {"agentId": a.agentId, "name": a.name or a.agentId, "order": i + 1, "state": "pending"}
        for i, a in enumerate(agents)
    ]


def _to_response(row: dict[str, Any]) -> RoundtableJobResponse:
    return RoundtableJobResponse(
        id=row["id"],
        draftId=row.get("draft_id"),
        status=row.get("status") or "queued",
        phase=row.get("phase") or "initializing",
        cycle=int(row.get("cycle") or 0),
        dispatchChain=row.get("dispatch_chain") or [],
        activeAgentId=row.get("active_agent_id"),
        consensusPercentage=int(row.get("consensus_percentage") or 0),
        pendingClarification=row.get("pending_clarification"),
        orchestrationMode=row.get("orchestration_mode") or "recommend",
        chain=row.get("chain"),
        orchestrationPlan=row.get("orchestration_plan"),
        error=row.get("error"),
        dialogues=row.get("dialogues") or [],
        step3=row.get("step3"),
        createdAt=row.get("created_at"),
        updatedAt=row.get("updated_at"),
    )


def _seat_refs_from_nodes(nodes: list[dict[str, Any]] | None) -> list[SeatRef]:
    out: list[SeatRef] = []
    for n in nodes or []:
        aid = str(n.get("agentId") or "")
        if aid:
            out.append(SeatRef(aid, str(n.get("name") or aid)))
    return out


# ── routes ───────────────────────────────────────────────────────────────────


@router.post("", response_model=RoundtableJobResponse, status_code=201)
async def start_job(request: Request, body: StartJobRequest) -> RoundtableJobResponse:
    store = _get_store(request)
    executor = _get_executor(request)
    user_id = _current_user_id(request)

    if not body.agents:
        raise HTTPException(status_code=400, detail="agents is required")

    # 防重复挂起（服务端幂等闸门）：同一草稿已有未终态作业（queued/running/awaiting_input）
    # 时直接返回它、不再新建。前端的「手动挂起按钮 / 离开页面自动挂起 / beforeunload 刷新
    # 兜底」可能对同一研讨多次触发本端点 —— 客户端去重只能挡住同页内的重复，跨触发源 /
    # 跨标签页的并发只能靠这里兜住，否则同一研讨会被双跑。
    task_id = (body.taskId or "").strip() or None
    if body.draftId:
        # task 作业按 task_id 去重（跨用户共享）；个人作业按 user 去重。
        existing = await store.get_by_draft(body.draftId, user_id=user_id, task_id=task_id)
        if existing is not None and (existing.get("status") or "") not in _TERMINAL:
            return _to_response(existing)

    agents = [SeatRef(a.agentId, a.name or a.agentId) for a in body.agents]
    job_id = uuid4().hex
    row = await store.create(
        id=job_id,
        draft_id=body.draftId,
        user_id=user_id,
        task_id=task_id,
        orchestration_mode=body.orchestrationMode,
        coordinator_name=body.coordinatorName,
        thread_ids=body.threadIds,
        chain=body.chain,
        orchestration_plan=body.orchestrationPlan,
        dispatch_chain=_pending_nodes(body.agents),
        status="queued",
    )

    executor.start_job(
        StartParams(
            job_id=job_id,
            user_id=user_id,
            agents=agents,
            seed_message=(body.seedMessage or body.intent or "").strip(),
            model=body.model,
            mode=body.orchestrationMode,
            intent_text=body.intent,
            thread_ids=body.threadIds,
            coordinator_name=body.coordinatorName,
            draft_id=body.draftId,
            task_id=task_id,
            chain=body.chain,
            orchestration_plan=body.orchestrationPlan,
            seat_thinking_enabled=body.thinking_enabled,
            seat_reasoning_effort=body.reasoning_effort,
            seat_subagent_enabled=body.subagent_enabled,
            seat_skill_directives=dict(body.seat_skill_directives or {}),
            seat_modes=dict(body.seat_modes or {}),
            gather_first=bool(body.gather_first),
            business_code=(body.business_code or None),
        )
    )
    return _to_response(row)


@router.get("", response_model=RoundtableJobResponse | None)
async def get_job_by_draft(request: Request, draft_id: str, task_id: str | None = None) -> RoundtableJobResponse | None:
    # task_id 非空 = task 作业查询（按 task 共享、不分权）；否则按 user 分权。
    store = _get_store(request)
    tid = (task_id or "").strip() or None
    user_id = _current_user_id(request)
    row = await store.get_by_draft(draft_id, user_id=user_id, task_id=tid)
    return _to_response(row) if row is not None else None


@router.get("/{job_id}", response_model=RoundtableJobResponse)
async def get_job(request: Request, job_id: str, task_id: str | None = None) -> RoundtableJobResponse:
    store = _get_store(request)
    tid = (task_id or "").strip() or None
    user_id = _current_user_id(request)
    row = await store.get(job_id, user_id=user_id, task_id=tid)
    if row is None:
        raise HTTPException(status_code=404, detail="Job not found")
    return _to_response(row)


@router.get("/{job_id}/stream")
async def stream_job(request: Request, job_id: str, task_id: str | None = None) -> StreamingResponse:
    store = _get_store(request)
    tid = (task_id or "").strip() or None
    user_id = _current_user_id(request)

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

    async def gen():
        for _ in range(_MAX_STREAM_TICKS):
            if await request.is_disconnected():
                return
            row = await store.get(job_id, user_id=user_id, task_id=tid)
            if row is None:
                yield _frame({"event": "error", "detail": "Job not found"})
                return
            yield _frame({"event": "progress", **_to_response(row).model_dump(mode="json")})
            if (row.get("status") or "") in _TERMINAL:
                return
            await asyncio.sleep(_POLL_INTERVAL_S)

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


@router.post("/{job_id}/resume", response_model=RoundtableJobResponse)
async def resume_job(
    request: Request, job_id: str, body: ResumeJobRequest, task_id: str | None = None
) -> RoundtableJobResponse:
    store = _get_store(request)
    executor = _get_executor(request)
    tid = (task_id or "").strip() or None
    user_id = _current_user_id(request)

    row = await store.get(job_id, user_id=user_id, task_id=tid)
    if row is None:
        raise HTTPException(status_code=404, detail="Job not found")
    if (row.get("status") or "") != "awaiting_input":
        raise HTTPException(status_code=409, detail="Job is not awaiting input")

    answer = body.answer.strip()
    if not answer:
        raise HTTPException(status_code=400, detail="answer is required")

    # task 作业不分权：放开写入归属校验（user_id=None）；并以**原作业所有者**身份续跑，
    # 保证编排上下文 / 写回 task-draft 用一致的 user 上下文。
    write_user_id = None if tid else user_id
    run_user_id = row.get("user_id") or user_id
    updated = await store.update_progress(
        job_id, user_id=write_user_id, status="running", pending_clarification=None
    )

    executor.start_job(
        StartParams(
            job_id=job_id,
            user_id=run_user_id,
            agents=_seat_refs_from_nodes(row.get("dispatch_chain")),
            seed_message=_RESUME_PREFIX + answer,
            model=None,
            mode=row.get("orchestration_mode") or "recommend",
            intent_text=answer,
            thread_ids=row.get("thread_ids"),
            coordinator_name=row.get("coordinator_name") or "roundtable-coordinator",
            draft_id=row.get("draft_id"),
            task_id=tid or row.get("task_id"),
            is_resume=True,
            chain=row.get("chain"),
            orchestration_plan=row.get("orchestration_plan"),
        )
    )
    return _to_response(updated or row)


async def _require_admin(request: Request) -> None:
    """放行 admin / auth-disabled，其余 403（与 ai_writing / scheduled_tasks 同套语义）。"""
    from app.gateway.deps import get_optional_user_from_request

    user = await get_optional_user_from_request(request)
    if user is None:
        return
    if getattr(user, "system_role", None) != "admin":
        raise HTTPException(status_code=403, detail="圆桌作业清理管理接口仅限管理员")


@router.post("/admin/cleanup", response_model=CleanupResponse)
async def admin_cleanup(request: Request, body: CleanupRequest) -> CleanupResponse:
    """清理过期作业行（活跃作业受 only_finished 保护，永不误删）。

    分批 ``list_older_than`` → ``delete_by_ids``，避免大事务。``retention_days=0``
    一刷清空所有终态作业（调试用）。
    """
    await _require_admin(request)
    store = _get_store(request)
    cutoff = datetime.now(UTC) - timedelta(days=body.retention_days)

    t0 = time.perf_counter()
    deleted = 0
    while True:
        ids = await store.list_older_than(cutoff, only_finished=body.only_finished, limit=body.batch_size)
        if not ids:
            break
        deleted += await store.delete_by_ids(ids)
        if len(ids) < body.batch_size:
            break

    return CleanupResponse(
        deleted=deleted,
        elapsed_seconds=round(time.perf_counter() - t0, 3),
        params={"retention_days": body.retention_days, "only_finished": body.only_finished, "batch_size": body.batch_size},
    )


@router.post("/{job_id}/cancel", response_model=RoundtableJobResponse)
async def cancel_job(request: Request, job_id: str, task_id: str | None = None) -> RoundtableJobResponse:
    store = _get_store(request)
    executor = _get_executor(request)
    tid = (task_id or "").strip() or None
    user_id = _current_user_id(request)

    row = await store.get(job_id, user_id=user_id, task_id=tid)
    if row is None:
        raise HTTPException(status_code=404, detail="Job not found")

    executor.cancel(job_id)
    # task 作业不分权：放开写入归属校验。
    updated = await store.update_progress(job_id, user_id=(None if tid else user_id), status="cancelled")
    return _to_response(updated or row)