From 778fb8b83edb6f82a314c356a4814440cbe4814b Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Tue, 6 Oct 2026 08:52:16 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=A1=E6=89=B9=E8=B4=A8=E8=AF=A2=E5=A4=8D?= =?UTF-8?q?=E6=A0=B8=E4=BF=AE=E6=AD=A3=EF=BC=88=E7=AC=AC=201=20=E8=BD=AE?= =?UTF-8?q?=EF=BC=8C=E5=90=8E=E7=AB=AF=EF=BC=89=EF=BC=9A=E9=81=AE=E7=BD=A9?= =?UTF-8?q?=E6=94=B9=E6=8C=89=20SQL=20=E8=AE=A4=E5=87=BA=E6=94=B9=E4=BA=86?= =?UTF-8?q?=E5=90=8D=E5=AD=97=E7=9A=84=E9=81=AE=E7=BD=A9=E5=88=97=E3=80=81?= =?UTF-8?q?=E5=86=8D=E6=8C=89=E5=80=BC=E9=81=AE=EF=BC=8C=E6=8C=89=E5=80=BC?= =?UTF-8?q?=E8=AE=A4=E6=9C=89=E9=97=A8=E6=A7=9B=E3=80=81SQL=20=E5=8F=AA?= =?UTF-8?q?=E6=8D=A2=E5=AD=97=E9=9D=A2=E9=87=8F=EF=BC=9B=E6=B2=A1=E5=86=99?= =?UTF-8?q?=E8=BF=9B=E8=AE=B0=E5=BD=95=E7=9A=84=E5=9B=9E=E7=AD=94=E7=85=A7?= =?UTF-8?q?=E6=A0=B7=E7=82=B9=E5=BE=97=E5=BC=80=EF=BC=9B=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E6=BA=90=E6=A0=B8=E5=AF=B9=E6=98=AF=E4=B8=8D=E6=98=AF=E8=BF=90?= =?UTF-8?q?=E8=A1=8C=E7=94=A8=E8=BF=87=E7=9A=84=E9=82=A3=E4=B8=80=E4=B8=AA?= =?UTF-8?q?=EF=BC=9B=E6=8A=A5=E9=94=99=E5=92=8C=E5=9B=9E=E7=AD=94=E6=9C=AB?= =?UTF-8?q?=E5=B0=BE=E7=9A=84=E8=AF=B4=E6=98=8E=E7=BB=99=E6=8F=90=E9=97=AE?= =?UTF-8?q?=E4=BA=BA=E7=9C=8B=E5=BE=97=E6=87=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 复核提出的问题和改法： - 遮罩列起了别名（本仓库查询的惯例是中文别名）、套了函数、经子查询或 CTE 转一道，原值会出现在回答正文、事件和质询工件里，第二遍组装也拦不住：按 SQL 认出改了名字的遮罩列（COUNT 只数个数、窗口的分区键只决定次序，不算；解析不了又写着遮罩列的，整份快照遮掉），快照另按值遮（别的列里写着遮罩值的格子）；组装两遍之后正文里还认得出遮罩值的回答不交出去；提示词写明遮罩列不要选出原值、不要起别名 - 按值遮罩范围过宽：年龄这类短的遮罩值会把阈值、天数、日期、LIMIT 换成「已遮罩」，封存的查询清单不再是实际执行的 SQL。纯数字至少 5 位才按值认，日期、小数、时间、千分位里的一截不算；SQL 只换字面量和注释里的值，短的数只在直接和遮罩列（含别名）比较时才换；问题、选中原文里提问人自己写的值不换 - 提问人写的阈值在回答里被标「无出处」：问题、选中原文里的数作为 allow_numbers 交给两遍组装 - 同一条 SQL 事件里遮了、质询工件接口里没遮：工件接口直接用事件里记下的那条，遮罩值照回答时收（含报告自己的查询快照） - 没写进运行记录的回答交回了工件 id，质询工件接口却一律 404：这样的回答在进程里留一小时（最多 200 条），这段时间里照样认它引用的工件 - 回答里引用的报告指标（m:）点不开：质询工件接口认回答文档目录里的口径卡 - 遮罩值上万个时每次回答、每次取工件都在事件循环里编一个上万分支的正则，卡住进程一秒上下：先按词、按开头几个字和长短筛出正文里真有的几个值再编；收值、解析 SQL、逐格比对放进线程 - 按名字认源：同名重建的源会被当成运行用过的去查，没固定版本的上传表格还会被悄悄补固定进已封存的运行。按运行里查询快照记的源 id、运行固定的版本核对，对不上的、没固定过的不交给 agent - 封存之后的追问查的是现在的数据，读者分不清是口径差异还是数据更新：任务里给报告生成的时间，回答末尾注明 - 模型轮数用完被说成查询次数到顶，引用一个都解析不了照样算回答完成：轮数用完 limit 记 null；轮数用完、查询执行失败、引用解析不了都在回答末尾注明 - 报错透出核查 agent 的内部说法（「交结果」「按结构」、裸写的字段名）：按情形换成发生了什么加下一步，原话只进日志 - 运行在执行时的说法对正式运行不成立、对过时的审批卡也没说清：带着审批卡来的叫人刷新，没带的按运行类别说什么时候能问 - 回答文档的工件引用 meta 记下问题的前 40 字，运行的工件列表据此认得出是哪个问题的回答 - 测试补上证明接线的用例：两道审批的竞态（approval_id 守卫）、上传表格按运行固定的版本查、设置里的上限和 data_versions 交到 agent 手里、开放途中运行被继续了、报告文档不属于这次运行；新增遮罩的单元测试。每条都做过去掉实现、测试变红的验证 --- backend/app/api/inquiry.py | 7 +- backend/app/engine/inquiry.py | 900 ++++++++++++++++++++++---- backend/tests/test_inquiry_api.py | 480 ++++++++++++-- backend/tests/test_inquiry_masking.py | 204 ++++++ 4 files changed, 1417 insertions(+), 174 deletions(-) create mode 100644 backend/tests/test_inquiry_masking.py diff --git a/backend/app/api/inquiry.py b/backend/app/api/inquiry.py index 111ae5fa..65b8b774 100644 --- a/backend/app/api/inquiry.py +++ b/backend/app/api/inquiry.py @@ -3,8 +3,11 @@ - 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}：质询引用过的工件 {id, content, masked[, mask_note]}—— - 查询快照的遮罩列照证据面板遮好（masked 是遮掉的列名）；回答文档（answer.doc_artifact）组装时就遮好了，原样给。 - 不是这次运行的质询引用过的回 404：工件 id 不能拿来取任意快照 + 查询快照（queries[].artifact）照回答时的办法遮好：遮罩列（含起了别名、套了函数的）整列遮，别的列里写着遮罩值的格子 + 也遮，masked 是有格子被遮掉的列名；content.sql 就是事件里记下的那条（遮罩值已换掉），两处一字不差。回答文档 + （answer.doc_artifact）组装时就遮好了，原样给；回答里引用的报告指标（doc.catalog 里 m: 条目的 artifact，口径卡） + 也从这里取，界面一律按 doc.catalog[alias].artifact 取。没写进运行记录的回答（recorded 为假）在本进程里留一小时， + 这段时间里它引用的工件照样能取；之后、或者服务重启过，回 404。别的工件一律 404 不受理：409 {detail, code}（inquiry_not_open / inquiry_busy / inquiry_disabled）；输入不合法 422 {detail, field}， field 指明是哪一项（前后端约定的形状，不是 FastAPI 自己那种校验错误数组，所以请求体自己解析）。 diff --git a/backend/app/engine/inquiry.py b/backend/app/engine/inquiry.py index 07b46f78..e6d3b762 100644 --- a/backend/app/engine/inquiry.py +++ b/backend/app/engine/inquiry.py @@ -8,43 +8,52 @@ （同按需裁判）；设置里关了「审批质询 · 启用」一律不答。同一次运行同时只答一条：进程内登记（_ANSWERING），登记期间 这次运行的审批回复、继续运行也不受理（api/runs.py），免得运行往下走了、问答还在路上 2. 回答（_answer）：交给 agent 被问的那句原文、它挂的证据摘录（照 judge.prepare 的办法取，数据源遮罩的列照旧不给，原报告 - 的编号标明「原报告」）、审批人选中的原文和问题；工具是这次运行用过的查询工具对应的数据源；上限按设置（每次的查询次数、 - 时长，全局每日金额）。回答先给结论，数字只能引用 agent 自己查询的格子，或者报告里已有的指标、运行输入；回答文档由 - compose_doc 组装，裸数字被标出来 -3. 遮罩：回答里引用到遮罩列的格子渲染成「已遮罩」（按遮罩过的快照组装，_masking_loader）；回答正文、查询清单里出现的 - 遮罩列的值换成「已遮罩」（_Secrets）。哪些列要遮照证据面板的规则：查询当时记下的，加上数据源现在设的 + 的编号标明「原报告」）、审批人选中的原文和问题；工具是这次运行用过的查询工具对应的数据源（同名重建的、没固定过版本的 + 上传表格不算，见 _usable_sources）；上限按设置（每次的查询次数、时长，全局每日金额）。回答先给结论，数字只能引用 agent + 自己查询的格子，或者报告里已有的指标、运行输入；回答文档由 compose_doc 组装，裸数字被标出来（提问人自己写的数除外）。 + 封存之后的追问查的是数据源现在的数据，任务里给报告生成的时间，回答末尾注明；触顶、查询失败、引用解析不了也在末尾注明 +3. 遮罩（_plan）：哪些列要遮照证据面板的规则（查询当时记下的，加上数据源现在设的），另按 SQL 认出改了名字的遮罩列 + （起别名、套函数、经子查询转一道，_derived_hidden）；快照按列遮，再按值遮（别的列里写着遮罩值的格子，_masked_view）。 + 回答正文、查询清单里的遮罩值换成「已遮罩」（_Secrets、_scrub_sql）：按值认有门槛，SQL 只换字面量，阈值、日期、LIMIT + 照原样，封存的是实际执行的语句。组装两遍之后正文里还认得出遮罩值的，这份回答不交出去 4. 写入（_record）：一条 approval.inquiry 事件，node_id 为空，用量写在数据的 usage 里（计进运行用量，见 runner 的 _usage_of_events）。质询的查询快照只放在 inquiry_artifacts / queries 里，绝不放在 query_artifact 下：证据接口按那个键 收集封存的查询，会把质询的查询当成报告的证据。回答期间运行状态变了（被批准、被放弃、继续运行之后重新封存）， - append_events 不写，回答照样交回给提问的人，说明没有写进运行记录。触顶、失败照样记一条 -5. 读取：列表读 approval.inquiry 事件（封存前后的都算，按先后）；质询工件只认质询事件引用过的，遮罩照证据面板 + append_events 不写，回答照样交回给提问的人，说明没有写进运行记录（这样的回答在进程里留一阵，_HELD）。触顶、失败 + 照样记一条 +5. 读取：列表读 approval.inquiry 事件（封存前后的都算，按先后）；质询工件只认质询引用过的（查询快照、回答文档、回答里 + 引用的报告口径卡），遮罩照回答时的办法 不发 tool.start、llm.start 这类事件（run_verify_agent 不传 emit）：界面从接口的返回取回答；runner 的 _stale_replay 只看 tool.start，质询不影响恢复后的重放。 """ from __future__ import annotations +import asyncio import contextlib import json import logging import re import time +from collections import OrderedDict 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 sqlalchemy import func, select from app.core import artifact_store from app.core.errors import describe_exception from app.core.events import EventType +from app.data.catalog import QUERY_SNAPSHOT_KIND 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.db.models import Approval, Artifact, 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.evidence import MARKER_RE, RenderError, compose_doc, iter_segments, iter_units, render_cell from app.engine.expressions import numeric_text +from app.engine.issuance import extract_numbers 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 @@ -69,24 +78,32 @@ BUSY = "这次运行另有一条质询正在回答，回答完再提问" #: 质询正在回答时，这次运行的审批回复（批准 / 驳回）、继续运行回的话（api/runs.py） BUSY_DECIDE = "正在回答质询，回答完再批准或驳回" BUSY_CONTINUE = "正在回答质询，回答完再继续运行" -#: 触顶原因：前后端约定的四种。verify_agent 另有 max_steps（模型想了太多轮），轮数是按查询次数推出来的，算作查询次数到顶 +#: 触顶原因：前后端约定的四种。verify_agent 另有 max_steps（模型想了太多轮）：走到那一步时查询次数一定还没用完， +#: 不能说成查询次数到顶，记为 null，回答末尾注明（NOTE_ROUNDS） 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") -#: 回答文档存成工件时的种类 +#: 回答文档存成工件时的种类；工件引用的 meta 里记下问题的前这么多字，运行的工件列表据此认得出是哪个问题的回答 ANSWER_KIND = "inquiry_answer" +QUESTION_BRIEF = 40 #: 没指明句子时交给 agent 的报告正文最多多少字 REPORT_CHARS = 4000 #: 列给 agent 的报告指标、运行输入最多几条 REFS_MAX = 40 #: 被问的是表格里的一格时，交给 agent 的表格最多几行（不含表头） TABLE_ROWS = 20 -#: 遮罩列的值短于这么多字就不在正文里换：一个字的值（「是」「男」）换掉，会把不相干的词也换得面目全非 +#: 按值认遮罩值的门槛：文字至少这么多字（一个字的「是」「男」换掉，会把不相干的词也换得面目全非）；纯数字至少这么多位 +#: （年龄、等级、小额金额这类短的数和阈值、天数、日期、LIMIT 里的数太容易撞上，按值换会把实际执行的 SQL、提问人自己 +#: 的条件改得面目全非。它们只在 SQL 里和遮罩列直接比较时才换，见 _scrub_sql；引用到的格子照旧按列遮） SECRET_MIN = 2 -#: 遮罩列的值最多收这么多个（一份快照最多上千行）：再多就按已经收到的换 +SECRET_DIGITS = 5 +#: 遮罩列的值最多收这么多种写法（一份快照最多上千行）：再多就按已经收到的换 SECRETS_MAX = 20_000 +#: 没写进运行记录的回答在进程里留多久、最多留几条：这段时间里质询工件接口照样认它引用的工件 +HELD_TTL_S = 3600 +HELD_MAX = 200 #: 运行的终态。封存之后的追问只在这几种状态上 _TERMINAL = ("succeeded", "failed", "cancelled") #: 按词边界认值的字符：值的头尾是这样的字时，前后不能紧挨着同类的字（13900000001 不换掉 139000000012 的一截） @@ -98,7 +115,8 @@ INQUIRY_SYSTEM = ( "- 回答直截了当：第一句直接给结论（是还是不是、是多少、差多少），再用一两句说依据；不要复述问题，不要客套\n" "- 回答里的每个数字都写成引用：你自己查询结果里的一格写 [[v:Qn.rX.列名]]，报告里已有的指标、运行输入照任务里列出的" "写法引用；不要直接写数字，没有出处的数字会被标出来\n" - "- 数据源设了遮罩的列（手机号、姓名这类）不要在回答里写出原值，需要指明时说「该客户」「这一笔」\n" + "- 数据源设了遮罩的列（手机号、姓名这类）：查询里不要选出原值，也不要给它们起别名；要数人数、笔数用 COUNT，回答里需要" + "指明时说「该客户」「这一笔」，不要写出原值\n" "- 查不到、数据里没有判断所需的信息，就照实说查不到，不要编造" ) INQUIRY_TASK_PROMPT = ( @@ -106,6 +124,7 @@ INQUIRY_TASK_PROMPT = ( "{evidence}" "{refs}" "可以查询的数据源：{sources}\n\n" + "{since}" "{masked}" "提问人的问题：\n{question}\n\n" "按上面的做法作答，回答写在 rationale 里。" @@ -124,7 +143,9 @@ INQUIRY_SCHEMA: dict[str, Any] = { # 不受理的说法 DISABLED_WORDS = "审批质询已关闭（设置中的「{label}」），如需质询，请在设置的「核查」一节开启" -STILL_RUNNING = "运行还在执行，停在人工审批时、或者运行结束封存之后才能质询" +STILL_RUNNING = "运行仍在执行。停在人工审批时，或运行结束并封存后，才能质询" +STILL_RUNNING_FORMAL = "运行仍在执行，停在人工审批时才能质询" +CARD_MOVED = "这张审批卡已处理，运行正在继续执行，请刷新页面查看最新状态" NO_CARD = "运行没有停在人工审批上（可能是服务重启时中断的），没有可以质询的审批" CARD_GONE = "这张审批卡已处理，或不是这次运行正在等待的审批，请刷新页面查看最新状态" CARD_DONE = "这张审批卡已处理，运行已经结束，请刷新页面查看最新状态" @@ -136,7 +157,20 @@ BAD_BODY = ("请求格式有误：需要一个对象，写明报告（report， "可选句子编号（unit）、选中的原文（quote）、审批卡（approval_id）") DOC_TAMPERED = "报告文档与哈希不一致，疑似被修改，不能据此质询" DOC_MISSING = "无法读取报告文档" -NO_ANSWER = "模型没有给出回答内容" +# 没能给出回答的说法：说发生了什么、下一步。核查 agent 的原话（「交结果」「按结构」、裸写的字段名）只进日志 +NO_ANSWER = "模型没有给出回答内容，请重新提问" +LEAKED = "回答中含有设了遮罩的值，无法遮住，已不予展示，请换个问法重新提问" +ERR_DAILY = "今日核查花费已达上限（{cap}），未能给出回答，可在设置的「核查」一节调高上限" +ERR_COST = "这次回答的花费已达上限，未能给出回答，可缩小问题范围后重新提问" +ERR_TIMEOUT = "回答超过时长上限{seconds}，未能给出回答，可缩小问题范围后重新提问" +ERR_SHAPE = "模型未能按要求给出回答，请重新提问；如反复出现，请在设置的「核查」一节更换核查模型" +ERR_SILENT = "模型没有给出回复，未能给出回答，请重新提问" +# 回答末尾的系统说明（with_note：另起一段，不挂出处、不查裸数字） +NOTE_LOST = "注：这次运行还用到了数据源{names}，已不存在、已停用或已被重建，未能查询，以上回答不含其中的数据。" +NOTE_ROUNDS = "注：调用模型的轮数已用完，以上回答只依据已经查到的结果，可能不完整。" +NOTE_FAILED = "注：有 {n} 次查询执行失败（查询清单只列成功的查询），以上回答可能缺少依据。" +NOTE_UNRESOLVED = "注：回答中有 {n} 处引用无法解析，相关的数没有出处。" +NOTE_LIVE = "注：以上回答依据的是数据源当前的数据，晚于报告生成的时间；与报告不一致时，可能是报告生成之后数据有更新。" class InquiryRefused(Exception): @@ -191,6 +225,32 @@ def inquiry_busy(run_id: str) -> bool: return _ANSWERING.busy(run_id) +class _Held: + """没写进运行记录的回答（回答期间运行状态变了）：进程内留 HELD_TTL_S 秒、最多 HELD_MAX 条，先进先出。 + + 提问人手里那份回答里的数要点得开，质询工件接口在这段时间里照样认它引用的工件（遮罩照旧）。过了时间、或者服务 + 重启过，接口照常回 404：回答本身已经交回去了，只是数字不能再下钻。""" + + def __init__(self) -> None: + self._items: OrderedDict[int, tuple[str, float, dict[str, Any]]] = OrderedDict() + self._next = 0 + + def keep(self, run_id: str, data: dict[str, Any]) -> None: + self._next += 1 + self._items[self._next] = (run_id, time.monotonic() + HELD_TTL_S, data) + while len(self._items) > HELD_MAX: + self._items.popitem(last=False) + + def of(self, run_id: str) -> list[dict[str, Any]]: + now = time.monotonic() + for key in [k for k, (_, until, _) in self._items.items() if until <= now]: + del self._items[key] + return [data for owner, _, data in self._items.values() if owner == run_id] + + +_HELD = _Held() + + # -------------------------------------------------------------------------- # 请求 # -------------------------------------------------------------------------- @@ -253,7 +313,8 @@ class _Opening: """开放了的这一次质询：往哪种运行里写（mode）、写的时候要核对什么（approval_id / seal），以及被问的报告文档。 loader 是取证据的函数：封存之后只认封存范围内的工件（同按需裁判，api/evidence._sealed_loader），停在审批时和 - 报告节点里的裁判一样直接读工件库。events 是这次运行的事件（封存之后只取封存范围内的）。""" + 报告节点里的裁判一样直接读工件库。events 是这次运行的事件（封存之后只取封存范围内的）。finished 是运行结束的时间 + （UTC，封存之后的追问才有）：报告是那时生成的，之后数据可能有更新。""" run_id: str mode: str @@ -264,6 +325,7 @@ class _Opening: loader: Callable[[str], Any] approval_id: str | None = None seal: tuple[int | None, str | None] | None = None + finished: str | None = None async def _open(run_id: str, ask: Ask) -> _Opening: @@ -275,6 +337,7 @@ async def _open(run_id: str, ask: Ask) -> _Opening: 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 + finished = _utc(run.finished_at) if status == "interrupted": cards = [str(a) for a in (await session.execute( select(Approval.id).where(Approval.run_id == run_id, Approval.status == "pending"))).scalars()] @@ -293,7 +356,9 @@ async def _open(run_id: str, ask: Ask) -> _Opening: 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) + # 带着审批卡来的，多半是卡刚被别人批了、运行往下走了：叫人刷新。没带的按运行类别说什么时候能问 + raise _not_open(CARD_MOVED if ask.approval_id is not None + else STILL_RUNNING_FORMAL if run_class == "formal" else STILL_RUNNING) if ask.approval_id is not None: raise _not_open(CARD_DONE) if run_class == "formal": @@ -318,7 +383,17 @@ async def _open(run_id: str, ask: Ask) -> _Opening: # 带期 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)) + doc=report.doc, loader=loader, seal=(sealed.run_manifest_seq, sealed.manifest_hash), + finished=finished) + + +def _utc(ts: datetime | None) -> str | None: + """运行记录上的时间 → UTC 的「2026-10-06T08:12:33」。带时区的换成 UTC，不带的本来就是 UTC。""" + if ts is None: + return None + if ts.tzinfo is not None: + ts = ts.astimezone(timezone.utc).replace(tzinfo=None) + return ts.isoformat(timespec="seconds") def _node_title(graph: dict[str, Any], node_id: str) -> str: @@ -430,17 +505,43 @@ def _run_sources(events: Events) -> list[str]: return out -async def _usable_sources(names: Iterable[str]) -> dict[str, list[str]]: - """现在能查的数据源（存在、没有停用，核查建工具时也只认这样的）→ 它现在设了遮罩的列。""" +async def _usable_sources(run_id: str, names: Iterable[str], versions: Any) -> dict[str, list[str]]: + """这次运行用过、现在还能查的数据源 → 它现在设了遮罩的列。 + + 工具按名字建（核查建工具时只认存在、没有停用的源），名字背后的源却可能换过：运行用过的源被删了、又用同一个名字 + 新建一个，查到的已经不是这次运行的数据。所以按名字找到现在的源之后，再核对是不是运行用过的那一个——运行里的查询 + 快照在工件引用的 meta 里记着源 id（质询自己的查询不算，它们的 node_id 为空），上传表格还记在运行固定的版本里 + （data_versions）。没固定过版本的上传表格也不交给 agent：建工具时会把当前版本补固定进运行记录（tools/datasource + 的 _pin_late），质询不发事件，已经封存的运行就被悄悄改了。升级前发起的运行（data_versions 为 None）照旧按当前版本。""" from app.data.engine import masked_columns wanted = list(dict.fromkeys(names)) if not wanted: return {} + pins = versions if isinstance(versions, dict) else None async with SessionLocal() as session: - rows = await session.execute(select(DataSource.name, DataSource.options) - .where(DataSource.name.in_(wanted), DataSource.enabled.is_(True))) - return {str(name): masked_columns(options) for name, options in rows.all()} + rows = (await session.execute( + select(DataSource.id, DataSource.name, DataSource.origin, DataSource.options) + .where(DataSource.name.in_(wanted), DataSource.enabled.is_(True)))).all() + metas = (await session.execute( + select(Artifact.meta).where(Artifact.run_id == run_id, Artifact.kind == QUERY_SNAPSHOT_KIND, + func.coalesce(Artifact.node_id, "") != ""))).scalars().all() + used: dict[str, set[str]] = {} + for meta in metas: + if isinstance(meta, dict) and meta.get("source") in wanted and meta.get("source_id"): + used.setdefault(str(meta["source"]), set()).add(str(meta["source_id"])) + for source_id, entry in (pins or {}).items(): + if isinstance(entry, dict) and entry.get("name") in wanted: + used.setdefault(str(entry["name"]), set()).add(str(source_id)) + out: dict[str, list[str]] = {} + for source_id, name, origin, options in rows: + known = used.get(str(name)) + if known and str(source_id) not in known: + continue + if (origin or "manual") == "upload" and pins is not None and source_id not in pins: + continue + out[str(name)] = masked_columns(options) + return out def _quoted(names: Iterable[str]) -> str: @@ -472,15 +573,19 @@ def _task(op: _Opening, ask: Ask, target: _Target | None, excerpts: list[str], r if lines else "") sources = "、".join(live) if live else "（没有可以查询的数据源，只能依据上面的报告内容作答）" if lost: - sources += f"（这次运行还用到了{_quoted(lost)}，已不存在或已停用，无法查询）" + sources += f"（这次运行还用到了{_quoted(lost)}，已不存在、已停用或已被重建，无法查询）" + # 封存之后的追问：报告是运行结束时生成的，现在查到的数据可能已经变了 + since = (f"报告生成于 {op.finished}（UTC），之后数据可能有更新：查询结果和报告对不上时，先说明这一点。\n\n" + if op.mode == "sealed" and op.finished else "") hidden = [f"「{name}」的 {'、'.join(cols)}" for name, cols in live.items() if cols] - masked = f"设了遮罩的列（回答里不要写出这些列的值）：{'；'.join(hidden)}\n\n" if hidden else "" + masked = (f"设了遮罩的列（查询里不要选出原值、不要起别名，回答里不要写出这些列的值）：{'；'.join(hidden)}\n\n" + if hidden else "") return INQUIRY_TASK_PROMPT.format(report=op.title, where=where, evidence=evidence, refs=refs_text, - sources=sources, masked=masked, question=ask.question) + sources=sources, since=since, masked=masked, question=ask.question) # -------------------------------------------------------------------------- -# 遮罩 +# 遮罩：哪些列 # -------------------------------------------------------------------------- @@ -505,66 +610,195 @@ def _masked_word() -> str: return MASKED -def masked_snapshot(snapshot: dict[str, Any], hidden: set[str]) -> dict[str, Any]: - """遮罩列的每一格换成「已遮罩」，这几列的类型记成文本（渲染时不当数）。没有要遮的列原样返回。""" +def _value_columns(node: Any) -> set[str]: + """一个选出列的值取自哪些列（列名小写）。COUNT 里的只数个数，窗口的分区键、排序键只决定次序，都带不出值，不算； + 套在里面的子查询只看它选出来的列（它的条件、连接里写的列带不出值）。""" + from sqlglot import exp + + out: set[str] = set() + stack = [node] + while stack: + item = stack.pop() + if isinstance(item, exp.Column): + if not isinstance(item.this, exp.Star) and item.name: + out.add(item.name.lower()) + elif isinstance(item, exp.Count): + continue + elif isinstance(item, exp.Window): + stack.append(item.this) + elif isinstance(item, exp.Select): + stack.extend(item.expressions) + else: + stack.extend(item.iter_expressions()) + return out + + +def _is_star(item: Any) -> bool: + from sqlglot import exp + + return isinstance(item, exp.Star) or (isinstance(item, exp.Column) and isinstance(item.this, exp.Star)) + + +def _branches(root: Any) -> list[list[Any]]: + """最外层选出的列：一条 SELECT 一支，UNION 这类的每一支各一支（结果的第 i 列来自每一支的第 i 项）。""" + from sqlglot import exp + + out: list[list[Any]] = [] + stack = [root] + while stack: + node = stack.pop() + if isinstance(node, exp.Subquery): + stack.append(node.this) + elif isinstance(node, exp.SetOperation): + stack.extend((node.expression, node.this)) + elif isinstance(node, exp.Select): + out.append(list(node.expressions)) + return out + + +def _tainted(sql: str, kind: str | None, masked: set[str]) -> tuple[set[str], set[int]] | None: + """SQL 里指代遮罩列的名字（小写），和最外层选出的列里值取自遮罩列的那几列的位置。解析不了返回 None。 + + 名字从遮罩列本身起，一层层传下去：选出列的值取自这些名字的（buyer_phone AS 手机号、UPPER(buyer_phone) AS x），它的 + 别名也算；派生来源写了列名表的（WITH t(p) AS (SELECT buyer_phone …)），按位置对上的列名也算，直到不再多出名字。 + 同名不同层的也一并算上（宁可多遮）。位置是给没起别名的表达式用的：UPPER(buyer_phone) 的结果列名对不上任何名字。""" + from sqlglot import exp + + from app.data.sql_semantics import _parse, _read + + try: + root, _ = _parse(sql, _read(kind)) + except Exception: # noqa: BLE001 - 解析器的意外按解析不了处理 + root = None + if root is None: + return None + try: + names = set(masked) + selects = list(root.find_all(exp.Select)) + lists = [a for a in root.find_all(exp.TableAlias) if a.columns + and isinstance(a.parent, (exp.CTE, exp.Subquery)) and isinstance(a.parent.this, exp.Select)] + for _ in range(len(selects) + len(lists) + 1): + before = len(names) + for select_ in selects: + for item in select_.expressions: + name = item.alias_or_name.lower() + if name and name not in names and names & _value_columns(item): + names.add(name) + for alias in lists: + for column, item in zip(alias.columns, alias.parent.this.expressions): + name = column.name.lower() + if name and name not in names and (names & _value_columns(item) + or item.alias_or_name.lower() in names): + names.add(name) + if len(names) == before: + break + positions: set[int] = set() + for branch in _branches(root): + if not any(_is_star(item) for item in branch): + positions |= {i for i, item in enumerate(branch) if names & _value_columns(item)} + return names, positions + except Exception: # noqa: BLE001 - 分析本身的缺陷不能让回答出错：按解析不了处理（宁可多遮） + logger.warning("质询的 SQL 遮罩分析失败", exc_info=True) + return None + + +def _mentions(sql: str, names: Iterable[str]) -> bool: + return any(re.search(rf"(? tuple[set[str], set[str]]: + """(这份快照要遮的列, SQL 里指代遮罩列的名字)，都是小写。 + + 证据面板只按列名遮（hidden）；agent 的查询起了别名、套了函数，结果列名就对不上。本仓库的查询惯例是起中文别名 + （region AS 区域），所以按 SQL 认出改了名字的遮罩列（_tainted）。SQL 解析不了、里面又写着遮罩列的，整份快照的 + 列都遮掉；合并查询的快照不看 SQL，它记着自己的来历（_hidden_columns 已按 _merge_masks 遮好）。""" + base = set(hidden) + sql = snapshot.get("sql") + if not base or not isinstance(sql, str) or not sql.strip() or snapshot.get("source") == MERGE_SOURCE: + return base, base 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} + found = _tainted(sql, kind, base) + if found is None: + return (base | {c.lower() for c in columns}, base) if _mentions(sql, base) else (base, base) + names, positions = found + return names | {columns[i].lower() for i in positions if i < len(columns)}, names -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 +#: 一个数的写法：可带正负号、千分位、小数、百分号 +_NUMERIC = re.compile(r"[-+]?(?:\d{1,3}(?:,\d{3})+|\d+)(?:\.\d+)?%?") +#: 正文里一个个「词」：字母数字下划线连成的一串。头尾都是这种字的遮罩值，只能整个对上这样一串 +_WORDS = re.compile(r"[0-9A-Za-z_]+") +#: 正文里的一串数（可带千分位、小数）：带千分位的遮罩值按去掉逗号的样子比 +_DIGITS = re.compile(r"\d[\d,]*(?:\.\d+)?") +#: 中文这类遮罩值按开头几个字分桶找（_Secrets._partition） +_PREFIX = 8 - return load + +def _plain_text(value: Any) -> str: + text = json.dumps(value, ensure_ascii=False, default=str) if isinstance(value, (dict, list)) else str(value) + return re.sub(r"\s+", " ", text).strip() + + +def _numeric(value: Any) -> bool: + if isinstance(value, bool): + return False + return isinstance(value, (int, float)) or (isinstance(value, str) and numeric_text(value) is not None) 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: + text = _plain_text(value) + forms = {text} if text else set() + if _numeric(value): with contextlib.suppress(RenderError, ValueError, ArithmeticError): forms.add(render_cell(value)) - return {f for f in forms if len(f) >= SECRET_MIN} + return forms + + +def _strong(form: str) -> bool: + """按值认得出的写法（见 SECRET_MIN、SECRET_DIGITS）。""" + if _NUMERIC.fullmatch(form): + return sum(c.isdigit() for c in form) >= SECRET_DIGITS + return len(form) >= SECRET_MIN + + +def _digits_of(form: str) -> str: + return form.replace(",", "").lstrip("+-").rstrip("%") def _pattern(value: str) -> str: - """一个值的正则：头尾是字母数字的按词边界认，数字还不能是更长的数的一截（1,390 里的 390、12.5 里的 2.5）。""" + """一个值的正则：头尾是字母数字的按词边界认；数字还不能是更长的数、日期、时间的一截（12.5 里的 2.5、2026-01-25 里的 + 25、12:30 里的 30），也不能是千分位写法里的一组（1,390 里的 390）。逗号隔开的一串长号码（13900000001,13900000002） + 各自照认：千分位一组只有三位。""" head = tail = "" if _WORD.match(value[0]): - head = r"(? None: if not _query_shaped(snapshot) or not hidden: @@ -572,8 +806,8 @@ class _Secrets: 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 len(self.weak) >= SECRETS_MAX: + break if isinstance(row, list): cells = [row[i] for i in index if i < len(row)] elif isinstance(row, dict): @@ -581,44 +815,360 @@ class _Secrets: else: continue for value in cells: - if value is not None and not isinstance(value, bool): - self.values |= _forms(value) + if value is None or isinstance(value, bool): + continue + for form in _forms(value): + self.weak.add(form) + if _strong(form): + self.strong.add(form) + self._index = None - 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 protect(self, texts: Iterable[str | None]) -> None: + """提问人自己写进问题、选中原文里的值不换（「13900000001 这位客户」「排除 300 以下的」）：问题本身就带着它 + （事件里原样记着），换了只会让回答、查询对不上问题。""" + given = " ".join(t for t in texts if t) + if not given: + return + numbers = {_digits_of(n) for n in _DIGITS.findall(given)} + + def asked(form: str) -> bool: + return _digits_of(form) in numbers if _NUMERIC.fullmatch(form) else form in given + + self.strong = {f for f in self.strong if not asked(f)} + self.weak = {f for f in self.weak if not asked(f)} + self._index, self._compiled = None, {} + + def hit(self, value: Any) -> bool: + """一格（或 SQL 里一个字面量）的值是不是按值认得出的遮罩值。短的数不算（见 SECRET_DIGITS）。""" + if value is None or isinstance(value, bool) or not self.strong: + return False + text = _plain_text(value) + if text in self.strong: + return True + if _numeric(value) and sum(c.isdigit() for c in text) >= SECRET_DIGITS: + with contextlib.suppress(RenderError, ValueError, ArithmeticError): + return render_cell(value) in self.strong + return False + + def weak_hit(self, text: str) -> bool: + """SQL 里一个字面量是不是遮罩列里出现过的值（短的数也算）。""" + return _plain_text(text) in self.weak + + def scrub(self, text: str) -> str: + if not text or not self.strong: + return text + pattern = self.pattern(text) + return pattern.sub(_masked_word(), text) if pattern is not None else text + + def pattern(self, text: str) -> re.Pattern[str] | None: + """认 text 里出现的遮罩值的正则；一个都没有时为 None。""" + found = self._found(text) if text and self.strong else set() + if not found: + return None + key = frozenset(found) + if key not in self._compiled: + if len(self._compiled) > 64: + self._compiled = {} + # 长的在前：13900000001 先于它的一截被认出来 + self._compiled[key] = re.compile("|".join(_pattern(v) for v in sorted(found, key=lambda v: (-len(v), v)))) + return self._compiled[key] + + def _found(self, text: str) -> set[str]: + words, numbers, others = self._partition() + found = {w for w in _WORDS.findall(text) if w in words} + if numbers: + for run in _DIGITS.findall(text): + found |= numbers.get(_digits_of(run), set()) + for i in range(len(text) - 1) if others else (): + for size, buckets in others.items(): + for length, values in buckets.get(text[i:i + size], {}).items(): + if (piece := text[i:i + length]) in values: + found.add(piece) + return found + + def prepare(self) -> None: + """先把索引建好（收完值、去掉提问人写的之后）：上万个值建一次要几十毫秒，放在线程里建，别等到事件循环里第一次换。""" + self._partition() + + def _partition(self) -> tuple[set[str], dict[str, set[str]], dict[int, dict[str, dict[int, set[str]]]]]: + """strong 分成三类：只有字母数字的（按词比）、带千分位小数的数（按去掉逗号的数比）、其余的（中文、日期、 + 邮箱）。其余的按开头至多 _PREFIX 个字分桶、桶里再按长短分：同一个市区的上万个地址开头七八个字都一样，只按开头 + 分的话，正文里每碰到一次那几个字都要逐个比；再按长短分，碰到一次只是按几种长短各查一下。三类各自只在正文里 + 真出现时才进正则。""" + if self._index is None: + words: set[str] = set() + numbers: dict[str, set[str]] = {} + others: dict[int, dict[str, dict[int, set[str]]]] = {} + for value in self.strong: + if _WORDS.fullmatch(value): + words.add(value) + elif _NUMERIC.fullmatch(value): + numbers.setdefault(_digits_of(value), set()).add(value) + else: + size = min(len(value), _PREFIX) + others.setdefault(size, {}).setdefault(value[:size], {}).setdefault(len(value), set()).add(value) + self._index = (words, numbers, others) + return self._index -def _outside_markers(text: str, scrub: Callable[[str], str]) -> str: +def _outside_markers(text: str, secrets: _Secrets) -> str: """只换引用标记之外的文字：标记里是编号、行号、列名，换了会把一个好好的引用弄坏。""" + pattern = secrets.pattern(text) + if pattern is None: + return text + spans = [m.span() for m in MARKER_RE.finditer(text)] + word = _masked_word() + return pattern.sub(lambda m: m.group(0) if any(s < m.end() and m.start() < e for s, e in spans) else word, text) + + +def _masked_view(snapshot: dict[str, Any], hidden: set[str], secrets: _Secrets) -> tuple[dict[str, Any], list[str]]: + """一份查询快照遮好的样子和遮掉的列：要遮的列（hidden，小写）每一格都换成「已遮罩」、类型记成文本（渲染时不当数）； + 别的列里写着按值认得出的遮罩值的格子也换（另一张表里没设遮罩的同一个手机号、改了名字没认出来的遮罩列）。 + 原快照不动；没有要遮的返回原快照。""" + columns = [str(c) for c in snapshot.get("columns") or []] + index = {i for i, c in enumerate(columns) if c.lower() in hidden} + by_value = bool(secrets.strong) + if not index and not by_value: + return snapshot, [] + word = _masked_word() + touched = set(index) + position = {c: i for i, c in enumerate(columns)} + + def cell(i: int | None, value: Any) -> Any: + if i in index: + return word + if by_value and secrets.hit(value): + if i is not None: + touched.add(i) + return word + return value + + rows: list[Any] = [] + for row in snapshot["rows"]: + if isinstance(row, list): + rows.append([cell(i, v) for i, v in enumerate(row)]) + elif isinstance(row, dict): + rows.append({k: word if str(k).lower() in hidden else cell(position.get(str(k)), v) + for k, v in row.items()}) + else: + rows.append(row) + if not touched: + return snapshot, [] + 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": rows, "column_types": types}, [columns[i] for i in sorted(touched)] + + +_COMPARE_TOKENS = frozenset({"EQ", "NEQ", "GT", "GTE", "LT", "LTE", "LIKE", "ILIKE", "GLOB", "RLIKE", "NULLSAFE_EQ"}) +_LITERAL_TOKENS = frozenset({"STRING", "NUMBER"}) +_NAME_TOKENS = frozenset({"VAR", "IDENTIFIER"}) + + +def _against(tokens: list[Any], i: int, compared: set[str]) -> bool: + """第 i 个字面量是不是直接和遮罩列比较：col = 'x'、col IN ('x', …)、col NOT LIKE 'x'、col BETWEEN a AND b， + 以及 'x' = col。列可以带表名（s.buyer_phone），可以是遮罩列的别名（compared 里已经有）。""" + kinds = [t.token_type.name for t in tokens] + j = i - 1 + k = j + while k >= 0 and kinds[k] in ("COMMA", "STRING", "NUMBER", "DASH"): + k -= 1 + if k >= 1 and kinds[k] == "L_PAREN" and kinds[k - 1] == "IN": + j = k - 2 + elif j >= 0 and kinds[j] in _COMPARE_TOKENS: + j -= 1 + elif j >= 2 and kinds[j] == "AND" and kinds[j - 1] in _LITERAL_TOKENS and kinds[j - 2] == "BETWEEN": + j -= 3 + elif j >= 0 and kinds[j] == "BETWEEN": + j -= 1 + else: + # 字面量在左边：'x' = col、'x' = t.col + if i + 2 < len(tokens) and kinds[i + 1] in _COMPARE_TOKENS and kinds[i + 2] in _NAME_TOKENS: + last = i + 2 + while last + 2 < len(tokens) and kinds[last + 1] == "DOT" and kinds[last + 2] in _NAME_TOKENS: + last += 2 + return tokens[last].text.lower() in compared + return False + if j >= 0 and kinds[j] == "NOT": + j -= 1 + return j >= 0 and kinds[j] in _NAME_TOKENS and tokens[j].text.lower() in compared + + +def _replace_literal(raw: str, kind: str, change: Callable[[str], str]) -> str: + """字面量的原文 → 换过的原文：字符串只换引号里面的（引号、N'' 这类前缀照留），数整个换。""" + if kind == "NUMBER": + return change(raw) + for quote in ("'", '"'): + start = raw.find(quote) + if start >= 0 and len(raw) - start >= 2 and raw.endswith(quote): + return raw[:start + 1] + change(raw[start + 1:-1]) + quote + return change(raw) + + +def _scrub_sql(sql: str, secrets: _Secrets, compared: set[str], kind: str | None) -> str: + """SQL 里遮罩列的值换成「已遮罩」，别的一个字不动：查询清单随运行封存，它得是实际执行的那条，审批人看 what-if 的 + 条件（阈值、日期、LIMIT、别的列的条件）看的就是它。 + + - 字符串字面量、注释里按值换（同正文：够长的数、两个字以上的文字）；数字字面量够长、又是遮罩值的整个换； + - 直接和遮罩列（或它的别名，compared）比较的字面量是遮罩列里出现过的值就换，短的数也换（age = 25）。 + 分词出错（引号没闭合这类）就整条按正文的办法换。""" + if not sql or not (secrets.strong or secrets.weak): + return sql + from app.data.sql_semantics import _read, _tokens + + try: + tokens = _tokens(sql, _read(kind)) + except Exception: # noqa: BLE001 - 分词器认不了的写法：按正文的办法换 + return secrets.scrub(sql) + word = _masked_word() 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:])) + for i, token in enumerate(tokens): + start, end = token.start, token.end + 1 + if start < last or end > len(sql): + continue + gap = sql[last:start] + out.append(secrets.scrub(gap) if gap.strip() else gap) + raw = sql[start:end] + kind_of = token.token_type.name + if kind_of in _LITERAL_TOKENS: + if compared and secrets.weak_hit(token.text) and _against(tokens, i, compared): + raw = _replace_literal(raw, kind_of, lambda _: word) + elif kind_of == "STRING": + raw = _replace_literal(raw, kind_of, secrets.scrub) + elif secrets.hit(token.text): + raw = word + out.append(raw) + last = end + rest = sql[last:] + out.append(secrets.scrub(rest) if rest.strip() else rest) return "".join(out) +@dataclass +class _Plan: + """一次回答（或一次取质询工件）的遮罩：agent 每份查询快照遮好的样子（views）和遮掉的列（masked）、SQL 里指代 + 遮罩列的名字（tainted）、数据源的类型（kinds，SQL 按它的方言分词）、遮罩列里出现过的值（secrets），以及已经不在的 + 数据源（gone，按查询当时记下的遮罩处理）。""" + + views: dict[str, dict[str, Any]] + masked: dict[str, list[str]] + tainted: dict[str, set[str]] + kinds: dict[str, str | None] + secrets: _Secrets + gone: dict[str, list[str]] + + +async def _source_kinds(names: Iterable[str]) -> dict[str, str | None]: + wanted = sorted({n for n in names if n}) + if not wanted: + return {} + async with SessionLocal() as session: + rows = (await session.execute(select(DataSource.name, DataSource.kind) + .where(DataSource.name.in_(wanted)))).all() + return {str(name): kind for name, kind in rows} + + +async def _plan(own: dict[str, dict[str, Any]], others: list[dict[str, Any]], given: list[str | None]) -> _Plan: + """own：agent 自己的查询快照（工件 id → 内容）；others：报告自己的查询快照（被问的那句可能直接写着遮罩列的值， + agent 会照着复述、写进查询条件）；given：问题和选中的原文（里面的值不换）。 + + 查数据源的遮罩要读库，在这里读；解析 SQL、收上万个值、逐格比对是纯计算，放进线程：放在事件循环里会挡住同一进程 + 里别的运行（同 tools/datasource 的 _sql_checks）。""" + from app.api.evidence import _Masks + + masks = _Masks() + named = {artifact: await _hidden_columns(snapshot, masks) for artifact, snapshot in own.items()} + based = [(snapshot, (await _hidden_columns(snapshot, masks))[0]) for snapshot in others] + kinds = await _source_kinds(str(s.get("source") or "") for s in [*own.values(), *others]) + return await asyncio.to_thread(_settle_plan, own, named, based, kinds, given) + + +def _settle_plan(own: dict[str, dict[str, Any]], named: dict[str, tuple[set[str], list[str]]], + based: list[tuple[dict[str, Any], set[str]]], kinds: dict[str, str | None], + given: list[str | None]) -> _Plan: + secrets = _Secrets() + hidden: dict[str, set[str]] = {} + tainted: dict[str, set[str]] = {} + for artifact, snapshot in own.items(): + hidden[artifact], tainted[artifact] = _derived_hidden(snapshot, named[artifact][0], + kinds.get(str(snapshot.get("source") or ""))) + secrets.add(snapshot, hidden[artifact]) + for snapshot, base in based: + secrets.add(snapshot, _derived_hidden(snapshot, base, kinds.get(str(snapshot.get("source") or "")))[0]) + secrets.protect(given) + secrets.prepare() + views: dict[str, dict[str, Any]] = {} + masked: dict[str, list[str]] = {} + for artifact, snapshot in own.items(): + views[artifact], masked[artifact] = _masked_view(snapshot, hidden[artifact], secrets) + return _Plan(views=views, masked=masked, tainted=tainted, kinds=kinds, secrets=secrets, + gone={artifact: named[artifact][1] for artifact in own}) + + +def _masking_loader(views: dict[str, dict[str, Any]]) -> Callable[[str], Any]: + """组装回答文档用的 loader：agent 自己的查询快照交遮好的那份（引用到遮罩的格子渲染成「已遮罩」）；别的查询快照 + 不交（回答的目录里只有 agent 自己的查询）；口径卡这类没有行的原样，哈希照旧复验。""" + + def load(artifact: str) -> Any: + if artifact in views: + return views[artifact] + content = artifact_store.load(artifact) + return None if _query_shaped(content) else content + + return load + + # -------------------------------------------------------------------------- # 回答 # -------------------------------------------------------------------------- +class _Leak(Exception): + """组装了两遍，正文里还认得出遮罩值（渲染出来才露的，比如恰好是遮罩值的运行输入）：这份回答不交出去。""" + + def _limit(outcome: VerifyOutcome) -> str | None: - if outcome.limit in LIMITS: - return outcome.limit - return "max_queries" if outcome.limit == "max_steps" else None + return outcome.limit if outcome.limit in LIMITS else None + + +def _error_words(outcome: VerifyOutcome, *, timeout_s: Any, daily_max_usd: Any) -> str | None: + """核查 agent 没交成结果的原话 → 提问人看得懂的话（发生了什么、下一步）。原话里有内部的说法（「交结果」「按结构」、 + 裸写的字段名），只进日志；模型不可用、调用失败的说明本来就是给人看的，照用。""" + error = outcome.error + if not error: + return None + if error.startswith("已达") and error.endswith("没能交结果"): + if "每日" in error: + cap = f"核查每日金额上限 {judging._usd(daily_max_usd)}" if daily_max_usd is not None else "核查每日金额上限" + return ERR_DAILY.format(cap=cap) + if "时长" in error: + return _timeout_words(timeout_s) + return ERR_COST + if error.startswith("收尾时超过时长上限"): + return _timeout_words(timeout_s) + if error.startswith("核查模型没有按结构交结果"): + return ERR_SHAPE + if error == "核查模型没有给出回复": + return ERR_SILENT + if error.startswith("核查出错（"): + return f"回答时出错{error[len('核查出错'):]}，请重新提问" + return error + + +def _timeout_words(timeout_s: Any) -> str: + seconds = timeout_s if isinstance(timeout_s, (int, float)) and not isinstance(timeout_s, bool) else None + return ERR_TIMEOUT.format(seconds=f"（{seconds:g} 秒）" if seconds is not None else "") + + +def _given_numbers(ask: Ask) -> list[str]: + """问题、选中的原文里的数：提问人给的阈值、条件（「排除金额低于 500 的」），回答里复述它们不算没有出处的结论。 + 查询结果里来的数照旧必须引用。""" + return [token.raw for text in (ask.question, ask.quote) if text for token in extract_numbers(text)] 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 @@ -627,7 +1177,7 @@ async def _answer(op: _Opening, ask: Ask, target: _Target | None, settings: dict 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) + usable = await _usable_sources(op.run_id, wanted, op.versions) live, lost = {s: usable[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: @@ -638,41 +1188,47 @@ async def _answer(op: _Opening, ask: Ask, target: _Target | None, settings: dict 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() + own: dict[str, dict[str, Any]] = {} 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() + own[str(query["artifact"])] = snapshot + others = [snapshot for entry in catalog.values() if isinstance(entry, dict) and entry.get("kind") == "query" + and _query_shaped(snapshot := _quiet_load(op.loader, entry.get("artifact")))] + plan = await _plan(own, others, [ask.question, ask.quote]) text, doc_id, failed = "", None, None source = (outcome.rationale or {}).get("source") if outcome.result is not None else None if isinstance(source, str) and source.strip(): try: - answer = _compose(source, {**refs, **((outcome.rationale or {}).get("catalog") or {})}, op.run_id, - hidden, scrub) + answer = _compose(source, {**refs, **((outcome.rationale or {}).get("catalog") or {})}, run_id=op.run_id, + views=plan.views, secrets=plan.secrets, allow=_given_numbers(ask)) + except _Leak: + logger.warning("质询的回答里有遮罩值遮不住，不交出去（运行 %s）", op.run_id) + failed = LEAKED except Exception as e: # noqa: BLE001 - 原文没组装成文档就不交出去：里面的引用没换成值，遮罩也没做 logger.warning("质询的回答组装失败（运行 %s）", op.run_id, exc_info=True) - failed = f"整理回答时出错（{describe_exception(e)}）" + failed = f"整理回答时出错（{describe_exception(e)}），请重新提问" else: - if lost: - answer = with_note(answer, f"注：这次运行还用到了数据源{_quoted(lost)}，已不存在或已停用，未能查询，" - "以上回答不含其中的数据。") + answer = _with_notes(answer, op, outcome, part, lost, own) 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] + error = None + if not ok: + error = _error_words(outcome, timeout_s=part.get("timeout_s"), daily_max_usd=settings.get("daily_max_usd")) + if outcome.error: + logger.info("质询没能给出回答（运行 %s）：%s", op.run_id, outcome.error) + queries = [] + for q in outcome.queries: + artifact, name = str(q.get("artifact") or ""), str(q.get("source") or "") + # 快照没取到的（存不下、被删了），SQL 里按这个源设的遮罩列认比较 + compared = plan.tainted.get(artifact) or {c.lower() for c in live.get(name, [])} + queries.append({**{k: q.get(k) for k in QUERY_KEYS}, + "sql": _scrub_sql(str(q.get("sql") or ""), plan.secrets, compared, plan.kinds.get(name))}) return { "answer_text": text, "answer_doc": doc_id, "ok": ok, "limit": _limit(outcome), - "error": None if ok else (outcome.error or failed or NO_ANSWER), + "error": None if ok else (error or failed or NO_ANSWER), # 结果一模一样的两次查询是同一份快照（工件按内容寻址）：引用列表里只记一次 "inquiry_artifacts": list(dict.fromkeys(q["artifact"] for q in queries if q.get("artifact"))), "queries": queries, @@ -690,23 +1246,54 @@ def _quiet_load(loader: Callable[[str], Any], artifact: Any) -> Any: 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]: +def _compose(source: str, catalog: dict[str, Any], *, run_id: str, views: dict[str, dict[str, Any]], + secrets: _Secrets, allow: list[str]) -> dict[str, Any]: """agent 写的回答原文（带引用标记）→ 回答文档：目录是 agent 自己的查询，加上报告里的指标和运行输入；快照按遮罩 - 遮好再渲染。遮罩列的值先在标记之外的文字里换掉；换完正文里还认得出遮罩列的值（标记里写了值，解析不了，原样 - 出现在正文里），就连标记一起换了再组装一遍——宁可弄坏一个引用，也不让值露出来。""" - loader = _masking_loader(hidden) - doc = compose_doc(_outside_markers(source, scrub), catalog, run_id=run_id, loader=loader) + 遮好再渲染。遮罩值先在标记之外的文字里换掉；换完正文里还认得出遮罩值（标记里写了值，解析不了，原样出现在正文里）， + 就连标记一起换了再组装一遍——宁可弄坏一个引用，也不让值露出来；还认得出（渲染出来才露的），抛 _Leak，不交出去。 + allow 是提问人自己写的数（_given_numbers），回答里复述它们不标「无出处」。""" + loader = _masking_loader(views) + allowed = list(allow) or None + doc = compose_doc(_outside_markers(source, secrets), catalog, run_id=run_id, loader=loader, allow_numbers=allowed) markdown = str(doc.get("markdown") or "") - if scrub(markdown) != markdown: - doc = compose_doc(scrub(source), catalog, run_id=run_id, loader=loader) + if secrets.scrub(markdown) != markdown: + doc = compose_doc(secrets.scrub(source), catalog, run_id=run_id, loader=loader, allow_numbers=allowed) + markdown = str(doc.get("markdown") or "") + if secrets.scrub(markdown) != markdown: + raise _Leak() return doc +def _with_notes(answer: dict[str, Any], op: _Opening, outcome: VerifyOutcome, part: dict[str, Any], lost: list[str], + own: dict[str, dict[str, Any]]) -> dict[str, Any]: + """回答末尾的系统说明：用不了的数据源、模型的轮数用完了、有查询执行失败、有引用解析不了、封存之后查的是现在的数据。 + 审批人据此知道这份回答哪里不完整，不会看到「回答完成」却满屏 ⟦?…⟧。""" + notes: list[str] = [] + if lost: + notes.append(NOTE_LOST.format(names=_quoted(lost))) + if outcome.limit == "max_steps": + notes.append(NOTE_ROUNDS) + cap = part.get("max_queries") + # 查询次数到顶时次数一定用满了（被拒的、执行失败的也算一次），清单里少的就是失败的那几次 + if outcome.limit == "max_queries" and isinstance(cap, int) and not isinstance(cap, bool) \ + and len(outcome.queries) < cap: + notes.append(NOTE_FAILED.format(n=cap - len(outcome.queries))) + unresolved = sum(1 for seg in iter_segments(answer) if seg.get("issue") == "unresolved_ref") + if unresolved: + notes.append(NOTE_UNRESOLVED.format(n=unresolved)) + # 上传表格按运行固定的版本查（快照记着 data_version），手工登记的源查的是现在的库 + if op.mode == "sealed" and any(not snapshot.get("data_version") for snapshot in own.values()): + notes.append(NOTE_LIVE) + for note in notes: + answer = with_note(answer, note) + return answer + + 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}) + meta={"report": ask.report, "unit": ask.unit, + "question": ask.question[:QUESTION_BRIEF]}) except Exception: # noqa: BLE001 - 存不下不影响回答本身：正文照常交回，只是数字点不开 logger.warning("质询的回答文档保存失败（运行 %s）", op.run_id, exc_info=True) return None @@ -778,6 +1365,7 @@ async def ask(run_id: str, request: Ask, *, actor: str | None = None) -> dict[st 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 + _HELD.keep(run_id, shown) return {"inquiry": inquiry_of(shown, seq=None, ts=time.time()), "recorded": False, "note": note} @@ -797,13 +1385,50 @@ async def list_inquiries(run_id: str, *, report: str | None = None, unit: str | 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 +def _cited_cards(doc_id: Any) -> set[str]: + """回答文档里引用的报告口径卡（m: 条目指向的工件）：界面按 doc.catalog[alias].artifact 取，和查询快照走同一个接口。""" + doc = _quiet_load(artifact_store.load, doc_id) + catalog = doc.get("catalog") if isinstance(doc, dict) and isinstance(doc.get("catalog"), dict) else {} + return {str(e["artifact"]) for e in catalog.values() + if isinstance(e, dict) and e.get("kind") in ("metric", "input") and e.get("artifact")} - 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] + +async def _report_snapshots(run_id: str, reports: Iterable[Any]) -> list[dict[str, Any]]: + """这几份报告（报告节点 id）的文档目录里的查询快照：和回答时收遮罩值的是同一批（报告最后一次 report.checked 的文档）。""" + wanted = sorted({str(r) for r in reports if isinstance(r, str) and r}) + if not wanted: + return [] + async with SessionLocal() as session: + rows = await session.execute( + select(RunEvent.node_id, RunEvent.data) + .where(RunEvent.run_id == run_id, RunEvent.type == EventType.REPORT_CHECKED, RunEvent.node_id.in_(wanted)) + .order_by(RunEvent.seq)) + docs = {node_id: str(data["doc_artifact"]) for node_id, data in rows + if isinstance(data, dict) and data.get("doc_artifact")} + out: list[dict[str, Any]] = [] + for artifact in docs.values(): + doc = _quiet_load(artifact_store.load, artifact) + catalog = doc.get("catalog") if isinstance(doc, dict) and isinstance(doc.get("catalog"), dict) else {} + for entry in catalog.values(): + if isinstance(entry, dict) and entry.get("kind") == "query": + snapshot = _quiet_load(artifact_store.load, entry.get("artifact")) + if _query_shaped(snapshot): + out.append(snapshot) + return out + + +async def inquiry_artifact(run_id: str, artifact_id: str) -> dict[str, Any]: + """质询引用过的工件：查询快照（遮罩照回答时的办法，SQL 用事件里记下的那条，两处一字不差）、回答文档（组装时就遮好 + 了）、回答里引用的报告口径卡。{id, content, masked[, mask_note]}。 + + 认的是这次运行的质询事件引用过的，加上没写进运行记录、还留在进程里的回答（_HELD）引用过的；别的一律 404。""" + from app.api.evidence import SOURCE_GONE + + datas = [data for _, _, data in await _inquiry_events(run_id)] + _HELD.of(run_id) + referenced = [d for d in datas + if artifact_id in (d.get("inquiry_artifacts") or []) or d.get("answer_doc") == artifact_id] + if not referenced: + referenced = [d for d in datas if d.get("answer_doc") and artifact_id in _cited_cards(d.get("answer_doc"))] if not referenced: raise InquiryRefused(404, "这次运行的质询没有引用这件工件", code=ARTIFACT_NOT_FOUND) try: @@ -814,21 +1439,28 @@ async def inquiry_artifact(run_id: str, artifact_id: str) -> dict[str, Any]: 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() + # 遮罩值照回答时收：引用它的那几条质询各自查到的全部快照（查询条件里写着的值，不一定在这一份的结果里），加上报告 + # 自己的查询快照；提问人自己写的值不换 + own: dict[str, dict[str, Any]] = {artifact_id: content} 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: + if other not in own and _query_shaped(snapshot := _quiet_load(artifact_store.load, other)): + own[other] = snapshot + others = await _report_snapshots(run_id, (d.get("report") for d in referenced)) + plan = await _plan(own, others, [t for d in referenced for t in (d.get("question"), d.get("quote")) + if isinstance(t, str)]) + shown = plan.views[artifact_id] + recorded = next((q["sql"] for d in referenced for q in d.get("queries") or [] + if isinstance(q, dict) and q.get("artifact") == artifact_id and isinstance(q.get("sql"), str)), + None) + if recorded is not None: + shown = {**shown, "sql": recorded} + elif isinstance(shown.get("sql"), str): + name = str(content.get("source") or "") + shown = {**shown, "sql": _scrub_sql(shown["sql"], plan.secrets, plan.tainted.get(artifact_id, set()), + plan.kinds.get(name))} + out: dict[str, Any] = {"id": artifact_id, "content": shown, "masked": plan.masked[artifact_id]} + if gone := plan.gone.get(artifact_id): out["mask_note"] = SOURCE_GONE.format(source="、".join(gone)) return out @@ -836,5 +1468,5 @@ async def inquiry_artifact(run_id: str, artifact_id: str) -> dict[str, Any]: __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", + "inquiry_busy", "inquiry_of", "list_inquiries", "parse_ask", ] diff --git a/backend/tests/test_inquiry_api.py b/backend/tests/test_inquiry_api.py index 7926b046..f5997e16 100644 --- a/backend/tests/test_inquiry_api.py +++ b/backend/tests/test_inquiry_api.py @@ -4,25 +4,32 @@ - 停在审批时质询：事件 node_id 为空、数据按前后端约定的形状；质询的查询工件只在 inquiry_artifacts / queries 里， 不在 query_artifact 下（证据接口不把它们当成报告的证据）；批准、运行结束封存之后 verify 一致，封存范围包含这条事件； 用量计入运行用量 -- 回答期间运行被批准（竞态）：回答照样交回，recorded 为假、note 说明，事件不落库 +- 回答期间运行被批准（竞态）：回答照样交回，recorded 为假、note 说明，事件不落库，回答里的工件照样取得到；两道审批时 + 问的是第一张、回答途中运行停到了第二张上，也不写 - 同一运行同时只答一条：第二条质询、审批回复（批准 / 驳回）、继续运行都回 409 inquiry_busy，拒掉的批准不留痕迹 -- 封存之后：探索运行可以追问（post_seal，落在封存之后，不计运行用量），正式运行 409；设置里关了 409； - 没停在那张审批卡上、还在执行、封存核对不通过都 409；输入不合法 422，指明是哪一项 -- 回答：先给结论，数字只能引用自己查询的格子或报告已有的指标，裸数字被标出；遮罩列的格子渲染成「已遮罩」， - 回答正文、查询清单、事件里都不出现遮罩列的值 -- 触顶、失败照样记一条事件 -- 读取：列表按报告、句子筛选，按时间排；质询工件只认质询引用过的，遮罩照证据面板 +- 封存之后：探索运行可以追问（post_seal，落在封存之后，不计运行用量，回答末尾注明查的是现在的数据），正式运行 409； + 设置里关了 409；没停在那张审批卡上、还在执行（按情形说下一步）、封存核对不通过、开放途中运行被继续了都 409； + 事件里记的报告文档不是这次运行的 422；输入不合法 422，指明是哪一项 +- 回答：先给结论，数字只能引用自己查询的格子或报告已有的指标，裸数字被标出（提问人自己写的数不标）；遮罩列（起了 + 别名的也算）的格子渲染成「已遮罩」，回答正文、查询清单、事件、质询工件里都不出现遮罩列的值；短的遮罩值不误伤阈值、 + 天数、LIMIT +- 接线：设置里的上限、运行固定的数据版本交到 agent 手里（上传表格重传之后查的仍是运行固定的那一版）；同名重建的 + 数据源、没固定过版本的上传表格不查 +- 触顶、失败照样记一条事件，说法给提问人看得懂、说下一步；轮数用完不算查询次数到顶，查询失败、引用解析不了在末尾注明 +- 读取：列表按报告、句子筛选，按时间排；质询工件只认质询引用过的（含回答里引用的口径卡），遮罩照回答时的办法 - 恢复后的重放不受影响：前面的节点不重复执行 -模型一律是 MockChatModel 剧本，数据源是现造的 SQLite 文件，区域、金额、手机号都是虚构的。 +模型一律是 MockChatModel 剧本，数据源是现造的 SQLite 文件和上传表格，区域、金额、手机号、年龄都是虚构的。 """ from __future__ import annotations import asyncio +import io import json import sqlite3 import uuid from dataclasses import dataclass, field +from datetime import timezone import pytest from httpx import ASGITransport, AsyncClient @@ -30,10 +37,11 @@ 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.core.config import settings +from app.data import introspect, table_versions from app.data.engine import engines from app.db.base import SessionLocal -from app.db.models import Approval, DataSource, Run, RunEvent, Setting +from app.db.models import Approval, Artifact, DataSource, Run, RunEvent, Setting from app.engine import compiler from app.engine import inquiry as inquiry_mod from app.engine import spend, verify_agent @@ -65,6 +73,12 @@ EVENT_KEYS = {"question", "quote", "report", "unit", "approval_id", "actor", "an QUERY_KEYS = {"alias", "artifact", "sql", "row_count", "truncated", "source"} NOT_RECORDED = "运行状态已经变化，这次质询的问答没有写入运行记录" BUSY_DECIDE = {"detail": "正在回答质询，回答完再批准或驳回", "code": "inquiry_busy"} +#: 封存之后的追问查的是数据源现在的数据：回答末尾的说明 +LIVE_NOTE = "注：以上回答依据的是数据源当前的数据，晚于报告生成的时间；与报告不一致时，可能是报告生成之后数据有更新。" +#: 模型用完了调用轮数、有查询执行失败、有引用解析不了：回答末尾的说明 +ROUNDS_NOTE = "注：调用模型的轮数已用完，以上回答只依据已经查到的结果，可能不完整。" +FAILED_NOTE = "注：有 {n} 次查询执行失败（查询清单只列成功的查询），以上回答可能缺少依据。" +UNRESOLVED_NOTE = "注：回答中有 {n} 处引用无法解析，相关的数没有出处。" async def _clean_settings() -> None: @@ -90,16 +104,19 @@ async def client(): yield c -async def _source(tmp_path, *, options=None) -> tuple[str, int]: +#: 四笔订单：华东两笔（其中一笔 300 是测试账号）、华南一笔 900、华北一笔 400 +SALES = ("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);") + + +async def _source(tmp_path, *, options=None, name: str | None = None, script: str = SALES) -> 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.executescript(script) db.commit() db.close() - name = f"shop_{uuid.uuid4().hex[:8]}" + name = name or 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 {}) @@ -147,10 +164,13 @@ def gated(source: str, *, gate: bool = True) -> dict: @dataclass class Play: - """质询 agent 对某个问题的剧本：依次执行的查询，最后交的回答。""" + """质询 agent 对某个问题的剧本：先看几次表结构（schema），再依次执行的查询，最后交的回答。final 给了就照它交 + （模型不按结构交的样子），不给就交 {rationale: answer}。""" queries: list[str] = field(default_factory=list) answer: str = "查不到判断所需的数据。" + schema: int = 0 + final: dict | None = None PLAYS = {"排除测试账号": Play([NO_TEST], NO_TEST_ANSWER)} @@ -180,10 +200,16 @@ class Models: play = which(task) if model.response_format: self.finals.append(list(messages)) - return AIMessage(content=json.dumps({"rationale": play.answer}, ensure_ascii=False)) + final = play.final if play.final is not None else {"rationale": play.answer} + return AIMessage(content=json.dumps(final, 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)) + answered = [str(m.tool_call_id) for m in messages if isinstance(m, ToolMessage)] + looked = sum(1 for i in answered if i.startswith("is-")) + if looked < play.schema: + call = {"name": f"db_schema__{source}", "args": {}, "id": f"is-{looked}"} + return AIMessage(content="先看表结构。", tool_calls=[call]) + done = sum(1 for i in answered if i.startswith("iq-")) 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]) @@ -300,6 +326,10 @@ async def test_asking_while_waiting_is_recorded_and_sealed_with_the_run(monkeypa assert answer["text"] == NO_TEST_TEXT, "回答先给结论，数字按查询结果换好" answer_doc = artifact_store.load(answer["doc_artifact"]) assert answer_doc["markdown"] == answer["text"] and answer_doc["violations"] == [] + # 运行的工件列表里认得出是哪个问题的回答 + async with SessionLocal() as session: + stored = await session.get(Artifact, (answer["doc_artifact"], paused.id)) + assert stored.kind == "inquiry_answer" and stored.meta == {"report": "write", "unit": east, "question": QUESTION} 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"] @@ -373,6 +403,12 @@ async def test_an_approval_that_lands_while_answering_leaves_no_record(monkeypat 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 + # 没写进记录，提问人手里的回答照样点得开：回答文档和查询快照都在质询工件接口上 + [query] = body["inquiry"]["queries"] + for artifact in (body["inquiry"]["answer"]["doc_artifact"], query["artifact"]): + got = await client.get(f"/api/runs/{paused.id}/inquiries/artifacts/{artifact}") + assert got.status_code == 200, got.text + assert got.json()["content"]["sql"] == NO_TEST final = await wait(paused.id) assert final.status == "succeeded", final.error @@ -431,7 +467,7 @@ async def test_while_answering_approving_resuming_and_continuing_are_refused(mon async def test_after_the_seal_only_exploratory_runs_take_inquiries(monkeypatch, shop, client): - Models(monkeypatch, shop) + models = Models(monkeypatch, shop) done = await finish(shop) assert done.status == "succeeded", done.error east = unit_of(await report_doc(done.id), EAST) @@ -441,7 +477,12 @@ async def test_after_the_seal_only_exploratory_runs_take_inquiries(monkeypatch, 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 + assert item["post_seal"] is True and item["approval_id"] is None + assert item["answer"]["text"] == f"{NO_TEST_TEXT}\n\n{LIVE_NOTE}", "封存之后查的是现在的数据，回答里交代清楚" + [task] = models.tasks + ended = done.finished_at.astimezone(timezone.utc).replace(tzinfo=None) if done.finished_at.tzinfo \ + else done.finished_at + assert f"报告生成于 {ended.isoformat(timespec='seconds')}（UTC）" in task [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) @@ -494,7 +535,8 @@ async def test_a_follow_up_is_not_written_once_the_run_is_continued(monkeypatch, assert r.status_code == 200, r.text body = r.json() assert body["recorded"] is False and body["note"] == NOT_RECORDED - assert body["inquiry"]["post_seal"] is True and body["inquiry"]["answer"]["text"] == NO_TEST_TEXT + assert body["inquiry"]["post_seal"] is True + assert body["inquiry"]["answer"]["text"] == f"{NO_TEST_TEXT}\n\n{LIVE_NOTE}" final = await wait(failed.id) assert final.status == "succeeded", final.error assert await events(failed.id, "approval.inquiry") == [] @@ -620,7 +662,8 @@ async def test_bare_numbers_are_flagged_and_masked_values_never_show(monkeypatch assert r.status_code == 200, r.text assert r.json()["inquiry"]["unit"] is None, "不指明句子时写空串和写 null 一样" [task] = models.tasks - assert f"设了遮罩的列（回答里不要写出这些列的值）：「{masked_shop}」的 buyer_phone" in task + assert f"设了遮罩的列（查询里不要选出原值、不要起别名，回答里不要写出这些列的值）：「{masked_shop}」的 buyer_phone" \ + in task assert "1390000000" not in r.text, "回答、查询清单里都不能出现遮罩列的值" item = r.json()["inquiry"] answer = item["answer"] @@ -674,6 +717,11 @@ async def test_the_answer_may_cite_the_reports_own_metrics(monkeypatch, client): [task] = models.tasks assert "[[m:gmv]]" in task and "销售额 = 45,678.5元" in task assert r.json()["inquiry"]["queries"] == [] + # 回答里的指标点得开：口径卡从质询工件接口取（界面按 doc.catalog[alias].artifact 取） + card = artifact_store.load(answer["doc_artifact"])["catalog"]["m:gmv"]["artifact"] + got = await client.get(f"/api/runs/{paused.id}/inquiries/artifacts/{card}") + assert got.status_code == 200, got.text + assert got.json()["content"]["metrics"][0]["id"] == "gmv" and got.json()["masked"] == [] async def test_a_cell_in_a_table_comes_with_the_whole_table(monkeypatch, shop, client): @@ -691,20 +739,6 @@ async def test_a_cell_in_a_table_comes_with_the_whole_table(monkeypatch, shop, c 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" - - # -------------------------------------------------------------------------- # 触顶与失败 # -------------------------------------------------------------------------- @@ -776,7 +810,8 @@ async def test_the_daily_cap_is_recorded_as_a_limit(monkeypatch, shop, client): 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 (answer["ok"], answer["limit"]) == (False, "daily_max_usd") + assert answer["error"] == "今日核查花费已达上限（核查每日金额上限 $2.00），未能给出回答，可在设置的「核查」一节调高上限" assert models.finals == [] [event] = await events(paused.id, "approval.inquiry") assert event.data["limit"] == "daily_max_usd" @@ -845,3 +880,372 @@ async def test_resuming_after_an_inquiry_does_not_repeat_the_nodes(monkeypatch, 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 + + +# -------------------------------------------------------------------------- +# 遮罩：改了名字的遮罩列、只出现在报告里的遮罩值、不误伤别的数 +# -------------------------------------------------------------------------- + +ALIASED = "SELECT buyer_phone AS 手机号, amount AS 金额 FROM sales ORDER BY amount DESC LIMIT 1" + + +async def test_a_masked_column_under_another_name_is_masked_everywhere(monkeypatch, masked_shop, client): + """遮罩列起了中文别名（本仓库查询的惯例）：回答里引用到的格子、事件、回答文档、质询工件里都不露出原值。 + 报告自己的查询没选手机号，原值哪里都认不出来，只能按 SQL 认出「手机号」就是 buyer_phone。""" + plays = {"买得最多": Play([ALIASED], "这一笔来自 [[v:Q1.r0.手机号]]，金额 [[v:Q1.r0.金额]]。[[see:Q1]]")} + models = 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"] + assert item["answer"]["text"] == "这一笔来自 已遮罩，金额 1,200。" + [query] = item["queries"] + assert query["sql"] == ALIASED, "SQL 里没有遮罩值，原样" + [event] = await events(paused.id, "approval.inquiry") + assert "1390000000" not in json.dumps(event.data, ensure_ascii=False) + assert "1390000000" not in json.dumps(artifact_store.load(item["answer"]["doc_artifact"]), ensure_ascii=False) + got = await client.get(f"/api/runs/{paused.id}/inquiries/artifacts/{query['artifact']}") + assert got.status_code == 200 and "1390000000" not in got.text + assert got.json()["content"]["rows"] == [["已遮罩", 1200]] and got.json()["masked"] == ["手机号"] + [task] = models.tasks + assert "不要起别名" in task + + +async def test_a_masked_value_seen_only_in_the_report_is_masked_alike_everywhere(monkeypatch, masked_shop, client): + """遮罩值只出现在报告自己的查询里（agent 照着它写了查询条件、在回答里复述）：回答、事件里的 SQL、质询工件里的 + SQL 都换成「已遮罩」，两处给的 SQL 一字不差。""" + graph = chain(node("start", "input"), + node("fetch", "tool", tool=f"db_query__{masked_shop}", + args={"sql": "SELECT buyer_phone, amount AS 金额 FROM sales ORDER BY amount DESC"}), + node("write", "report", "写周报", instructions="写周报"), + node("gate", "human", mode="approve", title="周报可以出具吗"), + node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}])) + asked = "SELECT COUNT(*) AS 笔数 FROM sales WHERE buyer_phone = '13900000003'" + plays = {"这位客户": Play([asked], "号码 13900000003 的客户共 [[v:Q1.r0.笔数]] 笔。[[see:Q1]]")} + Models(monkeypatch, masked_shop, writer="金额最高的一笔 [[v:Q1.r0.金额]]。[[see:Q1]]", plays=plays) + paused = await pause(masked_shop, graph=graph) + [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"] + assert item["answer"]["text"] == "号码 已遮罩 的客户共 1 笔。" + [query] = item["queries"] + assert query["sql"] == "SELECT COUNT(*) AS 笔数 FROM sales WHERE buyer_phone = '已遮罩'" + got = await client.get(f"/api/runs/{paused.id}/inquiries/artifacts/{query['artifact']}") + assert got.status_code == 200 and "1390000000" not in got.text + assert got.json()["content"]["sql"] == query["sql"], "质询工件给的 SQL 和事件里的一字不差" + + +#: 年龄设了遮罩的订单：年龄都是两位数，和阈值、日期、LIMIT 里的数撞在一起 +AGED = ("CREATE TABLE sales (id INTEGER PRIMARY KEY, region TEXT, age INTEGER, amount INTEGER, paid_on TEXT);" + "INSERT INTO sales VALUES (1, '华东', 25, 1200, '2026-01-25'), (2, '华东', 30, 300, '2026-01-20')," + " (3, '华南', 20, 900, '2026-02-01'), (4, '华北', 31, 400, '2026-02-03');") + + +async def test_numbers_that_merely_match_a_short_masked_value_are_left_alone(monkeypatch, tmp_path, client): + """遮罩列是年龄：阈值（300）、天数（30）、日期、LIMIT 里的数撞上年龄也不换，封存的 SQL 就是实际执行的那条； + 直接和年龄比较的值照样遮。提问人自己写的阈值复述在回答里，不算没有出处。""" + name, source_id = await _source(tmp_path, options={"mask_columns": "age"}, script=AGED) + graph = chain(node("start", "input"), + node("fetch", "tool", tool=f"db_query__{name}", + args={"sql": "SELECT region AS 区域, age, amount AS 金额 FROM sales ORDER BY id"}), + node("write", "report", "写周报", instructions="写周报"), + node("gate", "human", mode="approve", title="周报可以出具吗"), + node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}])) + recent = "SELECT COUNT(*) AS 笔数 FROM sales WHERE amount > 300 AND paid_on >= '2026-01-25' LIMIT 20" + same_age = "SELECT COUNT(*) AS 人数 FROM sales WHERE age = 25" + plays = {"不超过 300": Play([recent, same_age], + "排除金额不超过 300 元的订单后，近 30 天还剩 [[v:Q1.r0.笔数]] 笔，其中 [[v:Q2.r0.人数]] 笔" + "来自同一年龄段。[[see:Q1]]")} + Models(monkeypatch, name, writer="华东首单金额 [[v:Q1.r0.金额]]。[[see:Q1]]", plays=plays) + try: + paused = await pause(name, graph=graph) + [approval] = await pending(paused.id) + r = await ask(client, paused.id, report="write", unit=None, question="排除金额不超过 300 元的订单会怎样？", + approval_id=approval.id) + assert r.status_code == 200, r.text + item = r.json()["inquiry"] + assert item["answer"]["text"] == "排除金额不超过 300 元的订单后，近 30 天还剩 3 笔，其中 1 笔来自同一年龄段。" + doc = artifact_store.load(item["answer"]["doc_artifact"]) + assert [s["text"] for s in iter_segments(doc) if s.get("issue") == "uncited_number"] == ["30"], \ + "问题里的 300 是提问人给的，不标；30 没有出处，照标" + assert [q["sql"] for q in item["queries"]] == [recent, "SELECT COUNT(*) AS 人数 FROM sales WHERE age = 已遮罩"] + finally: + await engines.invalidate(source_id) + + +# -------------------------------------------------------------------------- +# 接线：两道审批的竞态、上传表格的版本、设置、开放之后状态又变了、张冠李戴的报告文档 +# -------------------------------------------------------------------------- + + +async def test_an_answer_about_the_first_of_two_approvals_is_not_written_into_the_second(monkeypatch, shop, client): + """问的是第一张审批卡，回答途中它被批掉、运行停到了第二张卡上：仍是「停在审批」，但问答不属于这一次停下，不写。""" + seen: dict = {} + + async def approve_and_wait_for_the_next(): + if seen.get("done"): + return + seen["done"] = True + await run_manager.resume(seen["run"], {"approved": True}, approval_id=seen["approval"]) + for _ in range(600): + cards = await pending(seen["run"]) + if cards and cards[0].id != seen["approval"] and (await _row(seen["run"])).status == "interrupted": + seen["next"] = cards[0].id + return + await asyncio.sleep(0.03) + raise AssertionError("运行没有停到第二道审批上") + + Models(monkeypatch, shop, before=approve_and_wait_for_the_next) + graph = chain(node("start", "input"), node("fetch", "tool", tool=f"db_query__{shop}", args={"sql": QUERY}), + node("write", "report", "写周报", instructions="写周报"), + node("gate", "human", mode="approve", title="周报可以出具吗"), + node("gate2", "human", mode="approve", title="再确认一次"), + node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}])) + paused = await pause(shop, graph=graph) + [first] = await pending(paused.id) + seen.update(run=paused.id, approval=first.id) + r = await ask(client, paused.id, report="write", unit=None, question=QUESTION, approval_id=first.id) + assert r.status_code == 200, r.text + assert r.json()["recorded"] is False and r.json()["note"] == NOT_RECORDED + assert (await _row(paused.id)).status == "interrupted", "运行停在第二道审批上" + assert await events(paused.id, "approval.inquiry") == [] + r = await client.post(f"/api/approvals/{seen['next']}/decide", json={"approved": True}) + assert r.status_code == 200, r.text + final = await wait(paused.id) + assert final.status == "succeeded" and (await verify_manifest(paused.id))["ok"] is True + + +def _book(rows: list[list]) -> bytes: + from openpyxl import Workbook + + wb = Workbook() + ws = wb.active + ws.title = "明细" + for row in rows: + ws.append(row) + buf = io.BytesIO() + wb.save(buf) + return buf.getvalue() + + +async def _publish(name: str, rows: list[list]) -> tuple[str, str]: + """发布一版上传表格（一张「明细」），返回 (源 id, 快照 id)。源不存在就新建。""" + (settings.data_dir / "uploads").mkdir(parents=True, exist_ok=True) + async with SessionLocal() as session: + row = (await session.execute(select(DataSource).where(DataSource.name == name))).scalar_one_or_none() + if row is None: + row = DataSource(id=uuid.uuid4().hex, name=name, kind="sqlite", readonly=True, origin="upload") + result = await table_versions.publish_upload(session, row, _book(rows), "客流.xlsx", header_row=1, + mixed="reject", raw_mode=False) + return row.id, result.snapshot_id + + +async def test_an_uploaded_table_is_queried_at_the_version_the_run_pinned(monkeypatch, client): + """运行停在审批时上传表格重传了一版：质询查的仍是运行固定的那一版，和报告用的是同一份数据。""" + name = f"upload_{uuid.uuid4().hex[:8]}" + _, first = await _publish(name, [["区域", "数量"], ["分区甲", 1], ["分区乙", 2]]) + plays = {"合计": Play(['SELECT SUM("数量") AS 合计 FROM "明细"'], "合计为 [[v:Q1.r0.合计]]。[[see:Q1]]")} + Models(monkeypatch, name, writer="数量合计 [[v:Q1.r0.total]]。[[see:Q1]]", plays=plays) + graph = chain(node("start", "input"), + node("fetch", "tool", tool=f"db_query__{name}", args={"sql": 'SELECT SUM("数量") AS total FROM "明细"'}), + node("write", "report", "写周报", instructions="写周报"), + node("gate", "human", mode="approve", title="周报可以出具吗"), + node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}])) + paused = await pause(name, graph=graph) + await _publish(name, [["区域", "数量"], ["分区甲", 60], ["分区乙", 40]]) + [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 + item = r.json()["inquiry"] + assert item["answer"]["text"] == "合计为 3。", "按运行固定的那一版查，不是重传之后的" + [query] = item["queries"] + assert artifact_store.load(query["artifact"])["data_version"] == first + assert (await _row(paused.id)).data_versions == paused.data_versions + + +async def test_the_settings_and_the_runs_pinned_versions_reach_the_agent(monkeypatch, shop, client): + await set_verify({"inquiry": {"max_queries": 3, "timeout_s": 45}, "daily_max_usd": 3.5}) + seen: dict = {} + original = inquiry_mod.run_verify_agent + + async def spy(**kwargs): + seen.update(kwargs) + return await original(**kwargs) + + monkeypatch.setattr(inquiry_mod, "run_verify_agent", spy) + 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 + assert (seen["max_queries"], seen["timeout_s"], seen["daily_max_usd"]) == (3, 45, 3.5) + assert seen.get("data_versions") == paused.data_versions and seen["sources"] == [shop] + assert seen["run_id"] == paused.id and seen.get("emit") is None + + +async def test_a_run_continued_while_the_inquiry_opens_is_refused(monkeypatch, shop, client): + """读完运行状态之后、核对封存之前，运行被继续了：记录上的封存是上一次结束时的，不开放（也不白问一次模型）。""" + from app.api import evidence + + models = Models(monkeypatch, shop) + done = await finish(shop) + original = evidence._sealed + + async def continued_meanwhile(run_id): + async with SessionLocal() as session: + row = await session.get(Run, run_id) + row.status = "running" + await session.commit() + return await original(run_id) + + monkeypatch.setattr(evidence, "_sealed", continued_meanwhile) + 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 == [] and models.specs == [] + + +async def test_a_report_document_that_belongs_to_another_run_is_refused(monkeypatch, shop, client): + """事件里记的报告文档其实是另一次运行的（内容寻址只保证没被改，不保证没被张冠李戴）：不据此质询。""" + models = Models(monkeypatch, shop) + paused = await pause(shop) + other = await pause(shop) + [approval] = await pending(paused.id) + [checked] = await events(paused.id, "report.checked", "write") + [foreign] = await events(other.id, "report.checked", "write") + async with SessionLocal() as session: + stored = await session.get(RunEvent, checked.id) + stored.data = {**stored.data, "doc_artifact": foreign.data["doc_artifact"]} + await session.commit() + r = await ask(client, paused.id, report="write", unit=None, question=QUESTION, approval_id=approval.id) + assert r.status_code == 422 and r.json()["field"] == "report" + assert models.tasks == [] + + +# -------------------------------------------------------------------------- +# 数据源：同名重建的不查 +# -------------------------------------------------------------------------- + + +async def test_a_source_rebuilt_under_the_same_name_is_not_queried(monkeypatch, tmp_path, client): + """运行用过的数据源被删了、又用同一个名字新建（手工源、上传表格都算）：那已经不是这次运行查过的数据，不交给 agent； + 上传表格也不会因此被补固定进运行记录。""" + name, source_id = await _source(tmp_path) + plays = {"华南为什么": Play([], "查不到判断所需的数据。")} + models = Models(monkeypatch, name, plays=plays) + done = await finish(name) + assert done.status == "succeeded", done.error + + async def drop(sid) -> None: + async with SessionLocal() as session: + await session.delete(await session.get(DataSource, sid)) + await session.commit() + await engines.invalidate(sid) + + await drop(source_id) + _, rebuilt = await _source(tmp_path, name=name) + note = f"注：这次运行还用到了数据源「{name}」，已不存在、已停用或已被重建，未能查询，以上回答不含其中的数据。" + r = await ask(client, done.id, report="write", unit=None, question="华南为什么低？") + assert r.status_code == 200, r.text + assert r.json()["inquiry"]["answer"]["text"] == f"查不到判断所需的数据。\n\n{note}" + [task] = models.tasks + assert f"（这次运行还用到了「{name}」，已不存在、已停用或已被重建，无法查询）" in task + + # 同名新建成上传表格。升级前的查询快照引用没记源 id（认不出是不是同一个源）：这次运行没固定过它的版本，照样不查 + await drop(rebuilt) + await _publish(name, [["区域", "数量"], ["分区甲", 1]]) + async with SessionLocal() as session: + for ref in (await session.execute(select(Artifact).where(Artifact.run_id == done.id))).scalars(): + ref.meta = {} + await session.commit() + r = await ask(client, done.id, report="write", unit=None, question="华南为什么低？") + assert r.status_code == 200 and r.json()["inquiry"]["answer"]["text"].endswith(note) + assert (await _row(done.id)).data_versions == done.data_versions, "上传表格没有被补固定进已封存的运行" + + +# -------------------------------------------------------------------------- +# 说法：触顶、失败、开放不了 +# -------------------------------------------------------------------------- + + +async def test_running_out_of_rounds_is_not_reported_as_the_query_limit(monkeypatch, shop, client): + """模型一直在看表结构、轮数用完了：没查过一次，不能说「查询次数到顶」；回答里的引用一个都解析不了，末尾说明。""" + await set_verify({"inquiry": {"max_queries": 1}}) + plays = {"排除测试账号": Play([], "仍居首：华东为 [[v:Q1.r0.销售额]]。", schema=99)} + 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"], item["queries"]) == (True, None, []) + assert item["answer"]["text"] == (f"仍居首：华东为 ⟦?v:Q1.r0.销售额⟧。\n\n{ROUNDS_NOTE}\n\n" + f"{UNRESOLVED_NOTE.format(n=1)}") + + +async def test_failed_queries_are_explained(monkeypatch, shop, client): + """两次查询都执行失败、次数用完了：查询清单只列成功的，回答末尾说清有几次失败、几处引用解析不了。""" + await set_verify({"inquiry": {"max_queries": 2}}) + plays = {"排除测试账号": Play(["SELECT nope FROM sales", "SELECT nothing FROM sales"], NO_TEST_ANSWER)} + 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"]["limit"], item["queries"]) == ("max_queries", []) + assert item["answer"]["text"] == ("仍居各区域之首：排除测试账号后华东为 ⟦?v:Q1.r0.销售额⟧，华南为 ⟦?v:Q1.r1.销售额⟧。\n\n" + f"{FAILED_NOTE.format(n=2)}\n\n{UNRESOLVED_NOTE.format(n=2)}") + + +async def test_an_answer_not_given_in_the_asked_shape_says_what_to_do(monkeypatch, shop, client): + plays = {"排除测试账号": Play([NO_TEST], final={"answer": "华东仍居首"})} + 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) + answer = r.json()["inquiry"]["answer"] + assert answer["ok"] is False + assert answer["error"] == "模型未能按要求给出回答，请重新提问；如反复出现，请在设置的「核查」一节更换核查模型" + + +def test_internal_wording_from_the_agent_is_not_shown(): + """核查 agent 的原话（「交结果」「按结构」、裸写的字段名）换成提问人看得懂的话，并说下一步。""" + def words(error, limit=None, result=None): + outcome = verify_agent.VerifyOutcome(result=result, cells=[], queries=[], rationale=None, usage={}, + limit=limit, error=error) + return inquiry_mod._error_words(outcome, timeout_s=120, daily_max_usd=2.0) + + assert words("收尾时超过时长上限，没能交结果", limit="timeout_s") == \ + "回答超过时长上限（120 秒），未能给出回答，可缩小问题范围后重新提问" + assert words("已达全局每日核查金额上限 $2.00，没能交结果", limit="daily_max_usd") == \ + "今日核查花费已达上限（核查每日金额上限 $2.00），未能给出回答，可在设置的「核查」一节调高上限" + assert words("核查模型没有按结构交结果（交回的内容里没有要求的字段：rationale）") == \ + "模型未能按要求给出回答，请重新提问；如反复出现，请在设置的「核查」一节更换核查模型" + assert words("核查模型没有给出回复") == "模型没有给出回复，未能给出回答，请重新提问" + assert words("核查出错（磁盘已满）") == "回答时出错（磁盘已满），请重新提问" + assert words("核查模型不可用（没有配置模型接入）") == "核查模型不可用（没有配置模型接入）" + assert words(None) is None + + +async def test_refusals_while_the_run_is_executing_say_what_to_do(monkeypatch, shop, client): + """运行在执行：带着审批卡来的多半是卡刚被别人批了，叫人刷新；没带卡的按运行类别说什么时候能问。""" + Models(monkeypatch, shop) + async with SessionLocal() as session: + exploring = Run(workflow_name="还在执行", status="running", graph={}, input={}, run_class="exploratory") + formal = Run(workflow_name="还在执行", status="queued", graph={}, input={}, run_class="formal") + session.add_all([exploring, formal]) + await session.commit() + ids = exploring.id, formal.id + r = await ask(client, ids[0], report="write", unit=None, question=QUESTION, approval_id="card-1") + assert r.status_code == 409 + assert r.json() == {"detail": "这张审批卡已处理，运行正在继续执行，请刷新页面查看最新状态", "code": "inquiry_not_open"} + r = await ask(client, ids[0], report="write", unit=None, question=QUESTION) + assert r.json()["detail"] == "运行仍在执行。停在人工审批时，或运行结束并封存后，才能质询" + r = await ask(client, ids[1], report="write", unit=None, question=QUESTION) + assert r.json()["detail"] == "运行仍在执行，停在人工审批时才能质询" diff --git a/backend/tests/test_inquiry_masking.py b/backend/tests/test_inquiry_masking.py new file mode 100644 index 00000000..146dddad --- /dev/null +++ b/backend/tests/test_inquiry_masking.py @@ -0,0 +1,204 @@ +"""审批质询的遮罩（engine/inquiry.py）：回答正文、查询清单、质询工件里都不露出数据源遮罩列的值，也不误伤别的数。 + +- 改了名字的遮罩列（起别名、套函数、经子查询或 CTE 转一道、UNION 的另一支）照样遮；COUNT 只数个数、窗口的分区键 + 只决定次序，不遮；解析不了、SQL 里又写着遮罩列的，整份快照遮掉 +- 按值认的遮罩值有门槛：纯数字够长（SECRET_DIGITS 位）、文字至少两个字；日期、小数、千分位里的一截不算；提问人自己 + 写进问题、选中原文里的不换 +- SQL 只换字面量（和注释）里的遮罩值：直接和遮罩列比较的字面量是遮罩列出现过的值就换，短的数也换；LIMIT、日期、 + 别的列的条件照原样 +- 快照按列遮，另按值遮：别的列里写着遮罩值的格子也换成「已遮罩」 +- 组装回答两遍之后正文里还认得出遮罩值的，不交出去 +- 遮罩值上万个时换一遍也不卡：不现编上万个分支的正则 + +数据都是虚构的：手机号、年龄、区域、金额。 +""" +from __future__ import annotations + +import time + +import pytest + +from app.engine import inquiry as inquiry_mod + +MASKED = "已遮罩" + + +def secrets_of(snapshot: dict, hidden: set[str]) -> inquiry_mod._Secrets: + secrets = inquiry_mod._Secrets() + secrets.add(snapshot, hidden) + return secrets + + +# -------------------------------------------------------------------------- +# 改了名字的遮罩列 +# -------------------------------------------------------------------------- + + +def hidden_of(sql: str, columns: list[str], masked: set[str] = frozenset({"buyer_phone"})) -> set[str]: + snapshot = {"source": "shop", "sql": sql, "columns": columns, "rows": []} + return inquiry_mod._derived_hidden(snapshot, set(masked), "sqlite")[0] + + +@pytest.mark.parametrize("sql, columns, hidden", [ + ("SELECT buyer_phone AS 手机号, amount AS 金额 FROM sales", ["手机号", "金额"], {"buyer_phone", "手机号"}), + ("SELECT UPPER(buyer_phone), amount FROM sales", ["UPPER(buyer_phone)", "amount"], + {"buyer_phone", "upper(buyer_phone)"}), + ("SELECT t.p AS 号码 FROM (SELECT buyer_phone AS p FROM sales) t", ["号码"], {"buyer_phone", "p", "号码"}), + ("WITH t AS (SELECT buyer_phone AS p, amount FROM sales) SELECT p, amount FROM t", ["p", "amount"], + {"buyer_phone", "p"}), + ("WITH t(p) AS (SELECT buyer_phone FROM sales) SELECT p AS 号码 FROM t", ["号码"], {"buyer_phone", "p", "号码"}), + ("SELECT region AS 区域 FROM sales UNION ALL SELECT buyer_phone FROM sales", ["区域"], {"buyer_phone", "区域"}), + ("SELECT (SELECT buyer_phone FROM sales LIMIT 1) AS 某个号码", ["某个号码"], {"buyer_phone", "某个号码"}), +]) +def test_a_masked_column_under_another_name_is_masked(sql, columns, hidden): + assert hidden_of(sql, columns) == hidden + + +@pytest.mark.parametrize("sql, columns", [ + ("SELECT region AS 区域, COUNT(DISTINCT buyer_phone) AS 买家数 FROM sales GROUP BY region", ["区域", "买家数"]), + ("SELECT region AS 区域, SUM(amount) AS 销售额 FROM sales WHERE buyer_phone = '13900000001' GROUP BY region", + ["区域", "销售额"]), + ("SELECT ROW_NUMBER() OVER (PARTITION BY buyer_phone ORDER BY amount) AS 序号 FROM sales", ["序号"]), + ("SELECT (SELECT COUNT(*) FROM sales s2 WHERE s2.buyer_phone = s.buyer_phone) AS 笔数 FROM sales s", ["笔数"]), +]) +def test_counting_filtering_and_ordering_by_a_masked_column_shows_nothing(sql, columns): + assert hidden_of(sql, columns) == {"buyer_phone"} + + +def test_sql_that_cannot_be_read_but_names_a_masked_column_hides_everything(): + assert hidden_of("SELECT buyer_phone AS 号码 FROM sales WHERE (", ["号码", "金额"]) == {"buyer_phone", "号码", "金额"} + assert hidden_of("SELECT region FROM sales WHERE (", ["region"]) == {"buyer_phone"} + + +def test_names_that_point_at_a_masked_column_are_known_for_comparisons(): + snapshot = {"source": "shop", "sql": "SELECT p FROM (SELECT buyer_phone AS p FROM sales) t", "columns": ["p"], + "rows": []} + assert inquiry_mod._derived_hidden(snapshot, {"buyer_phone"}, "sqlite")[1] == {"buyer_phone", "p"} + + +# -------------------------------------------------------------------------- +# 按值认：门槛、边界 +# -------------------------------------------------------------------------- + + +def test_masked_values_are_matched_as_whole_values(): + """遮罩列的值按整值认：更长的数里的一截、千分位写法里的一截不换；一个字的值、太短的数不换（换了会把不相干的 + 词、阈值弄坏）；数按原样和千分位两种写法认。""" + snapshot = {"columns": ["phone", "name", "gender", "code", "salary", "region"], + "rows": [["13900000001", "张伟", "男", "390", 12000, "华东"]]} + scrub = secrets_of(snapshot, {"phone", "name", "gender", "code", "salary"}).scrub + assert scrub("客户 13900000001、139000000012、1,390 元、390 笔、张伟买了男装，月薪 12,000，另 12000.5") == ( + "客户 已遮罩、139000000012、1,390 元、390 笔、已遮罩买了男装，月薪 已遮罩，另 12000.5") + assert scrub("华东") == "华东", "没遮的列不换" + assert inquiry_mod._Secrets().scrub("13900000001") == "13900000001" + + +def test_short_masked_numbers_leave_thresholds_dates_and_limits_alone(): + """遮罩列是年龄这类短的数：正文、SQL 里的阈值、日期、LIMIT 撞上同一个数也不换。""" + snapshot = {"columns": ["age", "vip_level"], "rows": [[25, 10], [31, 10], [47, 10]]} + scrub = secrets_of(snapshot, {"age", "vip_level"}).scrub + text = "近 30 天里，2026-01-25 之后的 25 笔订单比 1 月 10 日多 47 笔" + assert scrub(text) == text + + +def test_long_values_next_to_dates_and_decimals_are_not_cut_out_of_them(): + secrets = secrets_of({"columns": ["phone"], "rows": [["13900000001"]]}, {"phone"}) + assert secrets.scrub("13900000001-2、13900000001.5、20261390000000-13900000001") == ( + "13900000001-2、13900000001.5、20261390000000-13900000001") + assert secrets.scrub("号码：13900000001。") == "号码：已遮罩。" + + +def test_values_the_asker_wrote_themselves_are_left_alone(): + """提问人自己写进问题、选中原文里的数（「13900000001 这位客户」）照原样：问题本身就带着它，换了只会让回答对不上。""" + secrets = secrets_of({"columns": ["phone"], "rows": [["13900000001"], ["13900000002"]]}, {"phone"}) + secrets.protect(["13900000001 这位客户买了几笔？", ""]) + assert secrets.scrub("13900000001 共 2 笔，13900000002 共 1 笔") == "13900000001 共 2 笔，已遮罩 共 1 笔" + + +def test_a_cell_holding_a_masked_value_is_masked_in_any_column(): + """别的列里写着同一个手机号（另一张表的联系方式没设遮罩）：按值换。短的数不按值换，免得把同样是 25 的笔数遮了。""" + secrets = secrets_of({"columns": ["phone", "age"], "rows": [["13900000001", 25]]}, {"phone", "age"}) + snapshot = {"columns": ["联系方式", "笔数", "金额"], "column_types": {"联系方式": "text", "笔数": "number"}, + "rows": [["13900000001", 25, 13900000001], ["13900000009", 3, 1200]]} + shown, masked = inquiry_mod._masked_view(snapshot, set(), secrets) + assert shown["rows"] == [[MASKED, 25, MASKED], ["13900000009", 3, 1200]] + assert masked == ["联系方式", "金额"] + assert snapshot["rows"][0][0] == "13900000001", "原快照不动" + by_name, masked = inquiry_mod._masked_view(snapshot, {"笔数"}, inquiry_mod._Secrets()) + assert [r[1] for r in by_name["rows"]] == [MASKED, MASKED] and masked == ["笔数"] + assert by_name["column_types"]["笔数"] == "text" + + +# -------------------------------------------------------------------------- +# SQL：只换字面量 +# -------------------------------------------------------------------------- + + +@pytest.fixture +def people() -> inquiry_mod._Secrets: + return secrets_of({"columns": ["buyer_phone", "age"], "rows": [["13900000001", 25], ["13900000003", 31]]}, + {"buyer_phone", "age"}) + + +@pytest.mark.parametrize("sql, shown", [ + ("SELECT COUNT(*) AS 笔数 FROM sales WHERE buyer_phone = '13900000001'", + "SELECT COUNT(*) AS 笔数 FROM sales WHERE buyer_phone = '已遮罩'"), + ("SELECT COUNT(*) FROM sales s WHERE s.buyer_phone IN ('13900000001', '13900000003') AND age BETWEEN 25 AND 31", + "SELECT COUNT(*) FROM sales s WHERE s.buyer_phone IN ('已遮罩', '已遮罩') AND age BETWEEN 已遮罩 AND 已遮罩"), + ("SELECT COUNT(*) FROM sales WHERE age = 25 AND amount > 25 AND paid_on >= '2026-01-25' LIMIT 25", + "SELECT COUNT(*) FROM sales WHERE age = 已遮罩 AND amount > 25 AND paid_on >= '2026-01-25' LIMIT 25"), + ("SELECT COUNT(*) FROM sales WHERE 25 = age AND note LIKE '%13900000003%' -- 13900000001 的单\nLIMIT 20", + "SELECT COUNT(*) FROM sales WHERE 已遮罩 = age AND note LIKE '%已遮罩%' -- 已遮罩 的单\nLIMIT 20"), + ("SELECT region FROM sales WHERE buyer_phone NOT IN ('13900000009') AND age <> 40", + "SELECT region FROM sales WHERE buyer_phone NOT IN ('13900000009') AND age <> 40"), + ("SELECT amount FROM sales WHERE amount = 13900000001", "SELECT amount FROM sales WHERE amount = 已遮罩"), +]) +def test_sql_keeps_everything_but_the_masked_values(people, sql, shown): + assert inquiry_mod._scrub_sql(sql, people, {"buyer_phone", "age"}, "sqlite") == shown + + +def test_a_comparison_with_a_renamed_masked_column_is_masked_too(people): + sql = "SELECT * FROM (SELECT buyer_phone AS p FROM sales) t WHERE p = '13900000003' AND 1 = 1" + assert inquiry_mod._scrub_sql(sql, people, {"buyer_phone", "p"}, "sqlite") == ( + "SELECT * FROM (SELECT buyer_phone AS p FROM sales) t WHERE p = '已遮罩' AND 1 = 1") + + +def test_sql_that_cannot_be_split_into_words_is_scrubbed_like_text(people): + assert inquiry_mod._scrub_sql("SELECT 'x FROM t WHERE p = 13900000001 LIMIT 25", people, {"buyer_phone"}, + "sqlite") == "SELECT 'x FROM t WHERE p = 已遮罩 LIMIT 25" + + +# -------------------------------------------------------------------------- +# 组装回答的最后一道关 +# -------------------------------------------------------------------------- + + +def test_an_answer_still_showing_a_masked_value_is_not_handed_out(): + """运行输入里恰好是一个遮罩值（报告目录里的 i: 条目）：标记里看不出来，渲染出来才露，连标记一起换也换不掉—— + 宁可不交这份回答。""" + secrets = secrets_of({"columns": ["phone"], "rows": [["13900000001"]]}, {"phone"}) + catalog = {"i:客户": {"alias": "i:客户", "kind": "input", "eid": "input:客户", "locator": {"field": "客户"}, + "value": "13900000001", "label": "客户 = 13900000001"}} + with pytest.raises(inquiry_mod._Leak): + inquiry_mod._compose("这位客户是 [[i:客户]]。", catalog, run_id="r1", views={}, secrets=secrets, allow=[]) + doc = inquiry_mod._compose("没有遮罩值。", catalog, run_id="r1", views={}, secrets=secrets, allow=[]) + assert doc["markdown"] == "没有遮罩值。" + + +# -------------------------------------------------------------------------- +# 规模 +# -------------------------------------------------------------------------- + + +def test_twenty_thousand_masked_values_are_scrubbed_without_stalling(): + """上万个遮罩值（几份上千行的快照）：以前现编一个上万分支的正则，编一次一秒上下，每次回答、每次取工件都卡住整个 + 进程。现在先按词、按开头两个字筛出正文里真有的几个，再换。""" + rows = [[f"139{i:08d}", f"客户{i:05d}"] for i in range(10_000)] + secrets = secrets_of({"columns": ["phone", "name"], "rows": rows}, {"phone", "name"}) + text = "近 30 天里客户00042 下了 3 笔，号码 13900000042，" * 60 + started = time.perf_counter() + for _ in range(5): + out = secrets.scrub(text) + elapsed = (time.perf_counter() - started) / 5 + assert "13900000042" not in out and "客户00042" not in out and "近 30 天" in out + assert elapsed < 0.5, f"换一遍用了 {elapsed:.2f} 秒"