From 7df224106d6db63e0e3441fcfdc1cc47f0b0994a Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Mon, 5 Oct 2026 18:51:51 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=A1=E6=89=B9=E8=B4=A8=E8=AF=A2=EF=BC=88?= =?UTF-8?q?=E5=90=8E=E7=AB=AF=EF=BC=89=EF=BC=9A=E8=BF=90=E8=A1=8C=E5=81=9C?= =?UTF-8?q?=E5=9C=A8=E5=AE=A1=E6=89=B9=E6=97=B6=EF=BC=8C=E5=AE=A1=E6=89=B9?= =?UTF-8?q?=E4=BA=BA=E9=92=88=E5=AF=B9=E6=8A=A5=E5=91=8A=E9=87=8C=E7=9A=84?= =?UTF-8?q?=E4=B8=80=E5=8F=A5=E8=AF=9D=E6=8F=90=E9=97=AE=EF=BC=8C=E5=8F=AA?= =?UTF-8?q?=E8=AF=BB=20agent=20=E7=8E=B0=E6=9F=A5=E7=8E=B0=E7=AD=94?= =?UTF-8?q?=EF=BC=8C=E9=97=AE=E7=AD=94=E5=86=99=E8=BF=9B=E8=BF=90=E8=A1=8C?= =?UTF-8?q?=E3=80=81=E9=9A=8F=E8=BF=90=E8=A1=8C=E5=B0=81=E5=AD=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - engine/inquiry.py：开放条件（停在审批且问的是待处理的那张卡；或已封存、核对一致的探索运行，追加在封存之后； 正式运行封存后、设置关闭、还在执行一律 409）；同一运行同时只答一条（进程内登记）；回答交给 run_verify_agent （不传 emit，数据版本取运行记录，上限和模型按核查设置），任务里给被问的那句、照 judge.prepare 取的证据摘录 （遮罩照旧、原报告编号标明）、选中的原文和问题，数据源取这次运行用过的查询工具；回答由 compose_doc 组装成文档 落工件，数字可以引用自己的查询格子或报告已有的指标、运行输入，裸数字被标出；引用到遮罩列的格子渲染成「已遮罩」， 回答正文和查询清单里出现的遮罩列的值换成「已遮罩」；写入走 append_events，运行在回答期间变了就不写、照实说明， 触顶和失败照样记一条 - api/inquiry.py：POST 提问、GET 列表（按报告、句子筛选）、GET 质询工件（只认质询引用过的，遮罩照证据面板）； 输入不合法回 {detail, field} - api/runs.py：质询正在回答时，审批回复（批准 / 驳回）和继续运行回 409 inquiry_busy，拦在记签批人之前 --- backend/app/api/inquiry.py | 58 ++- backend/app/api/runs.py | 12 + backend/app/engine/inquiry.py | 814 ++++++++++++++++++++++++++++++ backend/tests/test_inquiry_api.py | 768 ++++++++++++++++++++++++++++ 4 files changed, 1649 insertions(+), 3 deletions(-) create mode 100644 backend/app/engine/inquiry.py create mode 100644 backend/tests/test_inquiry_api.py diff --git a/backend/app/api/inquiry.py b/backend/app/api/inquiry.py index 829b228c..7c5b73ad 100644 --- a/backend/app/api/inquiry.py +++ b/backend/app/api/inquiry.py @@ -1,9 +1,61 @@ -"""审批质询：审批人在批准前就待审的报告提出质询。 +"""审批质询：审批人针对待审报告里的一句话提问，只读 agent 现查现答（规则和做法见 engine/inquiry.py）。 -目前只登记了路由前缀，接口由审批质询的工作包补上。路径挂在 /api/runs/{run_id}/… 下面。 +- POST /api/runs/{run_id}/inquiries：{report, unit, quote?, question, approval_id?} → {inquiry, recorded, note} +- GET /api/runs/{run_id}/inquiries?report=&unit=：{items: [Inquiry]}，封存前后的都算，按先后 +- GET /api/runs/{run_id}/inquiries/artifacts/{artifact_id}：质询引用过的工件，查询快照的遮罩列照证据面板遮好 + +不受理：409 {detail, code}（inquiry_not_open / inquiry_busy / inquiry_disabled）；输入不合法 422 {detail, field}， +field 指明是哪一项（前后端约定的形状，不是 FastAPI 自己那种校验错误数组，所以请求体自己解析）。 """ from __future__ import annotations -from fastapi import APIRouter +from typing import Any + +from fastapi import APIRouter, Header, HTTPException, Request +from fastapi.responses import JSONResponse + +from app.api.coded import CodedHTTPException +from app.api.runs import actor_of +from app.engine import inquiry +from app.engine.inquiry import InquiryRefused router = APIRouter(prefix="/api/runs", tags=["inquiry"]) + + +def _refused(e: InquiryRefused) -> Any: + """不受理 → HTTP 响应。422 照约定只回 {detail, field}；其余带机读码的回 {detail, code}。""" + if e.status == 422: + return JSONResponse(status_code=422, content={"detail": e.message, "field": e.field}) + if e.code: + raise CodedHTTPException(e.status, e.message, e.code) + raise HTTPException(e.status, e.message) + + +@router.post("/{run_id}/inquiries", response_model=None) +async def ask_inquiry(run_id: str, request: Request, x_actor: str | None = Header(default=None)) -> Any: + """问一个问题。回答要等 agent 查完库（最长约为设置里的时长上限再加半分钟收尾），接口一直等到回答完才返回。""" + try: + body = await request.json() + except (ValueError, UnicodeDecodeError): + body = None + try: + return await inquiry.ask(run_id, inquiry.parse_ask(body), actor=actor_of(x_actor)) + except InquiryRefused as e: + return _refused(e) + + +@router.get("/{run_id}/inquiries", response_model=None) +async def list_inquiries(run_id: str, report: str | None = None, unit: str | None = None) -> Any: + try: + return {"items": await inquiry.list_inquiries(run_id, report=report or None, unit=unit or None)} + except InquiryRefused as e: + return _refused(e) + + +@router.get("/{run_id}/inquiries/artifacts/{artifact_id}", response_model=None) +async def inquiry_artifact(run_id: str, artifact_id: str) -> Any: + try: + return await inquiry.inquiry_artifact(run_id, artifact_id) + except InquiryRefused as e: + return _refused(e) + diff --git a/backend/app/api/runs.py b/backend/app/api/runs.py index 00fd3637..fc6c44dd 100644 --- a/backend/app/api/runs.py +++ b/backend/app/api/runs.py @@ -15,6 +15,7 @@ from app.api.coded import CodedHTTPException from app.core.bus import bus from app.db.base import SessionLocal, get_session from app.db.models import Approval, Artifact, Run, RunEvent +from app.engine.inquiry import BUSY_CONTINUE, BUSY_DECIDE, INQUIRY_BUSY, inquiry_busy # 发起规则和发起前检查绑定的工具搬到了 engine/launch.py。这几个名字以前从这里导入，照旧可用 from app.engine.launch import ( # noqa: F401 TOOL_MISSING, LaunchRefused, launch_run, missing_tools, node_setting, tool_bindings, @@ -261,10 +262,18 @@ def _missing(e: KeyError) -> HTTPException: return HTTPException(404, str(e.args[0]) if e.args else "运行记录不存在") +def _not_while_inquiring(run_id: str, detail: str) -> None: + """这次运行正在回答审批质询（engine/inquiry.py 的进程内登记）时，审批回复和继续运行一律 409：问答要写进 + 运行停着的这一段，运行先往下走了，审批人看着回答再批的那份依据就落了空。""" + if inquiry_busy(run_id): + raise CodedHTTPException(409, detail, INQUIRY_BUSY) + + @router.post("/{run_id}/resume", response_model=RunOut) async def resume_run( run_id: str, payload: ResumeIn, x_actor: str | None = Header(default=None) ) -> Run: + _not_while_inquiring(run_id, BUSY_DECIDE) try: # approval_id 以前定义了却从没传下去——一次运行里有多条待审批时， # 引擎无从知道这次回复的是哪一条 @@ -296,6 +305,7 @@ async def continue_run( 一行状态判定，而 checkpoint 一直都在——改个模型 id 就得把前面两分钟的取数 重跑一遍，纯属白等。 """ + _not_while_inquiring(run_id, BUSY_CONTINUE) try: return await run_manager.continue_failed( run_id, graph=payload.graph, actor=actor_of(x_actor), @@ -593,6 +603,8 @@ async def decide( raise HTTPException(404, "审批记录不存在，所属的运行可能已被删除") if approval.status != "pending": raise HTTPException(409, "这条审批已处理过，请刷新页面查看最新状态") + # 在落签批人之前拦：拒掉的批准不该留下「谁试图批的」 + _not_while_inquiring(approval.run_id, BUSY_DECIDE) # 责任归属先落库再恢复执行——恢复失败也要留下"谁试图批的" approval.resolved_by = actor_of(x_actor) diff --git a/backend/app/engine/inquiry.py b/backend/app/engine/inquiry.py new file mode 100644 index 00000000..85cbceda --- /dev/null +++ b/backend/app/engine/inquiry.py @@ -0,0 +1,814 @@ +"""审批质询：审批人针对待审报告里的一句话提问，一个只读 agent 当场查库作答，问答写进这次运行。 + +审批卡上只有报告正文和核对概况。审批人对某句话有疑问（「排除测试账号会怎样」「这个数含不含退款」），以前只能驳回重跑， +或者自己去查库。质询让一个只读 agent（engine/verify_agent.py）当场作答，报告本身一个字不改： + +1. 开放条件（_open）：运行停在审批上、问的是那张待处理的审批卡——问答落在以后的封存范围里，随运行结束一起封存；或者 + 运行已经结束封存、核对一致、是探索运行——问答追加在封存之后（post_seal），是批注不是证据。正式运行封存之后不再追加 + （同按需裁判）；设置里关了「审批质询 · 启用」一律不答。同一次运行同时只答一条：进程内登记（_ANSWERING），登记期间 + 这次运行的审批回复、继续运行也不受理（api/runs.py），免得运行往下走了、问答还在路上 +2. 回答（_answer）：交给 agent 被问的那句原文、它挂的证据摘录（照 judge.prepare 的办法取，数据源遮罩的列照旧不给，原报告 + 的编号标明「原报告」）、审批人选中的原文和问题；工具是这次运行用过的查询工具对应的数据源；上限按设置（每次的查询次数、 + 时长，全局每日金额）。回答先给结论，数字只能引用 agent 自己查询的格子，或者报告里已有的指标、运行输入；回答文档由 + compose_doc 组装，裸数字被标出来 +3. 遮罩：回答里引用到遮罩列的格子渲染成「已遮罩」（按遮罩过的快照组装，_masking_loader）；回答正文、查询清单里出现的 + 遮罩列的值换成「已遮罩」（_Secrets）。哪些列要遮照证据面板的规则：查询当时记下的，加上数据源现在设的 +4. 写入（_record）：一条 approval.inquiry 事件，node_id 为空，用量写在数据的 usage 里（计进运行用量，见 runner 的 + _usage_of_events）。质询的查询快照只放在 inquiry_artifacts / queries 里，绝不放在 query_artifact 下：证据接口按那个键 + 收集封存的查询，会把质询的查询当成报告的证据。回答期间运行状态变了（被批准、被放弃、继续运行之后重新封存）， + append_events 不写，回答照样交回给提问的人，说明没有写进运行记录。触顶、失败照样记一条 +5. 读取：列表读 approval.inquiry 事件（封存前后的都算，按先后）；质询工件只认质询事件引用过的，遮罩照证据面板 + +不发 tool.start、llm.start 这类事件（run_verify_agent 不传 emit）：界面从接口的返回取回答；runner 的 _stale_replay +只看 tool.start，质询不影响恢复后的重放。 +""" +from __future__ import annotations + +import contextlib +import json +import logging +import re +import time +from collections.abc import Callable, Iterable, Iterator +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Any + +from sqlalchemy import select + +from app.core import artifact_store +from app.core.errors import describe_exception +from app.core.events import EventType +from app.data.provenance_types import DOC_PROVENANCE +from app.db.base import SessionLocal +from app.db.models import Approval, DataSource, Run, RunEvent +from app.engine import judge as judging +from app.engine.evidence import MARKER_RE, RenderError, compose_doc, iter_units, render_cell +from app.engine.expressions import numeric_text +from app.engine.merge_query import MERGE_SOURCE +from app.engine.nodes.challenge import mark_original, with_note +from app.engine.run_append import append_events +from app.engine.toolcalls import QUERY_PREFIX +from app.engine.verify_agent import RATIONALE, VerifyOutcome, run_verify_agent + +logger = logging.getLogger(__name__) + +#: 不受理时的机读码（前后端约定）。审批回复、继续运行在质询期间回的也是 INQUIRY_BUSY（api/runs.py） +INQUIRY_NOT_OPEN = "inquiry_not_open" +INQUIRY_BUSY = "inquiry_busy" +INQUIRY_DISABLED = "inquiry_disabled" +RUN_NOT_FOUND = "run_not_found" +ARTIFACT_NOT_FOUND = "inquiry_artifact_not_found" +ARTIFACT_TAMPERED = "snapshot_tampered" +#: 问题、选中的原文最长多少字 +QUESTION_MAX = 1000 +QUOTE_MAX = 500 +#: 回答期间运行状态变了、问答没写进运行记录时给提问的人的说明 +NOT_RECORDED = "运行状态已经变化，这次质询的问答没有写入运行记录" +BUSY = "这次运行另有一条质询正在回答，回答完再提问" +#: 质询正在回答时，这次运行的审批回复（批准 / 驳回）、继续运行回的话（api/runs.py） +BUSY_DECIDE = "正在回答质询，回答完再批准或驳回" +BUSY_CONTINUE = "正在回答质询，回答完再继续运行" +#: 触顶原因：前后端约定的四种。verify_agent 另有 max_steps（模型想了太多轮），轮数是按查询次数推出来的，算作查询次数到顶 +LIMITS = ("max_queries", "timeout_s", "max_cost_usd", "daily_max_usd") +#: 事件、接口里的用量只给这几项（runner 计进运行用量的也是这几项，_INQUIRY_KEYS） +USAGE_KEYS = ("input_tokens", "output_tokens", "cost_usd", "calls", "unpriced_calls") +#: 查询清单每条给的几项：没有行，查询结果只在快照里（取的时候按遮罩遮好） +QUERY_KEYS = ("alias", "artifact", "sql", "row_count", "truncated", "source") +#: 回答文档存成工件时的种类 +ANSWER_KIND = "inquiry_answer" +#: 没指明句子时交给 agent 的报告正文最多多少字 +REPORT_CHARS = 4000 +#: 列给 agent 的报告指标、运行输入最多几条 +REFS_MAX = 40 +#: 被问的是表格里的一格时，交给 agent 的表格最多几行（不含表头） +TABLE_ROWS = 20 +#: 遮罩列的值短于这么多字就不在正文里换：一个字的值（「是」「男」）换掉，会把不相干的词也换得面目全非 +SECRET_MIN = 2 +#: 遮罩列的值最多收这么多个（一份快照最多上千行）：再多就按已经收到的换 +SECRETS_MAX = 20_000 +#: 运行的终态。封存之后的追问只在这几种状态上 +_TERMINAL = ("succeeded", "failed", "cancelled") +#: 按词边界认值的字符：值的头尾是这样的字时，前后不能紧挨着同类的字（13900000001 不换掉 139000000012 的一截） +_WORD = re.compile(r"[0-9A-Za-z_]") + +INQUIRY_SYSTEM = ( + "你在帮审批人核对一份待批准的报告：审批人针对报告里的一句话提了问题，你查库作答。你不写报告，也不改报告。\n" + "- 问题是「如果……会怎样」（排除测试账号、换个时间范围、只看某个区域）时，写只读查询把它算出来，不要凭印象推测\n" + "- 回答直截了当：第一句直接给结论（是还是不是、是多少、差多少），再用一两句说依据；不要复述问题，不要客套\n" + "- 回答里的每个数字都写成引用：你自己查询结果里的一格写 [[v:Qn.rX.列名]]，报告里已有的指标、运行输入照任务里列出的" + "写法引用；不要直接写数字，没有出处的数字会被标出来\n" + "- 查不到、数据里没有判断所需的信息，就照实说查不到，不要编造" +) +INQUIRY_TASK_PROMPT = ( + "报告「{report}」{where}\n\n" + "{evidence}" + "{refs}" + "可以查询的数据源：{sources}\n\n" + "审批人的问题：\n{question}\n\n" + "按上面的做法作答，回答写在 rationale 里。" +) +#: 交给核查 agent 的结果结构：只有回答一项（verify_agent 的理由字段，沿用这里写的说明） +INQUIRY_SCHEMA: dict[str, Any] = { + "type": "object", + "properties": {RATIONALE: { + "type": "string", + "description": "回答：第一句直接给结论，再用一两句说依据；数字一律写成引用（[[v:Qn.rX.列名]]，" + "或任务里列出的报告指标、运行输入）", + }}, + "required": [RATIONALE], + "additionalProperties": False, +} + +# 不受理的说法 +DISABLED_WORDS = "审批质询已关闭（设置中的「{label}」），如需质询，请在设置的「核查」一节开启" +STILL_RUNNING = "运行还在执行，停在人工审批时、或者运行结束封存之后才能质询" +NO_CARD = "运行没有停在人工审批上（可能是服务重启时中断的），没有可以质询的审批" +CARD_GONE = "这张审批卡已处理，或不是这次运行正在等待的审批，请刷新页面查看最新状态" +CARD_DONE = "这张审批卡已处理，运行已经结束，请刷新页面查看最新状态" +FORMAL_SEALED = "正式运行已结束并封存，问答不能再追加进运行记录；封存之后只有探索运行可以继续质询" +UNSEALED = "运行已结束，但还没有封存，暂时不能质询" +BROKEN_SEAL = "封存核对未通过：封存后有事件被修改、删除或插入，报告和证据已不可信，不能再质询" +NEED_CARD = "运行停在人工审批上，需要指明质询的是哪张审批卡" +BAD_BODY = ("请求格式有误：需要一个对象，写明报告（report，报告撰写节点的 ID）和问题（question），" + "可选句子编号（unit）、选中的原文（quote）、审批卡（approval_id）") +DOC_TAMPERED = "报告文档与哈希不一致，疑似被修改，不能据此质询" +DOC_MISSING = "无法读取报告文档" +NO_ANSWER = "模型没有给出回答内容" + + +class InquiryRefused(Exception): + """质询不受理：status 是 HTTP 状态码，message 是给人看的话；409 带机读码 code，422 带出错的那一项 field。""" + + def __init__(self, status: int, message: str, *, code: str | None = None, field: str | None = None) -> None: + super().__init__(message) + self.status, self.message, self.code, self.field = status, message, code, field + + +def _invalid(message: str, name: str | None) -> InquiryRefused: + return InquiryRefused(422, message, field=name) + + +def _not_open(message: str) -> InquiryRefused: + return InquiryRefused(409, message, code=INQUIRY_NOT_OPEN) + + +# -------------------------------------------------------------------------- +# 同一次运行同时只答一条 +# -------------------------------------------------------------------------- + + +class _Answering: + """正在回答质询的运行（进程内登记）。 + + 只在本进程里有效：另一个进程里的批准拦不住，那时追加按条件更新核对（append_events），写不进就照实说没写进。""" + + def __init__(self) -> None: + self._runs: set[str] = set() + + def busy(self, run_id: str) -> bool: + return run_id in self._runs + + @contextlib.contextmanager + def hold(self, run_id: str) -> Iterator[None]: + # 查和登记之间没有 await：同一个事件循环里的两条质询不会都看到「没人在答」 + if run_id in self._runs: + raise InquiryRefused(409, BUSY, code=INQUIRY_BUSY) + self._runs.add(run_id) + try: + yield + finally: + self._runs.discard(run_id) + + +_ANSWERING = _Answering() + + +def inquiry_busy(run_id: str) -> bool: + """这次运行是不是正在回答质询（api/runs.py 据此让审批回复、继续运行回 409）。""" + return _ANSWERING.busy(run_id) + + +# -------------------------------------------------------------------------- +# 请求 +# -------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class Ask: + """一条质询：哪份报告（报告撰写节点的 id）的哪一句（句子单元 id，可以不指明）、选中的原文、问题、问的是哪张审批卡。""" + + report: str + unit: str | None + quote: str | None + question: str + approval_id: str | None + + +def parse_ask(body: Any) -> Ask: + """请求体 → Ask。不合法时抛 422，field 指明是哪一项。""" + if not isinstance(body, dict): + raise _invalid(BAD_BODY, None) + report = body.get("report") + if not isinstance(report, str) or not report.strip(): + raise _invalid("请指明质询的是哪份报告（报告撰写节点的 ID）", "report") + unit = body.get("unit") + if unit is not None and (not isinstance(unit, str) or not unit.strip()): + raise _invalid("句子编号需要是文字，例如「u12」；不指明句子时留空", "unit") + quote = body.get("quote") + if quote is not None and not isinstance(quote, str): + raise _invalid("选中的原文需要是文字", "quote") + quote = (quote or "").strip() or None + if quote is not None and len(quote) > QUOTE_MAX: + raise _invalid(f"选中的原文最长 {QUOTE_MAX} 字，当前为 {len(quote)} 字，请缩短选区", "quote") + question = body.get("question") + if not isinstance(question, str) or not question.strip(): + raise _invalid("请填写要质询的问题", "question") + question = question.strip() + if len(question) > QUESTION_MAX: + raise _invalid(f"问题最长 {QUESTION_MAX} 字，当前为 {len(question)} 字", "question") + approval_id = body.get("approval_id") + if approval_id is not None and (not isinstance(approval_id, str) or not approval_id.strip()): + raise _invalid("审批卡的 ID 需要是文字；不是在审批卡上提问时留空", "approval_id") + return Ask(report=report.strip(), unit=unit.strip() if unit else None, quote=quote, question=question, + approval_id=approval_id.strip() if approval_id else None) + + +# -------------------------------------------------------------------------- +# 开放条件 +# -------------------------------------------------------------------------- + +Events = list[tuple[int, str, str | None, Any]] + + +@dataclass +class _Opening: + """开放了的这一次质询：往哪种运行里写（mode）、写的时候要核对什么（approval_id / seal），以及被问的报告文档。 + + loader 是取证据的函数：封存之后只认封存范围内的工件（同按需裁判，api/evidence._sealed_loader），停在审批时和 + 报告节点里的裁判一样直接读工件库。events 是这次运行的事件（封存之后只取封存范围内的）。""" + + run_id: str + mode: str + title: str + versions: Any + events: Events + doc: dict[str, Any] + loader: Callable[[str], Any] + approval_id: str | None = None + seal: tuple[int | None, str | None] | None = None + + +async def _open(run_id: str, ask: Ask) -> _Opening: + async with SessionLocal() as session: + run = await session.get(Run, run_id) + if run is None: + raise InquiryRefused(404, "运行记录不存在", code=RUN_NOT_FOUND) + status, run_class, graph, versions = run.status, run.run_class, run.graph or {}, run.data_versions + cards = [str(a) for a in (await session.execute( + select(Approval.id).where(Approval.run_id == run_id, Approval.status == "pending"))).scalars()] \ + if status == "interrupted" else [] + rows: Events = [tuple(r) for r in await session.execute( + select(RunEvent.seq, RunEvent.type, RunEvent.node_id, RunEvent.data) + .where(RunEvent.run_id == run_id).order_by(RunEvent.seq))] if status == "interrupted" else [] + title = _node_title(graph, ask.report) + if status == "interrupted": + if not cards: + raise _not_open(NO_CARD) + if ask.approval_id is None: + raise _invalid(NEED_CARD, "approval_id") + if ask.approval_id not in cards: + raise _not_open(CARD_GONE) + doc = _waiting_doc(rows, run_id, ask.report, title) + return _Opening(run_id=run_id, mode="interrupted", title=title, versions=versions, events=rows, doc=doc, + loader=artifact_store.load, approval_id=ask.approval_id) + if status not in _TERMINAL: + raise _not_open(STILL_RUNNING) + if ask.approval_id is not None: + raise _not_open(CARD_DONE) + if run_class == "formal": + raise _not_open(FORMAL_SEALED) + from app.api.evidence import _reports, _sealed, _sealed_loader # api.evidence 导入了 engine 里的模块 + + sealed = await _sealed(run_id) + if not sealed.sealed: + raise _not_open(UNSEALED) + if not sealed.trusted: + raise _not_open(BROKEN_SEAL) + report = next((r for r in _reports(sealed) if r.node_id == ask.report), None) + if report is None: + raise _invalid(_no_report(title), "report") + if report.hash_ok is False: + raise _invalid(DOC_TAMPERED, "report") + if report.doc is None: + raise _invalid(DOC_MISSING, "report") + # 带期 4 标记的文档才让摘录取得到哈希链上的清单，和按需裁判同一个口径 + loader = _sealed_loader(sealed, chain=report.doc.get("provenance") == DOC_PROVENANCE) + return _Opening(run_id=run_id, mode="sealed", title=title, versions=versions, events=list(sealed.events), + doc=report.doc, loader=loader, seal=(sealed.run_manifest_seq, sealed.manifest_hash)) + + +def _node_title(graph: dict[str, Any], node_id: str) -> str: + for n in graph.get("nodes") or []: + if isinstance(n, dict) and n.get("id") == node_id: + return str((n.get("data") or {}).get("label") or node_id) + return node_id + + +def _no_report(title: str) -> str: + return f"这次运行里没有报告「{title}」的报告文档，请刷新页面后重新选择" + + +def _waiting_doc(events: Events, run_id: str, report: str, title: str) -> dict[str, Any]: + """停在审批时被问的报告文档：这个报告节点最后一次 report.checked 里的那份（同证据接口的 _reports）。文档要真是 + 这次运行里这个节点写的——内容寻址只保证没被改，不保证没被张冠李戴。""" + artifact = None + for _, etype, node_id, data in events: + if etype == EventType.REPORT_CHECKED and node_id == report and isinstance(data, dict) and data.get("doc_artifact"): + artifact = str(data["doc_artifact"]) + if artifact is None: + raise _invalid(_no_report(title), "report") + try: + doc = artifact_store.load(artifact) + except ValueError as e: + raise _invalid(DOC_TAMPERED, "report") from e + if not isinstance(doc, dict): + raise _invalid(DOC_MISSING, "report") + if doc.get("node_id") != report or doc.get("run_id") != run_id: + raise _invalid(DOC_TAMPERED, "report") + return doc + + +# -------------------------------------------------------------------------- +# 被问的那句 +# -------------------------------------------------------------------------- + + +@dataclass +class _Target: + """被问的那句：按裁判的候选句建（judge._candidate，带它挂的依据），表格里的一格另带整张表。""" + + cand: judging.Candidate + table: str | None = None + + +def _target(doc: dict[str, Any], unit: str) -> _Target: + """文档里的这一句。按 judge.candidates 的走法记小节和先后；不是结论句的（表格里的一格、标题）也照样建， + 审批人问的不一定是结论句。没有这一句抛 422。""" + markdown = str(doc.get("markdown") or "") + section = "" + for order, (block, item) in enumerate(iter_units(doc)): + text = judging._unit_text(item, markdown) + if block.get("type") == "heading": + section = text + if item.get("id") != unit: + continue + priority = judging.claim_priority(item, block, markdown) or 3 + cand = judging._candidate(item, priority, order, text, section) + return _Target(cand, _table_text(block, markdown) if block.get("type") == "table" else None) + raise _invalid(f"报告里没有这句（{unit}），请刷新页面后重新选择", "unit") + + +def _table_text(block: dict[str, Any], markdown: str) -> str: + """表格整张压成几行（表头在前），一格的值离了行列说不清是什么。""" + rows: dict[int, dict[int, str]] = {} + for unit in block.get("units") or []: + loc = unit.get("loc") if isinstance(unit, dict) else None + if isinstance(loc, dict) and isinstance(loc.get("row"), int) and isinstance(loc.get("col"), int): + rows.setdefault(loc["row"], {})[loc["col"]] = judging._unit_text(unit, markdown) + lines = [" | ".join(cells[c] for c in sorted(cells)) for _, cells in sorted(rows.items())] + kept = lines[:TABLE_ROWS + 1] + return "\n".join(kept) + ("\n…（后面的行省略了）" if len(lines) > len(kept) else "") + + +def _excerpts(doc: dict[str, Any], catalog: dict[str, Any], cand: judging.Candidate, loader: Callable[[str], Any], + masked: dict[str, Any]) -> list[str]: + """这句挂的每件证据的摘录：和 judge.prepare 一样逐件摘（数据源遮罩的列不给，带期 4 标记的文档多来历行）， + 指向原报告编号的写法标上「原报告」——agent 自己的查询也编 Q1、Q2…，不分开的话会把它们混成一回事。""" + fetch = judging._Fetch(loader) + provenance_on = doc.get("provenance") == DOC_PROVENANCE + queries = {a for a, e in catalog.items() if isinstance(e, dict) and e.get("kind") == "query"} + out: list[str] = [] + for alias in cand.aliases: + entry = catalog.get(alias) + if isinstance(entry, dict) and (text := judging._excerpt(alias, entry, [cand], fetch, masked, catalog, + provenance_on=provenance_on)): + out.append(mark_original(text, alias, queries)) + return out + + +def _report_refs(catalog: dict[str, Any]) -> dict[str, dict[str, Any]]: + """报告目录里回答可以直接引用的出处：口径卡指标（m:）和运行输入（i:）。它们的编号和 agent 自己的 Q1… 不会撞； + 报告的查询格子不行——agent 的查询也编 Q1，引用 Q1 说不清是哪一份。""" + return {a: e for a, e in catalog.items() if isinstance(e, dict) and e.get("kind") in ("metric", "input")} + + +def _run_sources(events: Events) -> list[str]: + """这次运行用过的查询工具对应的数据源，按第一次用到的先后。""" + out: list[str] = [] + for _, etype, _, data in events: + if etype not in (EventType.TOOL_START, EventType.TOOL_END) or not isinstance(data, dict): + continue + tool = str(data.get("tool") or "") + name = tool[len(QUERY_PREFIX):] if tool.startswith(QUERY_PREFIX) else "" + if name and name not in out: + out.append(name) + return out + + +async def _usable_sources(names: Iterable[str]) -> set[str]: + """现在能查的：存在、没有停用（核查建工具时也只认这样的）。""" + wanted = list(dict.fromkeys(names)) + if not wanted: + return set() + async with SessionLocal() as session: + rows = await session.execute( + select(DataSource.name).where(DataSource.name.in_(wanted), DataSource.enabled.is_(True))) + return {str(name) for name in rows.scalars()} + + +def _quoted(names: Iterable[str]) -> str: + return "、".join(f"「{n}」" for n in names) + + +def _task(op: _Opening, ask: Ask, target: _Target | None, excerpts: list[str], refs: dict[str, dict[str, Any]], + live: list[str], lost: list[str]) -> str: + if target is not None: + cand = target.cand + where = f"中的这句话{f'（所在小节「{cand.section}」）' if cand.section else ''}：\n「{cand.text}」" + if target.table: + where += f"\n\n这句话在一张表格里，表格是（每行一条，单元格用 | 隔开）：\n{target.table}" + else: + where = "，审批人没有指明是哪一句" + if ask.quote: + where += f"\n\n审批人选中的原文：\n「{ask.quote}」" + if target is None: + markdown = str(op.doc.get("markdown") or "") + cut = markdown[:REPORT_CHARS] + ("\n…（后面的省略了）" if len(markdown) > REPORT_CHARS else "") + where += f"\n\n报告正文：\n{cut}" + evidence = "" + if target is not None: + evidence = ("这句话挂的证据摘录（编号属于原报告；你的查询另起编号，回答里只能引用你自己的查询）：\n" + + "\n\n".join(excerpts) + "\n\n") if excerpts else "这句话没有挂依据。\n\n" + lines = [f"[[{alias}]] {entry.get('label') or alias}" for alias, entry in list(refs.items())[:REFS_MAX]] + more = f"\n…另有 {len(refs) - REFS_MAX} 项" if len(refs) > REFS_MAX else "" + refs_text = ("报告里已有的指标和运行输入（回答里可以照这个写法直接引用）：\n" + "\n".join(lines) + more + "\n\n" + if lines else "") + sources = "、".join(live) if live else "（没有可以查询的数据源，只能依据上面的报告内容作答）" + if lost: + sources += f"（这次运行还用到了{_quoted(lost)}，已不存在或已停用，无法查询）" + return INQUIRY_TASK_PROMPT.format(report=op.title, where=where, evidence=evidence, refs=refs_text, + sources=sources, question=ask.question) + + +# -------------------------------------------------------------------------- +# 遮罩 +# -------------------------------------------------------------------------- + + +def _query_shaped(content: Any) -> bool: + return isinstance(content, dict) and isinstance(content.get("rows"), list) + + +async def _hidden_columns(snapshot: dict[str, Any], masks: Any) -> tuple[set[str], list[str]]: + """一份查询快照要遮的列（小写）和已经不在的数据源，照证据面板的规则（api/evidence.py 的 _query_step）：查询当时 + 记下的，加上数据源现在设的；合并查询按它合并的那几个数据源算。""" + from app.api.evidence import _merge_masks, _recorded_masks + + if snapshot.get("source") == MERGE_SOURCE: + return await _merge_masks(snapshot, masks) + cols, found = await masks.of(snapshot.get("source")) + return cols | _recorded_masks(snapshot), ([] if found or not snapshot.get("source") else [str(snapshot["source"])]) + + +def _masked_word() -> str: + from app.api.evidence import MASKED + + return MASKED + + +def masked_snapshot(snapshot: dict[str, Any], hidden: set[str]) -> dict[str, Any]: + """遮罩列的每一格换成「已遮罩」，这几列的类型记成文本（渲染时不当数）。没有要遮的列原样返回。""" + columns = [str(c) for c in snapshot.get("columns") or []] + index = {i for i, c in enumerate(columns) if c.lower() in hidden} + if not index: + return snapshot + word = _masked_word() + + def row_of(row: Any) -> Any: + if isinstance(row, list): + return [word if i in index else v for i, v in enumerate(row)] + if isinstance(row, dict): + return {k: word if str(k).lower() in hidden else v for k, v in row.items()} + return row + + types = dict(snapshot["column_types"]) if isinstance(snapshot.get("column_types"), dict) else {} + types.update({columns[i]: "text" for i in index}) + return {**snapshot, "rows": [row_of(r) for r in snapshot["rows"]], "column_types": types} + + +def _masking_loader(hidden: dict[str, set[str]]) -> Callable[[str], Any]: + """组装回答文档用的 loader：agent 自己的查询快照按遮罩遮好再交出去，引用到遮罩列的格子就渲染成「已遮罩」。 + 别的查询快照不交（回答的目录里只有 agent 自己的查询），表结构快照这类没有行的原样。哈希照旧复验。""" + + def load(artifact: str) -> Any: + content = artifact_store.load(artifact) + if not _query_shaped(content): + return content + return masked_snapshot(content, hidden[artifact]) if artifact in hidden else None + + return load + + +def _forms(value: Any) -> set[str]: + """一格的值在正文里可能的写法：原样（压成一行），是数的再加上千分位的写法（13,900,000,001）。""" + text = json.dumps(value, ensure_ascii=False, default=str) if isinstance(value, (dict, list)) else str(value) + text = re.sub(r"\s+", " ", text).strip() + forms = {text} + if isinstance(value, (int, float)) or numeric_text(value) is not None: + with contextlib.suppress(RenderError, ValueError, ArithmeticError): + forms.add(render_cell(value)) + return {f for f in forms if len(f) >= SECRET_MIN} + + +def _pattern(value: str) -> str: + """一个值的正则：头尾是字母数字的按词边界认，数字还不能是更长的数的一截（1,390 里的 390、12.5 里的 2.5）。""" + head = tail = "" + if _WORD.match(value[0]): + head = r"(? None: + if not _query_shaped(snapshot) or not hidden: + return + columns = [str(c) for c in snapshot.get("columns") or []] + index = [i for i, c in enumerate(columns) if c.lower() in hidden] + for row in snapshot["rows"]: + if len(self.values) >= SECRETS_MAX: + return + if isinstance(row, list): + cells = [row[i] for i in index if i < len(row)] + elif isinstance(row, dict): + cells = [v for k, v in row.items() if str(k).lower() in hidden] + else: + continue + for value in cells: + if value is not None and not isinstance(value, bool): + self.values |= _forms(value) + + def scrubber(self) -> Callable[[str], str]: + if not self.values: + return lambda text: text + # 长的在前：13900000001 先于它的一截被认出来 + pattern = re.compile("|".join(_pattern(v) for v in sorted(self.values, key=lambda v: (-len(v), v)))) + word = _masked_word() + return lambda text: pattern.sub(word, text) if text else text + + +def _outside_markers(text: str, scrub: Callable[[str], str]) -> str: + """只换引用标记之外的文字：标记里是编号、行号、列名，换了会把一个好好的引用弄坏。""" + out: list[str] = [] + last = 0 + for m in MARKER_RE.finditer(text): + out += [scrub(text[last:m.start()]), m.group(0)] + last = m.end() + out.append(scrub(text[last:])) + return "".join(out) + + +# -------------------------------------------------------------------------- +# 回答 +# -------------------------------------------------------------------------- + + +def _limit(outcome: VerifyOutcome) -> str | None: + if outcome.limit in LIMITS: + return outcome.limit + return "max_queries" if outcome.limit == "max_steps" else None + + +async def _answer(op: _Opening, ask: Ask, target: _Target | None, settings: dict[str, Any]) -> dict[str, Any]: + """问 agent，组装回答文档，交回事件数据里回答的那几项。永远不抛：模型用不了、触顶都落在 ok / limit / error 里。""" + from app.api.evidence import _Masks + from app.api.settings import verify_model_spec + + doc = op.doc + catalog = doc.get("catalog") if isinstance(doc.get("catalog"), dict) else {} + masked = await judging.source_masks(catalog, loader=op.loader) + excerpts = _excerpts(doc, catalog, target.cand, op.loader, masked) if target is not None else [] + refs = _report_refs(catalog) + wanted = _run_sources(op.events) + usable = await _usable_sources(wanted) + live, lost = [s for s in wanted if s in usable], [s for s in wanted if s not in usable] + part = settings.get("inquiry") if isinstance(settings.get("inquiry"), dict) else {} + async with SessionLocal() as session: + spec = await verify_model_spec(session, None, settings=settings) + outcome = await run_verify_agent( + spec=spec, system=INQUIRY_SYSTEM, task=_task(op, ask, target, excerpts, refs, live, lost), sources=live, + output_schema=INQUIRY_SCHEMA, max_queries=part.get("max_queries"), timeout_s=part.get("timeout_s"), + daily_max_usd=settings.get("daily_max_usd"), run_id=op.run_id, data_versions=op.versions) + + # 遮罩：agent 自己的查询快照、报告自己的查询快照（被问的那句可能直接写着遮罩列的值，agent 会照着复述） + masks = _Masks() + hidden: dict[str, set[str]] = {} + secrets = _Secrets() + for query in outcome.queries: + snapshot = _quiet_load(artifact_store.load, query.get("artifact")) + if _query_shaped(snapshot): + hidden[query["artifact"]] = (await _hidden_columns(snapshot, masks))[0] + secrets.add(snapshot, hidden[query["artifact"]]) + for entry in catalog.values(): + if isinstance(entry, dict) and entry.get("kind") == "query": + snapshot = _quiet_load(op.loader, entry.get("artifact")) + if _query_shaped(snapshot): + secrets.add(snapshot, (await _hidden_columns(snapshot, masks))[0]) + scrub = secrets.scrubber() + + text, doc_id = "", None + source = (outcome.rationale or {}).get("source") if outcome.result is not None else None + if isinstance(source, str) and source.strip(): + answer = _compose(source, {**refs, **((outcome.rationale or {}).get("catalog") or {})}, op.run_id, hidden, + scrub) + if lost: + answer = with_note(answer, f"注：这次运行还用到了数据源{_quoted(lost)}，已不存在或已停用，未能查询，" + "以上回答不含其中的数据。") + text = str(answer.get("markdown") or "") + doc_id = await _store(answer, op, ask) + ok = bool(text.strip()) + queries = [{**{k: q.get(k) for k in QUERY_KEYS}, "sql": scrub(str(q.get("sql") or ""))} for q in outcome.queries] + return { + "answer_text": text, "answer_doc": doc_id, "ok": ok, "limit": _limit(outcome), + "error": None if ok else (outcome.error or NO_ANSWER), + "inquiry_artifacts": [q["artifact"] for q in queries if q.get("artifact")], "queries": queries, + "model": outcome.model, "usage": {k: outcome.usage.get(k, 0) for k in USAGE_KEYS}, + } + + +def _quiet_load(loader: Callable[[str], Any], artifact: Any) -> Any: + """取一份快照；取不到、哈希不符都当没有（这里只是为了找遮罩列的值，取不到的那份本来也渲染不出来）。""" + if not artifact: + return None + try: + return loader(str(artifact)) + except Exception: # noqa: BLE001 + return None + + +def _compose(source: str, catalog: dict[str, Any], run_id: str, hidden: dict[str, set[str]], + scrub: Callable[[str], str]) -> dict[str, Any]: + """agent 写的回答原文（带引用标记）→ 回答文档：目录是 agent 自己的查询，加上报告里的指标和运行输入；快照按遮罩 + 遮好再渲染。遮罩列的值先在标记之外的文字里换掉；换完正文里还认得出遮罩列的值（标记里写了值，解析不了，原样 + 出现在正文里），就连标记一起换了再组装一遍——宁可弄坏一个引用，也不让值露出来。""" + loader = _masking_loader(hidden) + doc = compose_doc(_outside_markers(source, scrub), catalog, run_id=run_id, loader=loader) + markdown = str(doc.get("markdown") or "") + if scrub(markdown) != markdown: + doc = compose_doc(scrub(source), catalog, run_id=run_id, loader=loader) + return doc + + +async def _store(doc: dict[str, Any], op: _Opening, ask: Ask) -> str | None: + try: + return await artifact_store.put_json(doc, kind=ANSWER_KIND, run_id=op.run_id, + meta={"report": ask.report, "unit": ask.unit}) + except Exception: # noqa: BLE001 - 存不下不影响回答本身：正文照常交回，只是数字点不开 + logger.warning("质询的回答文档保存失败（运行 %s）", op.run_id, exc_info=True) + return None + + +# -------------------------------------------------------------------------- +# 写入与读取 +# -------------------------------------------------------------------------- + + +async def _record(op: _Opening, data: dict[str, Any]) -> tuple[Any, str | None]: + """记一条 approval.inquiry。返回 (写入的事件, 没写进时的说明)。 + + 停在审批时：那张卡仍待审批、运行仍停着才写（批过了，运行可能已经停到下一张卡上）。封存之后：仍是开始回答时 + 的那份封存才写。append_events 返回 None 就是运行在回答期间变了。""" + items = [(EventType.APPROVAL_INQUIRY, None, data)] + try: + if op.mode == "interrupted": + written = await append_events(op.run_id, items, require="interrupted", approval_id=op.approval_id) + else: + written = await append_events(op.run_id, items, require="sealed", seal=op.seal) + except Exception as e: # noqa: BLE001 - 写不进运行记录也要把回答交回给提问的人 + logger.warning("质询的问答写入运行记录失败（运行 %s）", op.run_id, exc_info=True) + return None, f"这次质询的问答没能写入运行记录（{describe_exception(e)}）" + return (written[0], None) if written else (None, NOT_RECORDED) + + +def _iso(ts: Any) -> str | None: + if isinstance(ts, bool) or not isinstance(ts, (int, float)): + return None + return datetime.fromtimestamp(float(ts), timezone.utc).isoformat() + + +def inquiry_of(data: Any, *, seq: int | None, ts: Any) -> dict[str, Any]: + """事件数据 → 接口上的 Inquiry 对象（前后端约定的形状）。POST 的返回和 GET 的列表都经这里，两边一样。""" + data = data if isinstance(data, dict) else {} + usage = data.get("usage") if isinstance(data.get("usage"), dict) else {} + return { + "seq": seq, "at": _iso(ts), "actor": data.get("actor"), "report": data.get("report"), "unit": data.get("unit"), + "quote": data.get("quote"), "question": str(data.get("question") or ""), + "answer": {"text": str(data.get("answer_text") or ""), "doc_artifact": data.get("answer_doc"), + "ok": data.get("ok") is True, "limit": data.get("limit"), "error": data.get("error")}, + "queries": [{k: q.get(k) for k in QUERY_KEYS} for q in data.get("queries") or [] if isinstance(q, dict)], + "model": data.get("model"), + "usage": {k: usage.get(k, 0) for k in USAGE_KEYS}, + "post_seal": bool(data.get("post_seal")), "approval_id": data.get("approval_id"), + } + + +async def ask(run_id: str, request: Ask, *, actor: str | None = None) -> dict[str, Any]: + """答一条质询：{inquiry, recorded, note}。不受理时抛 InquiryRefused。 + + 设置按现在的（不按发起时快照进运行选项的那份）：质询不在图里执行、不参与重放，关掉「审批质询」应当马上生效。""" + from app.api.settings import VERIFY_SETTING_LABEL, verify_settings + + async with SessionLocal() as session: + settings = await verify_settings(session) + part = settings.get("inquiry") if isinstance(settings.get("inquiry"), dict) else {} + if not part.get("enabled", True): + raise InquiryRefused(409, DISABLED_WORDS.format(label=VERIFY_SETTING_LABEL["inquiry.enabled"]), + code=INQUIRY_DISABLED) + with _ANSWERING.hold(run_id): + opening = await _open(run_id, request) + target = _target(opening.doc, request.unit) if request.unit else None + data = {"question": request.question, "quote": request.quote, "report": request.report, "unit": request.unit, + "approval_id": opening.approval_id, "actor": actor, + **await _answer(opening, request, target, settings)} + event, note = await _record(opening, data) + if event is not None: + return {"inquiry": inquiry_of(event.data, seq=event.seq, ts=event.ts), "recorded": True, "note": None} + shown = {**data, "post_seal": True} if opening.mode == "sealed" else data + return {"inquiry": inquiry_of(shown, seq=None, ts=time.time()), "recorded": False, "note": note} + + +async def _inquiry_events(run_id: str) -> list[tuple[int, float, dict[str, Any]]]: + async with SessionLocal() as session: + if await session.get(Run, run_id) is None: + raise InquiryRefused(404, "运行记录不存在", code=RUN_NOT_FOUND) + rows = await session.execute( + select(RunEvent.seq, RunEvent.ts, RunEvent.data) + .where(RunEvent.run_id == run_id, RunEvent.type == EventType.APPROVAL_INQUIRY).order_by(RunEvent.seq)) + return [(seq, ts, data if isinstance(data, dict) else {}) for seq, ts, data in rows] + + +async def list_inquiries(run_id: str, *, report: str | None = None, unit: str | None = None) -> list[dict[str, Any]]: + """这次运行的质询（封存前后的都算），按先后；给了 report / unit 只要问那份报告、那一句的。""" + return [inquiry_of(data, seq=seq, ts=ts) for seq, ts, data in await _inquiry_events(run_id) + if (not report or data.get("report") == report) and (not unit or data.get("unit") == unit)] + + +async def inquiry_artifact(run_id: str, artifact_id: str) -> dict[str, Any]: + """质询引用过的工件：查询快照（遮罩列照证据面板遮好，SQL 里写着的遮罩列的值同样换掉）或回答文档（组装时就遮好了）。 + {id, content, masked[, mask_note]}。不是这次运行的质询引用过的一律 404：工件 id 不能拿来取任意快照。""" + from app.api.evidence import SOURCE_GONE, _Masks + + referenced = [data for _, _, data in await _inquiry_events(run_id) + if artifact_id in (data.get("inquiry_artifacts") or []) or data.get("answer_doc") == artifact_id] + if not referenced: + raise InquiryRefused(404, "这次运行的质询没有引用这件工件", code=ARTIFACT_NOT_FOUND) + try: + content = artifact_store.load(artifact_id) + except ValueError as e: + raise InquiryRefused(409, "工件与哈希不一致，疑似被修改，不予展示", code=ARTIFACT_TAMPERED) from e + if content is None: + raise InquiryRefused(404, "无法读取这件工件，可能已被删除", code=ARTIFACT_NOT_FOUND) + if not _query_shaped(content): + return {"id": artifact_id, "content": content, "masked": []} + masks = _Masks() + hidden, gone = await _hidden_columns(content, masks) + # 遮罩列的值按引用它的那几条质询各自查到的全部快照找：查询条件里写着的值，不一定在这一份快照的结果里 + secrets = _Secrets() + for data in referenced: + for other in dict.fromkeys(str(a) for a in data.get("inquiry_artifacts") or []): + snapshot = content if other == artifact_id else _quiet_load(artifact_store.load, other) + if _query_shaped(snapshot): + secrets.add(snapshot, hidden if other == artifact_id else (await _hidden_columns(snapshot, masks))[0]) + shown = masked_snapshot(content, hidden) + if isinstance(shown.get("sql"), str): + shown = {**shown, "sql": secrets.scrubber()(shown["sql"])} + columns = [str(c) for c in content.get("columns") or []] + out: dict[str, Any] = {"id": artifact_id, "content": shown, "masked": [c for c in columns if c.lower() in hidden]} + if gone: + out["mask_note"] = SOURCE_GONE.format(source="、".join(gone)) + return out + + +__all__ = [ + "BUSY_CONTINUE", "BUSY_DECIDE", "INQUIRY_BUSY", "INQUIRY_DISABLED", "INQUIRY_NOT_OPEN", "INQUIRY_SCHEMA", + "INQUIRY_SYSTEM", "NOT_RECORDED", "QUESTION_MAX", "QUOTE_MAX", "Ask", "InquiryRefused", "ask", "inquiry_artifact", + "inquiry_busy", "inquiry_of", "list_inquiries", "masked_snapshot", "parse_ask", +] diff --git a/backend/tests/test_inquiry_api.py b/backend/tests/test_inquiry_api.py new file mode 100644 index 00000000..b6adf89e --- /dev/null +++ b/backend/tests/test_inquiry_api.py @@ -0,0 +1,768 @@ +"""审批质询（engine/inquiry.py、api/inquiry.py）：运行停在人工审批时，审批人针对报告里的一句话提问，一个只读 agent +现查现答；问答写进这次运行（approval.inquiry 事件），随运行结束一起封存，报告本身一个字不改。 + +- 停在审批时质询：事件 node_id 为空、数据按前后端约定的形状；质询的查询工件只在 inquiry_artifacts / queries 里， + 不在 query_artifact 下（证据接口不把它们当成报告的证据）；批准、运行结束封存之后 verify 一致，封存范围包含这条事件； + 用量计入运行用量 +- 回答期间运行被批准（竞态）：回答照样交回，recorded 为假、note 说明，事件不落库 +- 同一运行同时只答一条：第二条质询、审批回复（批准 / 驳回）、继续运行都回 409 inquiry_busy，拒掉的批准不留痕迹 +- 封存之后：探索运行可以追问（post_seal，落在封存之后，不计运行用量），正式运行 409；设置里关了 409； + 没停在那张审批卡上、还在执行、封存核对不通过都 409；输入不合法 422，指明是哪一项 +- 回答：先给结论，数字只能引用自己查询的格子或报告已有的指标，裸数字被标出；遮罩列的格子渲染成「已遮罩」， + 回答正文、查询清单、事件里都不出现遮罩列的值 +- 触顶、失败照样记一条事件 +- 读取：列表按报告、句子筛选，按时间排；质询工件只认质询引用过的，遮罩照证据面板 +- 恢复后的重放不受影响：前面的节点不重复执行 + +模型一律是 MockChatModel 剧本，数据源是现造的 SQLite 文件，区域、金额、手机号都是虚构的。 +""" +from __future__ import annotations + +import asyncio +import json +import sqlite3 +import uuid +from dataclasses import dataclass, field + +import pytest +from httpx import ASGITransport, AsyncClient +from langchain_core.messages import AIMessage, ToolMessage +from sqlalchemy import delete, select + +from app.core import artifact_store +from app.data import introspect +from app.data.engine import engines +from app.db.base import SessionLocal +from app.db.models import Approval, DataSource, Run, RunEvent, Setting +from app.engine import inquiry as inquiry_mod +from app.engine import spend, verify_agent +from app.engine.evidence import iter_segments, iter_units +from app.engine.judge import SPEND_KEY as JUDGE_SPEND_KEY +from app.engine.nodes import report as report_mod +from app.engine.runner import run_manager, verify_manifest +from app.main import app +from app.providers.factory import ProviderNotConfigured +from app.providers.mock_model import MockChatModel + +PRICED = "claude-sonnet-5" +#: 各区域销售额：华东 1500（其中 300 是测试账号下的单）、华南 900、华北 400 +QUERY = "SELECT region AS 区域, SUM(amount) AS 销售额 FROM sales GROUP BY region ORDER BY 销售额 DESC" +TEXT = "华东销售额 [[v:Q1.r0.销售额]]，居各区域之首。[[see:Q1]]华南销售额 [[v:Q1.r1.销售额]]。[[see:Q1]]" +EAST, SOUTH = "华东销售额 1,500，居各区域之首。", "华南销售额 900。" +NO_TEST = ("SELECT region AS 区域, SUM(amount) AS 销售额 FROM sales WHERE is_test = 0 GROUP BY region " + "ORDER BY 销售额 DESC") +NO_TEST_ANSWER = "仍居各区域之首：排除测试账号后华东为 [[v:Q1.r0.销售额]]，华南为 [[v:Q1.r1.销售额]]。[[see:Q1]]" +NO_TEST_TEXT = "仍居各区域之首：排除测试账号后华东为 1,200，华南为 900。" +QUESTION = "排除测试账号会怎样？" +USAGE_KEYS = {"input_tokens", "output_tokens", "cost_usd", "calls", "unpriced_calls"} +ITEM_KEYS = {"seq", "at", "actor", "report", "unit", "quote", "question", "answer", "queries", "model", "usage", + "post_seal", "approval_id"} +EVENT_KEYS = {"question", "quote", "report", "unit", "approval_id", "actor", "answer_text", "answer_doc", "ok", + "limit", "error", "inquiry_artifacts", "queries", "model", "usage"} +QUERY_KEYS = {"alias", "artifact", "sql", "row_count", "truncated", "source"} +NOT_RECORDED = "运行状态已经变化，这次质询的问答没有写入运行记录" +BUSY_DECIDE = {"detail": "正在回答质询，回答完再批准或驳回", "code": "inquiry_busy"} + + +@pytest.fixture(autouse=True) +async def engine_up(): + await run_manager.setup() + async with SessionLocal() as session: + await session.execute(delete(Setting).where(Setting.key.in_( + [spend.SPEND_KEY, JUDGE_SPEND_KEY, "verify", "judge", "copilot"]))) + await session.commit() + yield + await run_manager.shutdown() + + +@pytest.fixture +async def client(): + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + yield c + + +async def _source(tmp_path, *, options=None) -> tuple[str, int]: + path = tmp_path / f"{uuid.uuid4().hex[:8]}.db" + db = sqlite3.connect(path) + db.executescript( + "CREATE TABLE sales (id INTEGER PRIMARY KEY, region TEXT, buyer_phone TEXT, amount INTEGER, is_test INTEGER);" + "INSERT INTO sales VALUES (1, '华东', '13900000001', 1200, 0), (2, '华东', '13900000002', 300, 1)," + " (3, '华南', '13900000003', 900, 0), (4, '华北', '13900000004', 400, 0);") + db.commit() + db.close() + name = f"shop_{uuid.uuid4().hex[:8]}" + async with SessionLocal() as session: + row = DataSource(name=name, kind="sqlite", database=str(path), readonly=True, enabled=True, + options=options or {}) + session.add(row) + await session.flush() + row.schema_cache = await introspect.introspect(row) + await session.commit() + return name, row.id + + +@pytest.fixture +async def shop(tmp_path): + """只读的手工源：四笔订单，华东两笔（其中一笔 300 是测试账号）、华南一笔 900、华北一笔 400。返回源名。""" + name, source_id = await _source(tmp_path) + yield name + await engines.invalidate(source_id) + + +@pytest.fixture +async def masked_shop(tmp_path): + """同样的四笔订单，买家手机号这一列设了遮罩。""" + name, source_id = await _source(tmp_path, options={"mask_columns": "buyer_phone"}) + yield name + await engines.invalidate(source_id) + + +def node(nid, ntype, label=None, **config): + return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": label or nid, "config": config}} + + +def chain(*nodes): + return {"nodes": list(nodes), "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]} + + +def gated(source: str, *, gate: bool = True) -> dict: + """取数 → 报告 → 人工审批 → 成果。gate 为假时没有审批，一口气跑完、封存。""" + return chain( + node("start", "input"), + node("fetch", "tool", tool=f"db_query__{source}", args={"sql": QUERY}), + node("write", "report", "写周报", instructions="写周报"), + *([node("gate", "human", mode="approve", title="周报可以出具吗")] if gate else []), + node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}]), + ) + + +@dataclass +class Play: + """质询 agent 对某个问题的剧本：依次执行的查询，最后交的回答。""" + + queries: list[str] = field(default_factory=list) + answer: str = "查不到判断所需的数据。" + + +PLAYS = {"排除测试账号": Play([NO_TEST], NO_TEST_ANSWER)} + + +class Models: + """写作者和质询 agent 两份剧本。写作者回 writer；质询 agent 按任务里的问题认出剧本，照着查询、交回答。 + + before：质询 agent 每次调用模型之前先 await 它（测试在这里卡住回答、或者在回答途中批掉审批）。 + tasks 记下质询 agent 收到的任务原文，finals 记下交回答那次调用收到的消息，specs 记下每次要的模型。""" + + def __init__(self, monkeypatch, source: str | None = None, *, writer: str = TEXT, plays=None, + model_id: str = PRICED, broken: Exception | None = None, before=None): + self.tasks: list[str] = [] + self.finals: list[list] = [] + self.writes = 0 + self.specs: list = [] + plays = PLAYS if plays is None else plays + + def which(task: str) -> Play: + return next((p for k, p in plays.items() if k in task), Play()) + + def decide(model, messages): + system = str(messages[0].content) + if inquiry_mod.INQUIRY_SYSTEM in system: + task = str(messages[1].content) if len(messages) > 1 else "" + play = which(task) + if model.response_format: + self.finals.append(list(messages)) + return AIMessage(content=json.dumps({"rationale": play.answer}, ensure_ascii=False)) + if not any(isinstance(m, ToolMessage) for m in messages): + self.tasks.append(task) + done = sum(1 for m in messages if isinstance(m, ToolMessage)) + if done < len(play.queries): + call = {"name": f"db_query__{source}", "args": {"sql": play.queries[done]}, "id": f"iq-{done}"} + return AIMessage(content="查一下。", tool_calls=[call]) + return AIMessage(content="查完了。") + if report_mod.ROLE in system: + self.writes += 1 + return AIMessage(content=writer) + raise AssertionError(f"没想到的模型调用：{system[:60]}") + + monkeypatch.setattr(MockChatModel, "_decide", decide) + original = MockChatModel._agenerate + + async def held(chat, messages, stop=None, run_manager=None, **kwargs): + if before is not None and inquiry_mod.INQUIRY_SYSTEM in str(messages[0].content): + await before() + return await original(chat, messages, stop, run_manager, **kwargs) + + monkeypatch.setattr(MockChatModel, "_agenerate", held) + + async def verify_model(session, spec): + self.specs.append(spec) + if broken is not None: + raise broken + return MockChatModel(model_name=model_id), model_id + + monkeypatch.setattr(verify_agent, "get_chat_model", verify_model) + + +async def _row(run_id: str) -> Run: + async with SessionLocal() as session: + return await session.get(Run, run_id) + + +async def wait(run_id: str, statuses=("succeeded", "failed")) -> Run: + for _ in range(600): + row = await _row(run_id) + if row.status in statuses and (row.status == "interrupted" or row.manifest_seq is not None): + await run_manager.wait_idle(run_id) + return await _row(run_id) + await asyncio.sleep(0.03) + raise AssertionError(f"运行 {run_id} 停在 {row.status}") + + +async def pause(source: str, *, run_class: str = "exploratory", graph: dict | None = None) -> Run: + run = await run_manager.start(graph=graph or gated(source), input_payload={}, run_class=run_class) + return await wait(run.id, ("interrupted",)) + + +async def finish(source: str, *, run_class: str = "exploratory", graph: dict | None = None) -> Run: + run = await run_manager.start(graph=graph or gated(source, gate=False), input_payload={}, run_class=run_class) + return await wait(run.id) + + +async def pending(run_id: str) -> list[Approval]: + async with SessionLocal() as session: + return list((await session.execute( + select(Approval).where(Approval.run_id == run_id, Approval.status == "pending"))).scalars()) + + +async def events(run_id: str, etype: str, node_id: str | None = None) -> list[RunEvent]: + async with SessionLocal() as session: + q = select(RunEvent).where(RunEvent.run_id == run_id, RunEvent.type == etype) + if node_id: + q = q.where(RunEvent.node_id == node_id) + return list((await session.execute(q.order_by(RunEvent.seq))).scalars()) + + +async def report_doc(run_id: str) -> dict: + [checked] = await events(run_id, "report.checked", "write") + return artifact_store.load(checked.data["doc_artifact"]) + + +def unit_of(doc: dict, text: str) -> str: + md = doc["markdown"] + return next(u["id"] for _, u in iter_units(doc) if md[u["span"][0]:u["span"][1]].strip() == text) + + +async def ask(client, run_id: str, **body): + return await client.post(f"/api/runs/{run_id}/inquiries", json=body) + + +async def set_verify(value: dict) -> None: + async with SessionLocal() as session: + session.add(Setting(key="verify", value=value)) + await session.commit() + + +# -------------------------------------------------------------------------- +# 停在审批时质询 → 写进事件 → 批准 → 封存 +# -------------------------------------------------------------------------- + + +async def test_asking_while_waiting_is_recorded_and_sealed_with_the_run(monkeypatch, shop, client): + models = Models(monkeypatch, shop) + paused = await pause(shop) + [approval] = await pending(paused.id) + doc = await report_doc(paused.id) + east = unit_of(doc, EAST) + + r = await ask(client, paused.id, report="write", unit=east, quote="居各区域之首", question=QUESTION, + approval_id=approval.id) + assert r.status_code == 200, r.text + body = r.json() + assert body.keys() == {"inquiry", "recorded", "note"} + assert body["recorded"] is True and body["note"] is None + item = body["inquiry"] + assert item.keys() == ITEM_KEYS + assert (item["report"], item["unit"], item["quote"], item["question"], item["approval_id"]) == ( + "write", east, "居各区域之首", QUESTION, approval.id) + assert item["post_seal"] is False and item["actor"] is None and isinstance(item["at"], str) + answer = item["answer"] + assert answer.keys() == {"text", "doc_artifact", "ok", "limit", "error"} + assert (answer["ok"], answer["limit"], answer["error"]) == (True, None, None) + assert answer["text"] == NO_TEST_TEXT, "回答先给结论，数字按查询结果换好" + answer_doc = artifact_store.load(answer["doc_artifact"]) + assert answer_doc["markdown"] == answer["text"] and answer_doc["violations"] == [] + cited = [s["cite"] for s in iter_segments(answer_doc) if s.get("cite")] + assert [(c["alias"], c["locator"]["row"], c["value"]) for c in cited] == [("Q1", 0, 1200), ("Q1", 1, 900)] + [query] = item["queries"] + assert query.keys() == QUERY_KEYS + assert (query["alias"], query["source"], query["sql"], query["row_count"], query["truncated"]) == ( + "Q1", shop, NO_TEST, 3, False) + assert artifact_store.load(query["artifact"])["rows"] == [["华东", 1200], ["华南", 900], ["华北", 400]] + assert item["model"] == PRICED + assert item["usage"].keys() == USAGE_KEYS and item["usage"]["calls"] >= 2 and item["usage"]["cost_usd"] > 0 + + # 交给质询 agent 的：被问的那句、它挂的证据摘录（原报告的编号标明）、选中的原文、问题、可以查的数据源 + [task] = models.tasks + assert EAST in task and "【原报告 Q1】" in task and "居各区域之首" in task and QUESTION in task and shop in task + + # 事件：node_id 为空，数据按约定；查询工件不在 query_artifact 下 + [event] = await events(paused.id, "approval.inquiry") + assert event.node_id is None and event.seq == item["seq"] + assert event.data.keys() == EVENT_KEYS + assert event.data["inquiry_artifacts"] == [query["artifact"]] + assert event.data["queries"] == item["queries"] + assert (event.data["answer_text"], event.data["answer_doc"]) == (answer["text"], answer["doc_artifact"]) + assert "query_artifact" not in event.data + assert (await _row(paused.id)).status == "interrupted" + + # 列表里是同一条 + listed = await client.get(f"/api/runs/{paused.id}/inquiries") + assert listed.status_code == 200 and listed.json() == {"items": [item]} + + # 批准 → 运行结束封存：这条事件在封存范围里，verify 一致 + r = await client.post(f"/api/approvals/{approval.id}/decide", json={"approved": True}) + assert r.status_code == 200, r.text + final = await wait(paused.id) + assert final.status == "succeeded", final.error + assert final.manifest_seq > event.seq + assert (await verify_manifest(paused.id))["ok"] is True + # 用量计进运行用量，明细在 inquiry 下 + assert final.usage["inquiry"]["calls"] == item["usage"]["calls"] + assert final.usage["inquiry"]["input_tokens"] == item["usage"]["input_tokens"] + # 证据接口不把质询的查询当成报告的证据 + from app.api.evidence import _sealed + + assert query["artifact"] not in (await _sealed(paused.id)).queries + # 报告本身一个字没改 + assert (await report_doc(paused.id)) == doc + # 改它一个字，封存核对就对不上：它确实在封存范围里 + async with SessionLocal() as session: + stored = await session.get(RunEvent, event.id) + stored.data = {**stored.data, "answer_text": "改过的回答"} + await session.commit() + assert (await verify_manifest(paused.id))["ok"] is False + + +async def test_an_approval_that_lands_while_answering_leaves_no_record(monkeypatch, shop, client): + """回答途中运行被批准了（另一个进程、或者抢在登记之前的批准）：回答照样交回给提问的人，但不写进运行记录。""" + seen: dict = {} + + async def approve_meanwhile(): + if not seen.get("done"): + seen["done"] = True + await run_manager.resume(seen["run"], {"approved": True}, approval_id=seen["approval"]) + + Models(monkeypatch, shop, before=approve_meanwhile) + paused = await pause(shop) + [approval] = await pending(paused.id) + seen.update(run=paused.id, approval=approval.id) + + r = await ask(client, paused.id, report="write", unit=None, question=QUESTION, approval_id=approval.id) + assert r.status_code == 200, r.text + body = r.json() + assert body["recorded"] is False and body["note"] == NOT_RECORDED + assert body["inquiry"]["seq"] is None and body["inquiry"]["answer"]["ok"] is True + assert body["inquiry"]["answer"]["text"] == NO_TEST_TEXT + + final = await wait(paused.id) + assert final.status == "succeeded", final.error + assert await events(paused.id, "approval.inquiry") == [] + assert (await verify_manifest(paused.id))["ok"] is True + assert "inquiry" not in final.usage + assert (await client.get(f"/api/runs/{paused.id}/inquiries")).json() == {"items": []} + + +async def test_while_answering_approving_resuming_and_continuing_are_refused(monkeypatch, shop, client): + entered, release = asyncio.Event(), asyncio.Event() + + async def hold(): + entered.set() + await release.wait() + + Models(monkeypatch, shop, before=hold) + paused = await pause(shop) + [approval] = await pending(paused.id) + asking = asyncio.create_task(ask(client, paused.id, report="write", unit=None, question=QUESTION, + approval_id=approval.id)) + try: + await asyncio.wait_for(entered.wait(), 10) + refused = await client.post(f"/api/approvals/{approval.id}/decide", json={"approved": False, "note": "先别批"}, + headers={"X-Actor": "reviewer"}) + assert refused.status_code == 409 and refused.json() == BUSY_DECIDE + refused = await client.post(f"/api/runs/{paused.id}/resume", + json={"response": {"approved": True}, "approval_id": approval.id}) + assert refused.status_code == 409 and refused.json() == BUSY_DECIDE + refused = await client.post(f"/api/runs/{paused.id}/continue", json={}) + assert refused.status_code == 409 and refused.json()["code"] == "inquiry_busy" + second = await ask(client, paused.id, report="write", unit=None, question="华南为什么低？", + approval_id=approval.id) + assert second.status_code == 409 and second.json()["code"] == "inquiry_busy" + # 拒掉的批准不留痕迹：卡仍待审批，没记下签批人 + async with SessionLocal() as session: + card = await session.get(Approval, approval.id) + assert card.status == "pending" and card.resolved_by is None + assert (await _row(paused.id)).status == "interrupted" + finally: + release.set() + answered = await asyncio.wait_for(asking, 30) + assert answered.status_code == 200 and answered.json()["recorded"] is True + + # 回答完了，照常批准 + r = await client.post(f"/api/approvals/{approval.id}/decide", json={"approved": True}) + assert r.status_code == 200, r.text + assert (await wait(paused.id)).status == "succeeded" + + +# -------------------------------------------------------------------------- +# 封存之后：探索运行可以追问，正式运行不行 +# -------------------------------------------------------------------------- + + +async def test_after_the_seal_only_exploratory_runs_take_inquiries(monkeypatch, shop, client): + Models(monkeypatch, shop) + done = await finish(shop) + assert done.status == "succeeded", done.error + east = unit_of(await report_doc(done.id), EAST) + + r = await ask(client, done.id, report="write", unit=east, question=QUESTION) + assert r.status_code == 200, r.text + body = r.json() + assert body["recorded"] is True and body["note"] is None + item = body["inquiry"] + assert item["post_seal"] is True and item["approval_id"] is None and item["answer"]["text"] == NO_TEST_TEXT + [event] = await events(done.id, "approval.inquiry") + assert event.data["post_seal"] is True and event.seq > done.manifest_seq + after = await _row(done.id) + assert (after.manifest_seq, after.manifest_hash) == (done.manifest_seq, done.manifest_hash) + assert (await verify_manifest(done.id))["ok"] is True + assert "inquiry" not in after.usage, "封存之后的问答是批注，不计进运行用量" + assert (await client.get(f"/api/runs/{done.id}/inquiries")).json()["items"] == [item] + + # 带着审批卡来问：那张卡已经不在等了 + r = await ask(client, done.id, report="write", unit=east, question=QUESTION, approval_id="no-such-card") + assert r.status_code == 409 and r.json()["code"] == "inquiry_not_open" + + formal = await finish(shop, run_class="formal") + assert formal.status == "succeeded", formal.error + r = await ask(client, formal.id, report="write", unit=unit_of(await report_doc(formal.id), EAST), + question=QUESTION) + assert r.status_code == 409 and r.json()["code"] == "inquiry_not_open" + assert "正式运行" in r.json()["detail"] + assert await events(formal.id, "approval.inquiry") == [] + + +async def test_a_formal_run_waiting_for_approval_takes_inquiries(monkeypatch, shop, client): + """正式运行停在审批时可以质询：问答随运行一起封存，正式运行同样可以复核。""" + Models(monkeypatch, shop) + paused = await pause(shop, run_class="formal") + [approval] = await pending(paused.id) + r = await ask(client, paused.id, report="write", unit=None, question=QUESTION, approval_id=approval.id) + assert r.status_code == 200 and r.json()["recorded"] is True + await client.post(f"/api/approvals/{approval.id}/decide", json={"approved": True}) + final = await wait(paused.id) + [event] = await events(paused.id, "approval.inquiry") + assert final.manifest_seq > event.seq and (await verify_manifest(paused.id))["ok"] is True + + +async def test_switched_off_in_settings_it_is_refused(monkeypatch, shop, client): + await set_verify({"inquiry": {"enabled": False}}) + models = Models(monkeypatch, shop) + paused = await pause(shop) + [approval] = await pending(paused.id) + r = await ask(client, paused.id, report="write", unit=None, question=QUESTION, approval_id=approval.id) + assert r.status_code == 409 and r.json()["code"] == "inquiry_disabled" + assert "「审批质询 · 启用」" in r.json()["detail"] + assert models.tasks == [] and models.specs == [] + assert await events(paused.id, "approval.inquiry") == [] + + +async def test_the_run_must_be_waiting_on_that_approval(monkeypatch, shop, client): + models = Models(monkeypatch, shop) + paused = await pause(shop) + [approval] = await pending(paused.id) + + r = await ask(client, paused.id, report="write", unit=None, question=QUESTION) + assert r.status_code == 422 and r.json() == {"detail": r.json()["detail"], "field": "approval_id"} + r = await ask(client, paused.id, report="write", unit=None, question=QUESTION, approval_id="no-such-card") + assert r.status_code == 409 and r.json()["code"] == "inquiry_not_open" + + # 还在执行的运行 + async with SessionLocal() as session: + running = Run(workflow_name="还在执行", status="running", graph={}, input={}) + session.add(running) + await session.commit() + running_id = running.id + r = await ask(client, running_id, report="write", unit=None, question=QUESTION) + assert r.status_code == 409 and r.json()["code"] == "inquiry_not_open" + + r = await ask(client, "no-such-run", report="write", unit=None, question=QUESTION) + assert r.status_code == 404 + + # 封存核对不通过的探索运行：报告和证据已不可信，不再质询 + done = await finish(shop) + async with SessionLocal() as session: + first = (await session.execute(select(RunEvent).where(RunEvent.run_id == done.id) + .order_by(RunEvent.seq).limit(1))).scalar_one() + first.data = {**(first.data or {}), "tampered": True} + await session.commit() + r = await ask(client, done.id, report="write", unit=None, question=QUESTION) + assert r.status_code == 409 and r.json()["code"] == "inquiry_not_open" + assert models.tasks == [], "开放条件不满足时不问模型" + + +@pytest.mark.parametrize("body, field_name", [ + ({"unit": None, "question": QUESTION}, "report"), + ({"report": 3, "question": QUESTION}, "report"), + ({"report": "write", "unit": None}, "question"), + ({"report": "write", "unit": None, "question": " "}, "question"), + ({"report": "write", "unit": None, "question": "问" * 1001}, "question"), + ({"report": "write", "unit": None, "question": QUESTION, "quote": "引" * 501}, "quote"), + ({"report": "write", "unit": None, "question": QUESTION, "quote": 5}, "quote"), + ({"report": "write", "unit": 12, "question": QUESTION}, "unit"), + ({"report": "write", "unit": None, "question": QUESTION, "approval_id": 7}, "approval_id"), +]) +async def test_bad_requests_name_the_field(monkeypatch, shop, client, body, field_name): + Models(monkeypatch, shop) + paused = await pause(shop) + [approval] = await pending(paused.id) + r = await ask(client, paused.id, **{"approval_id": approval.id, **body}) + assert r.status_code == 422, r.text + assert r.json().keys() == {"detail", "field"} and r.json()["field"] == field_name + assert r.json()["detail"] and "[" not in r.json()["detail"] + + +async def test_the_report_and_the_sentence_must_exist(monkeypatch, shop, client): + models = Models(monkeypatch, shop) + paused = await pause(shop) + [approval] = await pending(paused.id) + r = await ask(client, paused.id, report="no-such-report", unit=None, question=QUESTION, approval_id=approval.id) + assert r.status_code == 422 and r.json()["field"] == "report" + r = await ask(client, paused.id, report="write", unit="u999", question=QUESTION, approval_id=approval.id) + assert r.status_code == 422 and r.json()["field"] == "unit" + r = await client.post(f"/api/runs/{paused.id}/inquiries", content=b"[1, 2]", + headers={"content-type": "application/json"}) + assert r.status_code == 422 and r.json().keys() == {"detail", "field"} + r = await client.post(f"/api/runs/{paused.id}/inquiries", content=b"{not json", + headers={"content-type": "application/json"}) + assert r.status_code == 422 and r.json().keys() == {"detail", "field"} + assert models.tasks == [] + + +# -------------------------------------------------------------------------- +# 回答：出处、裸数字、遮罩 +# -------------------------------------------------------------------------- + +TOP_BUYER = "SELECT buyer_phone, amount AS 金额 FROM sales ORDER BY amount DESC LIMIT 1" +BUYER_ORDERS = "SELECT COUNT(*) AS 笔数 FROM sales WHERE buyer_phone = '13900000001'" + + +async def test_bare_numbers_are_flagged_and_masked_values_never_show(monkeypatch, masked_shop, client): + plays = {"买得最多": Play([TOP_BUYER, BUYER_ORDERS], + "金额最高的一笔来自 [[v:Q1.r0.buyer_phone]]（手机号 13900000001），金额 [[v:Q1.r0.金额]]；" + "这位客户共 [[v:Q2.r0.笔数]] 笔订单，另有 3 笔小额订单。")} + Models(monkeypatch, masked_shop, plays=plays) + paused = await pause(masked_shop) + [approval] = await pending(paused.id) + r = await ask(client, paused.id, report="write", unit=None, question="买得最多的是哪位客户？", + approval_id=approval.id) + assert r.status_code == 200, r.text + assert "1390000000" not in r.text, "回答、查询清单里都不能出现遮罩列的值" + item = r.json()["inquiry"] + answer = item["answer"] + assert answer["ok"] is True + assert answer["text"] == "金额最高的一笔来自 已遮罩（手机号 已遮罩），金额 1,200；这位客户共 1 笔订单，另有 3 笔小额订单。" + doc = artifact_store.load(answer["doc_artifact"]) + assert "1390000000" not in json.dumps(doc, ensure_ascii=False) + # 裸数字被标出来：界面在它下面画「无出处」 + assert [s["text"] for s in iter_segments(doc) if s.get("issue") == "uncited_number"] == ["3"] + # 遮罩列的格子：出处照记，值是「已遮罩」 + masked = [s["cite"] for s in iter_segments(doc) if (s.get("cite") or {}).get("locator", {}).get("column") + == "buyer_phone"] + assert [(c["alias"], c["status"], c["value"]) for c in masked] == [("Q1", "resolved", "已遮罩")] + first, second = item["queries"] + assert first["sql"] == TOP_BUYER + assert second["sql"] == "SELECT COUNT(*) AS 笔数 FROM sales WHERE buyer_phone = '已遮罩'" + [event] = await events(paused.id, "approval.inquiry") + assert "1390000000" not in json.dumps(event.data, ensure_ascii=False) + + # 质询工件：遮罩照证据面板，查询里写着的遮罩列的值同样换掉 + got = await client.get(f"/api/runs/{paused.id}/inquiries/artifacts/{first['artifact']}") + assert got.status_code == 200, got.text + assert got.json()["content"]["rows"] == [["已遮罩", 1200]] and got.json()["masked"] == ["buyer_phone"] + got = await client.get(f"/api/runs/{paused.id}/inquiries/artifacts/{second['artifact']}") + assert got.status_code == 200 and "1390000000" not in got.text + assert got.json()["content"]["rows"] == [[1]] + + +async def test_the_answer_may_cite_the_reports_own_metrics(monkeypatch, client): + """报告里已有的出处（口径卡指标）回答里可以直接引用；这次运行没用到查询工具，agent 只能依据证据作答。""" + metrics = [{"id": "gmv", "name": "销售额", "unit": "元", "decimals": 1, "expression": "vars.kpi.gmv"}] + graph = chain( + node("start", "input"), + node("calc", "transform", mode="expression", expression="{'gmv': 45678.5}", assign_to="kpi"), + node("caliber", "metrics", caliber="周报口径", caliber_version="v2", metrics=metrics), + node("write", "report", instructions="写周报"), + node("gate", "human", mode="approve", title="确认"), + node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}]), + ) + plays = {"销售额": Play([], "按口径卡，本周销售额为 [[m:gmv]]，这是本周的口径值。[[see:m:gmv]]")} + models = Models(monkeypatch, None, writer="本周销售额 [[m:gmv]]。[[see:m:gmv]]", plays=plays) + paused = await pause("", graph=graph) + [approval] = await pending(paused.id) + doc = await report_doc(paused.id) + r = await ask(client, paused.id, report="write", unit=unit_of(doc, "本周销售额 45,678.5元。"), + question="销售额是怎么算的？", approval_id=approval.id) + assert r.status_code == 200, r.text + answer = r.json()["inquiry"]["answer"] + assert answer["text"] == "按口径卡，本周销售额为 45,678.5元，这是本周的口径值。" + assert artifact_store.load(answer["doc_artifact"])["violations"] == [] + [task] = models.tasks + assert "[[m:gmv]]" in task and "销售额 = 45,678.5元" in task + assert r.json()["inquiry"]["queries"] == [] + + +async def test_a_cell_in_a_table_comes_with_the_whole_table(monkeypatch, shop, client): + """问的是表格里的一格：一格的值离了行列说不清是什么，任务里另带整张表（表头在前）和这一格的出处摘录。""" + models = Models(monkeypatch, shop, writer="各区域销售额如下。\n\n[[table:Q1]]") + paused = await pause(shop) + [approval] = await pending(paused.id) + doc = await report_doc(paused.id) + cell = unit_of(doc, "1,500") + r = await ask(client, paused.id, report="write", unit=cell, question=QUESTION, approval_id=approval.id) + assert r.status_code == 200, r.text + [task] = models.tasks + assert "「1,500」" in task and "区域 | 销售额\n华东 | 1,500\n华南 | 900\n华北 | 400" in task + # 摘录照裁判的办法：被引用的列排在前面，只给被引用的那一行 + assert "【原报告 Q1】" in task and "列：销售额 | 区域\nr0：1,500 | 华东\n" in task + + +def test_masked_values_are_matched_as_whole_values(): + """遮罩列的值按整值认：更长的数里的一截、千分位写法里的一截不换；一个字的值不换（换了会把不相干的词弄坏）； + 数按原样和千分位两种写法认。""" + secrets = inquiry_mod._Secrets() + snapshot = {"columns": ["phone", "name", "gender", "code", "salary", "region"], + "rows": [["13900000001", "张伟", "男", "390", 12000, "华东"]]} + secrets.add(snapshot, {"phone", "name", "gender", "code", "salary"}) + scrub = secrets.scrubber() + assert scrub("客户 13900000001、139000000012、1,390 元、390 笔、张伟买了男装，月薪 12,000，另 12000.5") == ( + "客户 已遮罩、139000000012、1,390 元、已遮罩 笔、已遮罩买了男装，月薪 已遮罩，另 12000.5") + assert scrub("华东") == "华东", "没遮的列不换" + assert inquiry_mod._Secrets().scrubber()("13900000001") == "13900000001" + + +# -------------------------------------------------------------------------- +# 触顶与失败 +# -------------------------------------------------------------------------- + + +async def test_hitting_the_query_limit_is_recorded(monkeypatch, shop, client): + await set_verify({"inquiry": {"max_queries": 1}}) + plays = {"排除测试账号": Play([NO_TEST, QUERY], NO_TEST_ANSWER)} + models = Models(monkeypatch, shop, plays=plays) + paused = await pause(shop) + [approval] = await pending(paused.id) + r = await ask(client, paused.id, report="write", unit=None, question=QUESTION, approval_id=approval.id) + assert r.status_code == 200, r.text + item = r.json()["inquiry"] + assert (item["answer"]["ok"], item["answer"]["limit"]) == (True, "max_queries"), "触顶照样按查到的结果收尾" + # 没指明句子：任务里给报告正文，不给哪一句的摘录 + [task] = models.tasks + assert "审批人没有指明是哪一句" in task and f"报告正文：\n{EAST}{SOUTH}" in task and "原报告" not in task + assert [q["sql"] for q in item["queries"]] == [NO_TEST] + [event] = await events(paused.id, "approval.inquiry") + assert (event.data["ok"], event.data["limit"]) == (True, "max_queries") + + +async def test_an_unavailable_model_is_recorded_as_a_failure(monkeypatch, shop, client): + Models(monkeypatch, shop, broken=ProviderNotConfigured("没有配置模型接入")) + # 写作者照常用演示模型：只有质询 agent 的模型用不了 + paused = await pause(shop) + [approval] = await pending(paused.id) + r = await ask(client, paused.id, report="write", unit=None, question=QUESTION, approval_id=approval.id) + assert r.status_code == 200, r.text + body = r.json() + assert body["recorded"] is True + answer = body["inquiry"]["answer"] + assert answer["ok"] is False and answer["text"] == "" and answer["doc_artifact"] is None + assert answer["error"] and "模型" in answer["error"] + assert body["inquiry"]["queries"] == [] and body["inquiry"]["usage"]["calls"] == 0 + [event] = await events(paused.id, "approval.inquiry") + assert event.data["ok"] is False and event.data["error"] == answer["error"] + + +async def test_the_daily_cap_is_recorded_as_a_limit(monkeypatch, shop, client): + async with SessionLocal() as session: + session.add(Setting(key=spend.SPEND_KEY, value={"date": spend._today(), "usd": 5.0, "calls": 3})) + await session.commit() + models = Models(monkeypatch, shop) + paused = await pause(shop) + [approval] = await pending(paused.id) + r = await ask(client, paused.id, report="write", unit=None, question=QUESTION, approval_id=approval.id) + assert r.status_code == 200, r.text + answer = r.json()["inquiry"]["answer"] + assert (answer["ok"], answer["limit"]) == (False, "daily_max_usd") and answer["error"] + assert models.finals == [] + [event] = await events(paused.id, "approval.inquiry") + assert event.data["limit"] == "daily_max_usd" + + +# -------------------------------------------------------------------------- +# 读取 +# -------------------------------------------------------------------------- + + +async def test_listing_filters_by_report_and_sentence_and_artifacts_are_whitelisted(monkeypatch, shop, client): + plays = {"排除测试账号": Play([NO_TEST], NO_TEST_ANSWER), "华南为什么": Play([], "查不到判断所需的数据。")} + Models(monkeypatch, shop, plays=plays) + paused = await pause(shop) + [approval] = await pending(paused.id) + doc = await report_doc(paused.id) + east, south = unit_of(doc, EAST), unit_of(doc, SOUTH) + first = (await ask(client, paused.id, report="write", unit=east, question=QUESTION, + approval_id=approval.id)).json()["inquiry"] + second = (await ask(client, paused.id, report="write", unit=south, question="华南为什么低？", + approval_id=approval.id)).json()["inquiry"] + assert first["seq"] < second["seq"] + + listed = await client.get(f"/api/runs/{paused.id}/inquiries") + assert listed.json() == {"items": [first, second]} + assert (await client.get(f"/api/runs/{paused.id}/inquiries", params={"report": "write", "unit": south}) + ).json() == {"items": [second]} + assert (await client.get(f"/api/runs/{paused.id}/inquiries", params={"report": "other"})).json() == {"items": []} + assert (await client.get("/api/runs/no-such-run/inquiries")).status_code == 404 + + [query] = first["queries"] + got = await client.get(f"/api/runs/{paused.id}/inquiries/artifacts/{query['artifact']}") + assert got.status_code == 200, got.text + assert got.json()["id"] == query["artifact"] and got.json()["masked"] == [] + assert got.json()["content"]["rows"] == [["华东", 1200], ["华南", 900], ["华北", 400]] + # 回答文档也是质询引用过的工件 + got = await client.get(f"/api/runs/{paused.id}/inquiries/artifacts/{first['answer']['doc_artifact']}") + assert got.status_code == 200 and got.json()["content"]["markdown"] == NO_TEST_TEXT + # 报告自己的查询快照不是质询引用过的 + [fetched] = [e for e in await events(paused.id, "tool.end", "fetch")] + got = await client.get(f"/api/runs/{paused.id}/inquiries/artifacts/{fetched.data['query_artifact']}") + assert got.status_code == 404 + assert (await client.get(f"/api/runs/{paused.id}/inquiries/artifacts/nothing")).status_code == 404 + + +# -------------------------------------------------------------------------- +# 重放 +# -------------------------------------------------------------------------- + + +async def test_resuming_after_an_inquiry_does_not_repeat_the_nodes(monkeypatch, shop, client): + """质询不发 tool.start（_stale_replay 只看它），批准后从断点继续：取数、报告都不再执行一遍。""" + models = Models(monkeypatch, shop) + paused = await pause(shop) + [approval] = await pending(paused.id) + r = await ask(client, paused.id, report="write", unit=None, question=QUESTION, approval_id=approval.id) + assert r.json()["recorded"] is True + assert [e for e in await events(paused.id, "tool.start") if e.node_id is None] == [] + await client.post(f"/api/approvals/{approval.id}/decide", json={"approved": True}) + final = await wait(paused.id) + assert final.status == "succeeded", final.error + for node_id in ("fetch", "write"): + assert len(await events(paused.id, "node.started", node_id)) == 1, f"{node_id} 又执行了一遍" + assert len(await events(paused.id, "tool.start", "fetch")) == 1 + assert models.writes == 1