AgentLab/backend/app/api/conversations.py

"""会话：把一串问答串成一次对话。

在此之前"对话"只是前端内存里的一个数组——刷新页面就没了，而每一轮又各自
从零建图，所以"上一轮问过什么"在系统里无处可查。这个模块把它落到库里，
同时服务两件事：给用户看的历史列表，和给模型看的上下文（`_history_text`
被 copilot 拿去拼进 prompt）。

问数据的轮次由服务端写（docs/design/orchestration-round §1）：一轮是一个后台任务（app/chat/job.py），建图、运行、复核、
按复核重来、落库都在服务端，这一行的状态、运行 id、答案、复核、meta 由它写（app/chat/record.py 说明了每一项）；页面只发起
（POST /api/chat/turns）、只看。以前这一行是页面写的（开轮 → 回填），页面一关就没人收尾。开轮、回填这两个接口留着：画布助手的
对话和老页面还在用，回填也用来改标题这类事，但问数据的一轮不再依赖它们。每一轮带着它的任务现在怎样（TurnOut.job），列表判
「还在跑」先看任务，不再只靠 15 分钟没动静来猜。

对话按人隔离（D-d）：新建时记下建立人（created_by = actor_of）。开登录后 user 角色只列得出、读得到、改得动、删得掉
自己建的对话；created_by 为空的老对话（关登录时建的）只有 builder 以上看得到。应用编辑器的助手对话（kind=app）另要
这个应用的 manage。接受 conversation_id 的别的接口（助手的生成流、应用润色、从运行提取）按同一套规则核对：
ensure_conversation_access。关登录时一切照旧。
"""

from __future__ import annotations

from datetime import datetime, timedelta, timezone
from typing import Any

from fastapi import APIRouter, Depends, Header, HTTPException
from pydantic import BaseModel, Field
from sqlalchemy import and_, delete, func, select, update
from sqlalchemy.ext.asyncio import AsyncSession

from app.api.deps import actor_of, current_principal
from app.db.base import get_session, utcnow
from app.db.models import Approval, Conversation, ConversationTurn, Run

router = APIRouter(prefix="/api/conversations", tags=["conversations"])

#: 喂给模型的历史轮数上限。再多意义不大，而 prompt 是要花钱的
HISTORY_TURNS = 6
#: 单轮问答在上下文里的截断长度。一次查询结果可能有几千字，整段塞进去
#: 会把后面几轮挤掉——那还不如少说点、多留几轮
HISTORY_ANSWER_CHARS = 600


def title_from(question: str, limit: int = 24) -> str:
    """用第一问当标题。没有标题的会话在列表里是一排"新对话"，等于没有列表。"""
    text = " ".join((question or "").split())
    if not text:
        return "新对话"
    return text if len(text) <= limit else text[:limit] + "…"


# --------------------------------------------------------------------------
# 出入参
# --------------------------------------------------------------------------


class ConversationIn(BaseModel):
    kind: str = Field(default="chat", pattern="^(chat|canvas)$")
    workflow_id: str | None = None
    title: str = ""


class ConversationPatch(BaseModel):
    title: str | None = None
    archived: bool | None = None


class TurnIn(BaseModel):
    question: str = ""


class TurnPatch(BaseModel):
    """回填。全是可选——一轮可能在任何一步停下，能填多少填多少。"""

    question: str | None = None
    answer: str | None = None
    explanation: str | None = None
    graph: dict[str, Any] | None = None
    run_id: str | None = None
    status: str | None = Field(default=None, pattern="^(running|done|error)$")
    error: str | None = None
    review: dict[str, Any] | None = None
    #: 可信度元数据，前端整块写、整块读，后端不解读它的内容（列表状态除外，见 last_status）
    meta: dict[str, Any] | None = None


class TurnOut(BaseModel):
    id: str
    seq: int
    question: str
    answer: str
    explanation: str
    graph: dict[str, Any] | None = None
    run_id: str | None = None
    status: str
    error: str
    #: 复核结论。None = 这一轮没复核过，和「复核过、没发现问题」不是一回事
    review: dict[str, Any] | None = None
    #: 可信度元数据。老轮次挂在 review.meta 下，读出来时统一放到这里
    meta: dict[str, Any] | None = None
    created_at: Any = None
    #: 最后一次落库。列表判「建图断了」按它算（BUILD_STALE），前端用同一只钟
    updated_at: Any = None
    #: 这一轮在服务端的任务现在怎样（app/chat/job.job_view：status、stage、last_seq、attempt_seq……）。老轮次没有任务，为 None
    job: dict[str, Any] | None = None

    model_config = {"from_attributes": True}


class ConversationOut(BaseModel):
    id: str
    title: str
    kind: str
    workflow_id: str | None = None
    #: 应用编辑器的助手对话（kind=app）归属的应用
    app_id: str | None = None
    #: 谁建的。关登录时是自报的署名；老对话为空
    created_by: str | None = None
    archived: bool
    created_at: Any = None
    last_active_at: Any = None
    turn_count: int = 0
    #: 列表里给个副标题，让人一眼认出是哪次聊天
    last_question: str = ""
    #: 最后一轮停在哪：running | waiting | error | cancelled | rejected | suspended | done，
    #: 没有轮次时为 None。见 last_status()
    last_status: str | None = None
    #: 最后一轮的运行。建图阶段就断了的没有
    last_run_id: str | None = None

    model_config = {"from_attributes": True}


class ConversationDetail(ConversationOut):
    turns: list[TurnOut] = Field(default_factory=list)

    # from_attributes 不能直接吃 Conversation：pydantic 会去读同名的 turns
    # 关系属性，而那是个未加载的 lazy relationship，在异步上下文里一读就是
    # MissingGreenlet。turns 一律显式传进来
    model_config = {"from_attributes": False}


def turn_meta(turn: Any) -> dict[str, Any] | None:
    """这一轮的元数据：meta 列优先，老数据退回 review.meta。"""
    if isinstance(turn.meta, dict):
        return turn.meta
    bridged = turn.review.get("meta") if isinstance(turn.review, dict) else None
    return bridged if isinstance(bridged, dict) else None


def turn_run_id(turn: Any) -> str | None:
    """这一轮现在的运行。meta 里写了 runId 就以它为准，哪怕是 null：重试会换一次运行，
    run_id 列却清不掉（回填里 None 表示「不改」），还指着上一次的。老轮次没有这个键，看列。"""
    meta = turn_meta(turn) or {}
    if "runId" in meta:
        return meta["runId"] or None
    return turn.run_id or None


def _turn_out(turn: ConversationTurn, job: Any = None) -> TurnOut:
    from app.chat.job import job_view

    out = TurnOut.model_validate(turn)
    out.meta = turn_meta(turn)
    out.job = job_view(job) if job is not None else None
    return out


async def _jobs(session: AsyncSession, turn_ids: list[str]) -> dict[str, Any]:
    """这几轮在服务端的任务（一次查完）。"""
    from app.db.models_chat import ChatTurnJob

    if not turn_ids:
        return {}
    rows = (await session.execute(select(ChatTurnJob).where(ChatTurnJob.id.in_(turn_ids)))).scalars()
    return {j.id: j for j in rows}


async def turn_out(session: AsyncSession, turn: ConversationTurn) -> TurnOut:
    """一轮（带着它的任务），同 GET /api/conversations/{id} 里的一轮：服务端跑的一轮收尾时把它放进 end 事件。"""
    return _turn_out(turn, (await _jobs(session, [turn.id])).get(turn.id))


def _detail(row: Conversation, turns: list[ConversationTurn], jobs: dict[str, Any] | None = None) -> ConversationDetail:
    base = ConversationOut.model_validate(row)
    return ConversationDetail(
        **base.model_dump(),
        turns=[_turn_out(t, (jobs or {}).get(t.id)) for t in turns],
    )


#: 还没有运行的轮次停在 running 超过这么久，就当建图断了。建图跟着浏览器那条流走，
#: 页面一关就停，轮次却一直是 running；最长的正常情况是模型超时再加两轮自查修正
BUILD_STALE = timedelta(minutes=15)


def last_status(
    turn: Any, run_status: str | None, pending: bool, now: datetime | None = None, end_reason: str | None = None,
    job_stage: str | None = None,
) -> str:
    """一轮停在哪。轮次状态只有 running / done / error 三种，细分要看运行和元数据。

    job_stage：这一轮在服务端的任务还在跑时它在哪一步（app/chat/job.py；没有任务、任务不在跑时为 None）。任务在跑就是在跑
    （停在审批上的是 waiting），不按 15 分钟没动静去猜——建图慢的一轮，那一行要到建完才改。

    没有任务的老轮次是页面写的：页面关掉、服务重启时它停在 running，这时要以运行的状态为准；
    运行也没有的，是断在了建图阶段。已经收尾的轮次以轮次自己为准——它写下的就是
    用户看到的结局（取消、服务重启挂起在 error 里分开记）。

    被审批人驳回结束的（运行 cancelled + end_reason=rejected，05-已知遗留 (a)）单独记 rejected，不和手动停止混成
    cancelled：前端落库时 meta 里记着 rejected（outcome 照旧是 cancelled），页面关掉没落库的看运行的 end_reason。
    """
    meta = turn_meta(turn) or {}
    outcome = meta.get("outcome")
    if job_stage is not None and turn.status == "running":
        return "waiting" if job_stage == "waiting" else "running"
    if turn.status == "done":
        return "done"
    if turn.status == "error":
        if outcome == "cancelled" and meta.get("rejected"):
            return "rejected"
        if outcome == "cancelled" or (turn.error or "").strip() == "已取消":
            return "cancelled"
        if outcome == "suspended":
            # 挂起之后可能已经在别处接着跑了
            live = _from_run(run_status, pending)
            return live if live in ("running", "waiting") else "suspended"
        return "error"
    live = _from_run(run_status, pending, end_reason)
    if live:
        return live
    updated = turn.updated_at
    if updated is not None and updated.tzinfo is None:
        updated = updated.replace(tzinfo=timezone.utc)
    stale = updated is not None and (now or utcnow()) - updated > BUILD_STALE
    return "error" if stale else "running"


def _from_run(run_status: str | None, pending: bool, end_reason: str | None = None) -> str | None:
    if run_status == "cancelled" and end_reason == "rejected":
        return "rejected"
    return {
        "queued": "running", "running": "running",
        "interrupted": "waiting" if pending else "suspended",
        "succeeded": "done", "failed": "error", "cancelled": "cancelled",
    }.get(run_status or "")


async def _last_turns(
    session: AsyncSession, ids: list[str]
) -> dict[str, ConversationTurn]:
    """每个会话 seq 最大的那一轮。一次查完，列表长了也只是一次往返。"""
    latest = (
        select(ConversationTurn.conversation_id, func.max(ConversationTurn.seq).label("seq"))
        .where(ConversationTurn.conversation_id.in_(ids))
        .group_by(ConversationTurn.conversation_id)
        .subquery()
    )
    rows = (await session.execute(
        select(ConversationTurn).join(latest, and_(
            ConversationTurn.conversation_id == latest.c.conversation_id,
            ConversationTurn.seq == latest.c.seq,
        ))
    )).scalars()
    return {t.conversation_id: t for t in rows}


async def _statuses(
    session: AsyncSession, turns: dict[str, ConversationTurn]
) -> dict[str, tuple[str, str | None]]:
    """会话 id → (last_status, last_run_id)。运行状态和待审批各一次查询。"""
    run_of = {cid: turn_run_id(t) for cid, t in turns.items()}
    run_ids = {r for r in run_of.values() if r}
    runs: dict[str, str] = {}
    ended: dict[str, str | None] = {}
    waiting: set[str] = set()
    if run_ids:
        rows = (await session.execute(
            select(Run.id, Run.status, Run.end_reason).where(Run.id.in_(run_ids))
        )).all()
        runs = {r: st for r, st, _ in rows}
        ended = {r: reason for r, _, reason in rows}
        stopped = [r for r, st in runs.items() if st == "interrupted"]
        if stopped:
            waiting = set((await session.execute(
                select(Approval.run_id).where(
                    Approval.run_id.in_(stopped), Approval.status == "pending")
            )).scalars())
    from app.core.job_runtime import ACTIVE

    jobs = await _jobs(session, [t.id for t in turns.values()])
    now = utcnow()
    out: dict[str, tuple[str, str | None]] = {}
    for cid, turn in turns.items():
        run_id = run_of[cid] if run_of[cid] in runs else None
        job = jobs.get(turn.id)
        stage = (job.stage or "build") if job is not None and job.status in ACTIVE else None
        out[cid] = (last_status(turn, runs.get(run_id or ""), run_id in waiting, now,
                                end_reason=ended.get(run_id or ""), job_stage=stage),
                    run_id)
    return out


# --------------------------------------------------------------------------
# 会话
# --------------------------------------------------------------------------


#: 对话按人隔离时拒绝的说法
NOT_YOURS = "没有权限查看这段对话：开启登录后，使用者只能查看自己建立的对话"
APP_MANAGE_ONLY = "没有权限查看这段对话：应用编辑器的助手对话只对能管理这个应用的人开放"


def _isolated() -> str | None:
    """开登录后的 user 角色：只看得到自己建的对话，交回它的用户名；关登录、builder 以上为 None（不按人收窄）。"""
    from app.api.deps import restricted

    return (current_principal().username or "") if restricted() else None


async def _manages_app(session: AsyncSession, app_id: str) -> bool:
    """这个人能不能管理这个应用（app/apps/access.py：关登录、admin 都能；builder 要是建立人或有 manage 授权）。应用已经
    删掉的，只有 admin 看得到它留下的对话。"""
    from app.apps.access import app_role, at_least
    from app.db.models import App

    principal = current_principal()
    if principal.mode == "off" or principal.role == "admin":
        return True
    app = await session.get(App, app_id)
    return app is not None and at_least(await app_role(session, principal, app), "manage")


async def _check_access(session: AsyncSession, row: Conversation) -> None:
    """这段对话这个人能不能用，不能用抛 403。蓝图对话跟着蓝图走（blueprints/store.ensure_access）：别人的个人蓝图的对话，
    搭建者、管理员也看不到（§3.12：别人的个人应用内容谁都不看），当它不存在（404，同蓝图）。"""
    mine = _isolated()
    if mine is not None and (not row.created_by or row.created_by != mine):
        raise HTTPException(403, NOT_YOURS)
    if row.kind == "app" and row.app_id and not await _manages_app(session, row.app_id):
        raise HTTPException(403, APP_MANAGE_ONLY)
    if row.kind == "blueprint":
        from app.blueprints.store import BLUEPRINT_GONE, ensure_access
        from app.db.models_blueprints import Blueprint

        bp = (await session.execute(select(Blueprint).where(Blueprint.conversation_id == row.id))).scalars().first()
        if bp is None:
            raise HTTPException(404, BLUEPRINT_GONE)
        await ensure_access(session, bp)


async def ensure_conversation_access(session: AsyncSession, conversation_id: str | None) -> Conversation | None:
    """接受 conversation_id 的接口（助手的 generate-stream、generate、app-stream、from-run……）在读历史、记来源之前调。

    能用交回对话；没给 id、对话不存在交回 None——和加这道核对之前一样当没有历史，前端带着一个刚删掉的对话 id 也不至于
    整轮失败；存在但这个人不能用（开登录后 user 角色碰别人的或老对话，没有 manage 的人碰应用的助手对话）回 403，免得
    别人的问答经「历史」拼进这一轮的提示、再从回答里带出来。关登录时只要存在就能用。"""
    if not conversation_id:
        return None
    row = await session.get(Conversation, conversation_id)
    if row is None:
        return None
    await _check_access(session, row)
    return row


async def _get(session: AsyncSession, conversation_id: str) -> Conversation:
    row = await session.get(Conversation, conversation_id)
    if not row:
        raise HTTPException(404, "对话不存在，可能已被删除")
    await _check_access(session, row)
    return row


@router.get("", response_model=list[ConversationOut])
async def list_conversations(
    kind: str = "chat",
    workflow_id: str | None = None,
    app_id: str | None = None,
    include_archived: bool = False,
    session: AsyncSession = Depends(get_session),
) -> list[ConversationOut]:
    query = select(Conversation).where(Conversation.kind == kind)
    if workflow_id:
        query = query.where(Conversation.workflow_id == workflow_id)
    if app_id:
        query = query.where(Conversation.app_id == app_id)
    if not include_archived:
        query = query.where(Conversation.archived.is_(False))
    if (mine := _isolated()) is not None:
        # 开登录后的 user 角色：只列自己建的；建立人为空的老对话不列
        query = query.where(Conversation.created_by == mine)
    rows = list((await session.execute(
        query.order_by(Conversation.last_active_at.desc(), Conversation.id.desc())
    )).scalars())
    # 应用的助手对话只列能管理的应用的（一个应用一次判断）
    managed: dict[str, bool] = {}
    kept = []
    for row in rows:
        if row.kind == "app" and row.app_id:
            if row.app_id not in managed:
                managed[row.app_id] = await _manages_app(session, row.app_id)
            if not managed[row.app_id]:
                continue
        if row.kind == "blueprint":
            # 蓝图对话跟着蓝图走：别人的个人蓝图的对话不列
            try:
                await _check_access(session, row)
            except HTTPException:
                continue
        kept.append(row)
    rows = kept
    if not rows:
        return []

    # 轮数、最后一轮、它的运行状态一次查完。每条会话各查一次的话，列表长了就是
    # 几十次往返。状态要带上：没打开过的会话也得标得出「在等审批」「失败了」
    ids = [r.id for r in rows]
    counts = dict((await session.execute(
        select(ConversationTurn.conversation_id, func.count(ConversationTurn.id))
        .where(ConversationTurn.conversation_id.in_(ids))
        .group_by(ConversationTurn.conversation_id)
    )).all())
    last = await _last_turns(session, ids)
    status = await _statuses(session, last)

    out: list[ConversationOut] = []
    for row in rows:
        item = ConversationOut.model_validate(row)
        item.turn_count = int(counts.get(row.id, 0))
        if row.id in last:
            item.last_question = last[row.id].question or ""
            item.last_status, item.last_run_id = status[row.id]
        out.append(item)
    return out


@router.post("", response_model=ConversationDetail, status_code=201)
async def create_conversation(
    payload: ConversationIn,
    session: AsyncSession = Depends(get_session),
    x_actor: str | None = Header(default=None),
) -> ConversationDetail:
    # 应用编辑器的助手对话不从这里建：GET /api/apps/{id}/assistant-conversation 先核对 manage，再取或建
    row = Conversation(
        kind=payload.kind,
        workflow_id=payload.workflow_id,
        title=payload.title or "新对话",
        created_by=actor_of(x_actor),
    )
    session.add(row)
    await session.commit()
    await session.refresh(row)
    return _detail(row, [])


@router.get("/{conversation_id}", response_model=ConversationDetail)
async def get_conversation(
    conversation_id: str, session: AsyncSession = Depends(get_session)
) -> ConversationDetail:
    return await conversation_detail(session, await _get(session, conversation_id))


async def conversation_detail(session: AsyncSession, row: Conversation) -> ConversationDetail:
    """一段对话连同各轮（权限由调用方先判过）。蓝图对话另经 GET /api/blueprints/{id}/conversation 取：那条路按蓝图判，
    没开问数据的使用者也读得到自己蓝图的对话。"""
    conversation_id = row.id
    turns = list((await session.execute(
        select(ConversationTurn)
        .where(ConversationTurn.conversation_id == conversation_id)
        .order_by(ConversationTurn.seq)
    )).scalars())
    out = _detail(row, turns, await _jobs(session, [t.id for t in turns]))
    out.turn_count = len(turns)
    out.last_question = turns[-1].question if turns else ""
    if turns:
        out.last_status, out.last_run_id = (await _statuses(session, {row.id: turns[-1]}))[row.id]
    return out


@router.patch("/{conversation_id}", response_model=ConversationOut)
async def patch_conversation(
    conversation_id: str,
    payload: ConversationPatch,
    session: AsyncSession = Depends(get_session),
) -> ConversationOut:
    row = await _get(session, conversation_id)
    if payload.title is not None:
        row.title = payload.title.strip() or row.title
    if payload.archived is not None:
        row.archived = payload.archived
    await session.commit()
    await session.refresh(row)
    return ConversationOut.model_validate(row)


@router.delete("/{conversation_id}", status_code=204)
async def delete_conversation(
    conversation_id: str, session: AsyncSession = Depends(get_session)
) -> None:
    from app.chat import job as chat_job

    row = await _get(session, conversation_id)
    # 正在服务端跑的几轮先停下（取消各自的运行）：不然删掉之后它还在跑、还在计费，收尾时又找不到那一行
    await chat_job.stop_conversation(conversation_id)
    # 显式删 turns 而不是只靠 ondelete=CASCADE：SQLite 的外键级联要 PRAGMA
    # 打开才生效（base.py 里是打开的），但这条路径不该依赖一个连接级的开关。任务和事件日志同理
    await chat_job.forget(session, conversation_id)
    await session.execute(
        delete(ConversationTurn).where(ConversationTurn.conversation_id == conversation_id)
    )
    await session.delete(row)
    await session.commit()


# --------------------------------------------------------------------------
# 轮次
# --------------------------------------------------------------------------


@router.post("/{conversation_id}/turns", response_model=TurnOut, status_code=201)
async def create_turn(
    conversation_id: str,
    payload: TurnIn,
    session: AsyncSession = Depends(get_session),
) -> TurnOut:
    conversation = await _get(session, conversation_id)
    turn = await new_turn(session, conversation, payload.question)
    await session.commit()
    await session.refresh(turn)
    return _turn_out(turn)


async def new_turn(session: AsyncSession, conversation: Conversation, question: str) -> ConversationTurn:
    """开一轮（status=running，只有问题），不提交：开轮接口和服务端发起的一轮（app/chat/job.start）共用。

    先记活跃时间、再数下一轮排第几：记活跃时间是这个事务的第一句写，占住这段对话（SQLite 上拿到写锁、PostgreSQL 上锁住
    这一行），同时开的两轮一个接一个数，不会都拿到同一个 seq。"""
    now = utcnow()
    await session.execute(update(Conversation).where(Conversation.id == conversation.id).values(last_active_at=now)
                          .execution_options(synchronize_session=False))
    next_seq = int((await session.execute(
        select(func.coalesce(func.max(ConversationTurn.seq), -1))
        .where(ConversationTurn.conversation_id == conversation.id)
    )).scalar_one()) + 1

    turn = ConversationTurn(
        conversation_id=conversation.id, seq=next_seq,
        question=question, status="running",
    )
    session.add(turn)

    conversation.last_active_at = now
    # 第一轮的问题就是这次对话的标题。用户改过名就不要再覆盖回去
    if next_seq == 0 and conversation.title in ("", "新对话"):
        conversation.title = title_from(question)
    await session.flush()
    return turn


async def turn_access(session: AsyncSession, turn_id: str) -> tuple[Conversation, ConversationTurn]:
    """这一轮这个人能不能用（按它所属的对话判，规则同 _get：对话按人隔离）。不存在 404，不能用 403。"""
    turn = await session.get(ConversationTurn, turn_id)
    if turn is None:
        raise HTTPException(404, "该轮对话不存在，可能已被删除")
    return await _get(session, turn.conversation_id), turn


@router.patch("/{conversation_id}/turns/{turn_id}", response_model=TurnOut)
async def patch_turn(
    conversation_id: str,
    turn_id: str,
    payload: TurnPatch,
    session: AsyncSession = Depends(get_session),
) -> TurnOut:
    await _get(session, conversation_id)
    turn = await session.get(ConversationTurn, turn_id)
    if not turn or turn.conversation_id != conversation_id:
        raise HTTPException(404, "该轮对话不存在，可能已被删除")

    for field in (
        "question", "answer", "explanation", "graph", "run_id", "status", "error", "review", "meta",
    ):
        value = getattr(payload, field)
        if value is not None:
            setattr(turn, field, value)

    # 活跃时间只认人在这里开始做事：改问题、按下运行、重试（状态回到 running）。
    # 打开会话时前端的回源核对也走这里——补全截断的答案、记下运行已删、把断掉的
    # 建图写回 error、补交付错过的结果——这些不算，否则旧会话一打开就顶到列表最上面
    if payload.question is not None or payload.run_id is not None or payload.status == "running":
        conversation = await session.get(Conversation, conversation_id)
        if conversation:
            conversation.last_active_at = utcnow()
    await session.commit()
    await session.refresh(turn)
    return _turn_out(turn)


# --------------------------------------------------------------------------
# 给模型看的那一份
# --------------------------------------------------------------------------


def _answer_of(turn: ConversationTurn) -> str:
    """这一轮的结果。问数据页是 answer，画布那条路径上只有 explanation。"""
    text = (turn.answer or turn.explanation or "").strip()
    if turn.status == "error" and turn.error:
        return f"（这一轮失败了：{turn.error[:200]}）"
    if len(text) > HISTORY_ANSWER_CHARS:
        text = text[:HISTORY_ANSWER_CHARS] + "…（略）"
    # 复核判过不可信的答案，下一轮必须知道。只给正文的话，模型会拿一个已知
    # 有问题的结论当事实继续往下推，错误就这么一轮轮传下去了
    note = (turn.review or {}).get("note", "") if isinstance(turn.review, dict) else ""
    if note:
        text = f"{text}\n（复核提示：{note[:200]}）".strip()
    return text


async def recent_turns(
    session: AsyncSession, conversation_id: str, limit: int = HISTORY_TURNS
) -> list[ConversationTurn]:
    """最近 limit 轮，按时间正序。当前这一轮（还没有答案）不算。"""
    rows = list((await session.execute(
        select(ConversationTurn)
        .where(ConversationTurn.conversation_id == conversation_id)
        .where(ConversationTurn.status != "running")
        .order_by(ConversationTurn.seq.desc())
        .limit(limit)
    )).scalars())
    return list(reversed(rows))


#: 运行入口的 history 里每轮的答案截到多少字（同以前问数据页 historyOf 的 slice(0, 600)）
RUN_HISTORY_CHARS = 600


async def run_history(session: AsyncSession, conversation_id: str | None, *, before_seq: int,
                      limit: int = HISTORY_TURNS) -> str:
    """问数据一轮发起运行时填进入口的 history（`{{ input.history }}`）：只取这一轮之前的（seq < before_seq），照以前问数据页
    historyOf 的规则——答案没有就用说明，都没有的跳过，每轮截 600 字，最多 limit 轮，「问：…\n答：…」之间空一行。

    以前这一份是页面按它内存里的轮次拼的；一轮搬到服务端之后在这里拼。和 history_text 不一样的地方是只看这一轮之前的：
    重试、重新运行较早的一轮时，后面接着问过的几轮不该进它的入口——那时候它还没被问过。"""
    if not conversation_id:
        return ""
    rows = list((await session.execute(
        select(ConversationTurn.question, ConversationTurn.answer, ConversationTurn.explanation)
        .where(ConversationTurn.conversation_id == conversation_id, ConversationTurn.seq < before_seq)
        .order_by(ConversationTurn.seq)
    )).all())
    lines = []
    for question, answer, explanation in rows:
        text = answer or explanation or ""
        if question and text:
            lines.append(f"问：{question}\n答：{text[:RUN_HISTORY_CHARS]}")
    return "\n\n".join(lines[-limit:])


async def history_text(
    session: AsyncSession, conversation_id: str | None, limit: int = HISTORY_TURNS
) -> str:
    """纯文本的问答历史，给运行时的图用（`{{ input.history }}`）。

    管的是指代：用户问"它们平均多少"，图里的 agent 得知道"它们"是上一轮
    那 12 家店。Copilot 侧的上下文解决不了这个——那只影响图长什么样，
    不影响图跑起来的时候模型看到什么。
    """
    if not conversation_id:
        return ""
    turns = await recent_turns(session, conversation_id, limit)
    if not turns:
        return ""
    lines = [f"问：{t.question}\n答：{_answer_of(t)}" for t in turns if t.question]
    return "\n\n".join(lines)