AgentLab/backend/app/chat/record.py

"""问数据一轮落进 conversation_turns 的那一行：服务端写（状态、运行 id、答案、复核、meta），页面只读。

以前这一行是页面写的（frontend/src/store/chat.ts 的 persist / buildMeta），一轮的编排搬到服务端之后由这里写，字段和页面以前
写的一一对应，读的一方（问数据页的 restoreTurn、助手拼历史的 copilot._history_section、列表的 conversations.last_status）不用变：

- status：running | done | error；error 列是一句原话（运行的报错、助手的说法、「已取消」「服务重启，本轮已中断」），助手拼
  历史时照它写「这一轮失败了：…」；
- answer：答案正文（成果各字段拼起来，同页面的 answerText，上限 ANSWER_CAP）；复核改写了的存改写后的；
- review：复核原样（含 verdict=ok、usage、retry_instruction）；没复核过是 {}；
- meta（v=1）：runId、runClass、runStatus、issuance（瘦过的出具信息）、noQuery、outcome（cancelled | suspended）、rejected
  （驳回的人、哪一步、备注：run.cancelled 的原样）、cause（失败的原因，见下）、attempts（用户点重试之前的各次）、ms、queries、
  scope、asked、looks、evidence、usage（{copilot, review, earlier, run}）、at。

**失败只记原因（cause），不记说法。** 页面以前记的是翻成人话之后的 failure（标题、原因、怎么办：lib/explain 那一套），说法在
界面层；服务端只记发生了什么，页面恢复时照同一套 lib/explain 讲（frontend/src/store/chat.ts 的 failureFromCause）：
  {kind: refused, message, code}        开流之前就被拒（模型没配好、限定的数据源都不在）
  {kind: copilot, message, hint, detail, code?, diagnostics?} 生成过程交出 error（助手本轮未完成；一个操作都没给的带
                                        code=copilot_empty 和诊断 copilot_gen.StreamDiagnosis.as_dict，画布轮次同样记）
  {kind: blocked, issues, self_checked} 生成的图没过检查
  {kind: start, message, code, status}  发起运行被拒（绑定的工具不在、使用者用不了的节点）
  {kind: run, error, detail}            运行失败
  {kind: gone}                          运行记录不见了
  {kind: build_interrupted}             建图时服务重启，没有可恢复的断点
  {kind: crash, message}                这一轮的后台任务自己出错
老轮次的 meta.failure 照旧认（页面优先读它）。
"""
from __future__ import annotations

import json
import time
from collections.abc import Mapping
from typing import Any

#: 落库答案的上限（同页面的 ANSWER_CAP）：只是防跑飞，不是显示上限
ANSWER_CAP = 20_000
ANSWER_CLIPPED = "\n\n…（内容过长，已截断，完整内容请查看运行记录）"
#: 归档的旧答案存多少（同页面的 ATTEMPT_ANSWER_CAP）
ATTEMPT_ANSWER_CAP = 4000
#: 停止、服务重启挂起时 error 列里的那一句（同页面落库时写的）
CANCELLED = "已取消"
SUSPENDED = "服务重启，本轮已中断"
#: 建图时服务重启
BUILD_INTERRUPTED = "服务重启，本轮在生成工作流时中断"
GONE = "运行记录不见了"
#: 用量里记的几项（同页面的 USAGE_KEYS）
USAGE_KEYS = ("cost_usd", "unpriced_calls", "calls", "unfinished_calls", "input_tokens", "output_tokens")
#: 出具信息里无法回指的数字最多存几个（同页面的 slimIssuance）
ISSUANCE_UNMATCHED_CAP = 20


def answer_text(output: Mapping[str, Any] | None) -> str:
    """成果压成一句话（同页面的 answerText）：下划线开头的、空的不要；字符串原样，别的按 JSON；超过上限截断并说出来。"""
    if not output:
        return ""
    parts = [value if isinstance(value, str) else json.dumps(value, ensure_ascii=False, separators=(",", ":"))
             for key, value in output.items() if not str(key).startswith("_") and value not in (None, "")]
    text = "\n".join(parts)
    return text[:ANSWER_CAP] + ANSWER_CLIPPED if len(text) > ANSWER_CAP else text


def slim_issuance(issuance: Any) -> dict[str, Any] | None:
    """出具信息存进 meta 前瘦一下身（同页面的 slimIssuance）：无法回指的数字只留前 20 个。"""
    if not isinstance(issuance, Mapping) or not issuance.get("tier"):
        return None
    out = dict(issuance)
    unmatched = issuance.get("unmatched_numbers")
    if isinstance(unmatched, list):
        out["unmatched_numbers"] = unmatched[:ISSUANCE_UNMATCHED_CAP]
    return out


def has_evidence(output: Mapping[str, Any] | None) -> bool:
    """成果带不带逐段证据（同页面 lib/evidence 的 hasEvidence）：_evidence 或它的 others 里有文档、有字段。"""
    ev = (output or {}).get("_evidence") if isinstance(output, Mapping) else None
    if not isinstance(ev, Mapping):
        return False
    refs = [ev, *[r for r in ev.get("others") or [] if isinstance(r, Mapping)]]
    return any(isinstance(r.get("doc_artifact"), str) and r.get("doc_artifact") and isinstance(r.get("fields"), list)
               and r.get("fields") for r in refs)


def _names(value: Any) -> list[str]:
    return [str(v) for v in value if v not in (None, "")] if isinstance(value, list) else []


def _num(value: Any) -> float | None:
    return value if isinstance(value, (int, float)) and not isinstance(value, bool) else None


def look_event(op: Mapping[str, Any]) -> dict[str, Any] | None:
    """助手的 look 操作 → 存进 meta.looks 的那一份（同页面 run/decode 的 lookEventOf）：下一轮服务端据此写「上一轮查看过」。"""
    if op.get("op") != "look":
        return None
    listed = op.get("objects") if isinstance(op.get("objects"), list) else op.get("sources")
    if not isinstance(listed, list):
        return None
    objects = []
    for x in listed:
        if not isinstance(x, Mapping):
            continue
        kind = x.get("kind") if x.get("kind") in ("capability", "workflow_version", "caliber") else "table"
        total, explored = _num(x.get("total")), _num(x.get("explored"))
        obj: dict[str, Any] = {"kind": kind}
        if isinstance(x.get("source"), str) and x.get("source"):
            obj["source"] = x["source"]
        if isinstance(x.get("source_id"), str) and x.get("source_id"):
            obj["source_id"] = x["source_id"]
        obj.update(named=_names(x.get("named")), picked=_names(x.get("picked")), joined=_names(x.get("joined")),
                   missing=_names(x.get("missing")))
        if total is not None:
            obj["total"] = total
            if explored is not None and explored < total:
                obj["explored"] = explored
        if _num(x.get("catalog_tables")) is not None:
            obj["catalog_tables"] = x["catalog_tables"]
        if _names(x.get("added_to_scope")):
            obj["added_to_scope"] = _names(x.get("added_to_scope"))
        if _num(x.get("prefiltered")) is not None:
            obj["prefiltered"] = x["prefiltered"]
        objects.append(obj)
    out: dict[str, Any] = {"op": "look", "id": str(op.get("id") or ""), "by": "system" if op.get("by") == "system" else "model",
                           "need": str(op.get("need") or ""),
                           "status": op.get("status") if op.get("status") in ("index", "failed") else "ok"}
    if isinstance(op.get("reason"), str) and op.get("reason"):
        out["reason"] = op["reason"]
    out["elapsed_ms"] = _num(op.get("elapsed_ms")) or 0
    out["objects"] = objects
    return out


# --------------------------------------------------------------------------
# 用量：{copilot, review, earlier, run}，每份是 UsageTotal（同页面的 TurnUsage）
# --------------------------------------------------------------------------


def usage_total(usage: Mapping[str, Any] | None) -> dict[str, Any] | None:
    """一份用量（providers/usage 的汇总、运行的 usage）只留要记的几项。不是对象交回 None。"""
    return add_total({}, usage) if isinstance(usage, Mapping) else None


def add_total(a: Mapping[str, Any] | None, b: Mapping[str, Any] | None) -> dict[str, Any]:
    """两份用量相加（缺的当 0，同页面的 addUsage）：只记不为 0 的项，金额保留 6 位。"""
    out = {k: v for k, v in (a or {}).items() if k in USAGE_KEYS}
    for key in USAGE_KEYS:
        try:
            value = float((b or {}).get(key) or 0)
        except (TypeError, ValueError):
            continue
        if value:
            out[key] = (out.get(key) or 0) + value
    for key in USAGE_KEYS:
        if key in out and key != "cost_usd":
            out[key] = int(out[key])
    if "cost_usd" in out:
        out["cost_usd"] = round(float(out["cost_usd"]), 6)
    return out


def retire_run(usage: Mapping[str, Any] | None) -> dict[str, Any]:
    """当前这次运行换下来了（自动重来、用户点重试）：它的用量并进 earlier，这一轮的花费不丢这一笔（同页面的 retireRun）。"""
    out = dict(usage or {})
    run = out.pop("run", None)
    if run:
        out["earlier"] = add_total(out.get("earlier"), run)
    return out


# --------------------------------------------------------------------------
# 用户点重试：上一次的结局归档
# --------------------------------------------------------------------------

#: 归档那一次的一句话（同页面以前 attemptOf 的写法）。也是重试时交给助手的原因
ATTEMPT_TEXT = {
    "cancelled": "已手动停止",
    "suspended": "服务重启，运行中断",
    "rejected": "被驳回",
    "unusable": "结论不可用：{note}",
    "partial": "存在缺口：{note}",
    "done": "已给出答案",
}


def _rejected_line(rejected: Mapping[str, Any]) -> str:
    """被驳回的那一句（同页面 REJECTED_TEXT.sub）：谁在哪一步驳回、备注；页面记过 sub 的照用。"""
    if isinstance(rejected.get("sub"), str) and rejected["sub"]:
        return rejected["sub"]
    who = str(rejected.get("actor") or "").strip()
    label = str(rejected.get("label") or "").strip()
    where = f"「{label}」" if label else "这一步"
    note = str(rejected.get("note") or "").strip()
    if rejected.get("decision") == "expired_reject":
        return f"{where}超过时限，按设置驳回"
    return (f"{who} 驳回了{where}" if who else f"{where}被驳回") + (f"：{note}" if note else "")


def attempt_record(row: Any, meta: Mapping[str, Any], *, n: int, at_ms: int | None = None) -> dict[str, Any]:
    """这一轮现在的样子折成一条留档（同页面以前的 attemptOf）：{n, outcome, summary, detail?, runId, answer?, at, cause?}。

    用户主动重试不能把上一次抹掉：那一次的答案可能只是有缺口、并非全错，新的一次也未必更好，两份要能对照。"""
    from app.api.conversations import turn_run_id

    review = row.review if isinstance(row.review, Mapping) else {}
    answer = row.answer or ""
    if len(answer) > ATTEMPT_ANSWER_CAP:
        answer = answer[:ATTEMPT_ANSWER_CAP] + "…"
    base: dict[str, Any] = {"n": n, "at": at_ms if at_ms is not None else int(time.time() * 1000),
                            "runId": turn_run_id(row)}
    if answer:
        base["answer"] = answer
    signals = "\n".join(str(s.get("detail") or "") for s in review.get("signals") or [] if isinstance(s, Mapping)) or None
    outcome = meta.get("outcome")
    if outcome == "cancelled" and isinstance(meta.get("rejected"), Mapping):
        sub = _rejected_line(meta["rejected"])
        return {**base, "outcome": "rejected", "summary": ATTEMPT_TEXT["rejected"] + (f"：{sub}" if sub else "")}
    if outcome == "cancelled" or (row.status == "error" and (row.error or "").strip() == CANCELLED):
        return {**base, "outcome": "cancelled", "summary": ATTEMPT_TEXT["cancelled"]}
    if outcome == "suspended":
        return {**base, "outcome": "suspended", "summary": ATTEMPT_TEXT["suspended"]}
    if row.status == "error":
        failure = meta.get("failure") if isinstance(meta.get("failure"), Mapping) else None
        line = (f"{failure.get('title')}：{failure.get('reason')}" if failure and failure.get("reason")
                else str(failure.get("title")) if failure and failure.get("title") else (row.error or "发生错误"))
        out = {**base, "outcome": "failed", "summary": line}
        if failure and failure.get("detail"):
            out["detail"] = str(failure["detail"])
        if isinstance(meta.get("cause"), Mapping):
            out["cause"] = dict(meta["cause"])
        return out
    note = str(review.get("note") or "")
    if review.get("severity") == "broken":
        return {**base, "outcome": "unusable", "summary": ATTEMPT_TEXT["unusable"].format(note=note), "detail": signals}
    if review.get("severity") == "degraded":
        return {**base, "outcome": "partial", "summary": ATTEMPT_TEXT["partial"].format(note=note), "detail": signals}
    return {**base, "outcome": "done", "summary": ATTEMPT_TEXT["done"]}


def attempt_reason(record: Mapping[str, Any]) -> str:
    """重试时交给助手的原因（同页面以前的写法）：那一次的一句话，有细节的括号里带前 300 字。"""
    summary = str(record.get("summary") or "")
    detail = record.get("detail")
    return f"{summary}（{str(detail)[:300]}）" if detail else summary


def with_max_steps(graph: Mapping[str, Any], steps: int) -> dict[str, Any]:
    """图里每个 agent 节点的步数上限调到 steps（「放宽步数重跑」，同页面的 withMaxSteps）。"""
    nodes = []
    for node in graph.get("nodes") or []:
        if isinstance(node, Mapping) and node.get("type") == "agent":
            data = dict(node.get("data") or {})
            data["config"] = {**dict(data.get("config") or {}), "max_steps": steps}
            node = {**node, "data": data}
        nodes.append(node)
    return {**dict(graph), "nodes": nodes}


__all__ = ["ANSWER_CAP", "ATTEMPT_ANSWER_CAP", "ATTEMPT_TEXT", "BUILD_INTERRUPTED", "CANCELLED", "GONE", "SUSPENDED",
           "USAGE_KEYS", "add_total", "answer_text", "attempt_reason", "attempt_record", "has_evidence", "look_event",
           "retire_run", "slim_issuance", "usage_total", "with_max_steps"]