diff --git a/tools/soak/scenarios/govdoc.py b/tools/soak/scenarios/govdoc.py new file mode 100644 index 0000000..fea5ed1 --- /dev/null +++ b/tools/soak/scenarios/govdoc.py @@ -0,0 +1,1268 @@ +"""GovDoc 公文审核场景：把真实政府采购文书接成 PolyLoop 认得的四样东西。 + +交出去的是一个决策解释器（`GovDocParser`）、一组工具与它们派生的动作执行器 +（`GovDocTools`）、一份上下文（`build_context`）和一份装好的运行请求（`build_run_request`）。 + +**一个审核任务被拆成三次独立的运行**（plan → execute → summarize）。PolyLoop 的治理单位是 +一次运行，而「按阶段收窄工具集」在它这里只能表达成「每个阶段一次 `run`，每次传不同的 +`ToolRegistry`」。三次运行共享同一个工作区目录，阶段之间靠工作区里的文件传状态。这照的是 +`reference/GovDoc-Editor/agents/gov-auditor.yaml` 的三阶段与工具收窄，以及 +`reference/GovDoc-Editor/govdoc/pipelines/audit_tender.py` 的「每个审核点一个独立 workspace +加一次独立 run」。 + +**GovDoc 的代码在这里只当协议文档看，一行都不 import。** 它是 PolyLoop 的下游，反向 import +是硬约束（CLAUDE.md §1.2，由 import-linter 断言）。取值（三个阶段的轮次上限、`verdict` 的 +三个取值）照抄它的配置，实现是独立写的。 + +**语料必须先过脱敏。** 数据源里的四份公文是真实的政府采购文书原文，含工商全称、机构名、 +固定电话、统一社会信用代码、邮箱、联系人姓名。第一节的脱敏器在内存里替换掉这些， +`assert_no_residue` 是硬闸——装配语料时调用它，不通过就拒绝启动压测。原文与脱敏文本都不写盘。 + +**已知缺口：门牌级地址不在替换范围内。** 原文里有「某街道某社区某号」这种能精确定位的地址， +但中文地址没有可靠的结尾标志（`号`、`室`、`栋` 在别处也大量出现），按形态认会把设备规格里的 +编号一起吃掉。机构名与联系方式都换掉之后，剩下的地址指向的是一个已经改了名的主体。要补的话， +补在检测器里、连校验函数一起补——两者共用同一份检测，不许各写一套。 +""" + +from __future__ import annotations + +import hashlib +import json +import re +import sqlite3 +from collections.abc import Mapping +from contextlib import closing +from dataclasses import dataclass +from pathlib import Path + +from polyloop.ports import Action, InvalidDecision, ParsedReply, ToolCall +from polyloop.session import RunRequest +from polyloop.tools import ToolRegistry, ToolSpec +from polyloop.types import ( + Budget, + Context, + Message, + ModelReply, + ReplayPolicy, + Role, + SyntheticObservations, + TextBlock, +) + + +class GovDocScenarioError(RuntimeError): + """这个场景装配不出来：数据源不在、表里没有行、语料文件缺失。 + + 只有一个错误类型，因为调用方对这几种情形的处置完全一样——压测跑不起来，人得去看一眼。 + """ + + +class RedactionResidueError(GovDocScenarioError): + """脱敏之后仍然检出可识别信息。 + + **这是硬闸，不是告警。** 把未脱敏的第三方真实信息（工商全称、信用代码、联系人电话）发给 + 外部模型服务是不可逆的：请求一旦出去就收不回来，而且漏掉一处的表现是「压测正常跑完」， + 事后从任何一份产物里都看不出来发生过泄漏。所以宁可拒绝启动。 + + 异常消息里的样本一律掩码，否则拦截泄漏的那条日志本身就成了泄漏。 + """ + + +# --------------------------------------------------------------------------- +# 一、脱敏 +# --------------------------------------------------------------------------- + +#: 所有假名共用的前缀。**检测器把以它开头的命中一律当成「已经脱敏过的占位符」跳过**， +#: 否则「示例甲有限公司」会被自己的校验函数判成残留，脱敏永远收敛不了。 +#: +#: 代价是原文里真有一个以「示例」开头的机构名时会被漏掉。这个代价是被接受的：正式工商名称与 +#: 机关名称里不会出现这两个字打头，而换一个更古怪的前缀会让脱敏后的文本读起来像乱码，模型的 +#: 行为跟着变，压出来的负载就不再像真实负载。 +PSEUDONYM_PREFIX = "示例" + +#: 假名的编号用天干，用完了退回数字。天干让假名一眼看出是编造的，而 `示例甲有限公司` 与 +#: `示例乙有限公司` 在模型眼里是两个不同主体——这一点必须保住，否则任务难度会变。 +_TAGS = "甲乙丙丁戊己庚辛壬癸" + +#: 机构名的字符集：中文、拉丁字母、数字。**不含标点与空白**，所以一次匹配不会跨过顿号、 +#: 括号、换行——那些正是机构名的天然边界。 +_NAME_CHAR = r"[一-龥A-Za-z0-9]" + +#: 机构名向左最多再吃多少个字符。真实的正式名称最长在二十几个汉字，取 24 是给它留余量； +#: 上限存在本身是必要的，没有它一整段没有标点的中文会被当成一个名字。 +_MAX_ORG_LEFT_CHARS = 24 + +#: 每一类的后缀、名称最短长度、以及假名里用的类别词。 +#: +#: **最短长度是用来滤掉通用名词的**：`公司`、`医院`、`财政局` 这些单独出现时是普通名词不是 +#: 身份，而 `广州市财政局` 是身份。长度阈值挡不住全部通用词，剩下的交给下面的 `_GENERIC_ORGS`。 +_ORG_CATEGORIES: tuple[tuple[str, tuple[str, ...], int, str], ...] = ( + ("company", ("股份有限公司", "有限责任公司", "有限公司", "公司"), 5, "有限公司"), + ("hospital", ("医院", "卫生院"), 5, "医院"), + ("bureau", ("局",), 4, "行政局"), + ("committee", ("委员会", "管委会", "发改委", "卫健委"), 5, "委员会"), + ("center", ("中心",), 4, "中心"), + ("government", ("人民政府",), 6, "人民政府"), +) + +_ORG_SUFFIX_TO_CATEGORY: Mapping[str, str] = { + suffix: category for category, suffixes, _, _ in _ORG_CATEGORIES for suffix in suffixes +} +_ORG_MIN_LEN: Mapping[str, int] = {category: min_len for category, _, min_len, _ in _ORG_CATEGORIES} +_ORG_CATEGORY_WORD: Mapping[str, str] = {category: word for category, _, _, word in _ORG_CATEGORIES} + +#: 后缀按长度倒序排进正则，长的先试。 +_ORG_RE = re.compile( + _NAME_CHAR + + r"{0," + + str(_MAX_ORG_LEFT_CHARS) + + r"}(?:" + + "|".join(sorted(_ORG_SUFFIX_TO_CATEGORY, key=len, reverse=True)) + + r")" +) + +#: 机构名不会跨过这些字。命中之后从**最后一个**这样的字之后重新起头，把「须提供」「本项目 +#: 是指」「否则」这类粘在名字前面的句子成分切掉。 +#: +#: **这一份是保守的**：只收语法性的虚词和高频动词，不收任何可能出现在名字里的字。`中`、 +#: `国`、`省`、`市`、`区`、`从`（从化区）刻意不在里面——把它们收进来会把真名字腰斩成两段， +#: 而腰斩之后剩下的那半可能短到过不了长度阈值，于是整个名字漏检，那是泄漏。切不干净只是把 +#: 多余的字一起替换掉，读起来别扭而已。 +_ORG_BOUNDARY_CHARS = frozenset( + "的和或与及由在是指须并且对经按以被向为其该此将不否则还已果可据应即如若等再又也都只先凡受含给让把于而但因之者们到使请提何" +) + +#: 通用机构名词：它们是普通名词、机构类别、或者全国性主管部门，不指向某个具体的被审计主体， +#: 替换掉只会让审核任务读不懂。`评标委员会`、`总公司`、`分公司` 在招标文书里逐段出现； +#: `内镜中心`、`控制中心` 是医疗设备规格里的词；`市场监管总局`、`监狱管理局` 出现在资格条件 +#: 与加分条款里，而那些条款恰恰是审核点要判的东西——把它们换成假名，任务就没得判了。 +#: +#: **这份清单只减不加地影响安全性**：漏收一个通用词只是多替换一处（读起来别扭），错收一个 +#: 真实主体则是漏检（泄漏）。所以往里加名字之前要先确认它不指向任何具体的被审计对象。 +_GENERIC_ORGS = frozenset( + { + "评标委员会", + "采购委员会", + "仲裁委员会", + "评审委员会", + "改革委员会", + "国家发改委", + "管委会", + "市场监管总局", + "监狱管理局", + "戒毒管理局", + "密码管理局", + "工商行政管理局", + "金融监督管理局", + "授时中心", + "集团公司", + "总公司", + "分公司", + "子公司", + "母公司", + "本公司", + "我公司", + "该公司", + "贵公司", + "他公司", + "支公司", + "中心支公司", + "甲公司", + "乙公司", + "丙公司", + "保险公司", + "人民政府", + "内镜中心", + "控制中心", + "数据中心", + "自动控制中心", + "布局", + "结局", + "格局", + "全局", + "大局", + "当局", + "开局", + "僵局", + "败局", + "棋局", + "邮局", + "平局", + "战局", + "时局", + "变局", + "残局", + "终局", + "危局", + "定局", + } +) + +#: 行政区划的收尾字。一个候选名以通用词结尾时，只有它前面那半**不是**区划名，才判成通用词。 +#: 这样 `本级人民政府` 是通用词，`广州市人民政府` 不是。 +_PLACE_SUFFIX_CHARS = frozenset("省市区县州镇乡旗盟岛港") + +_EMAIL_RE = re.compile(r"[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}") + +#: 统一社会信用代码：18 位，用的是去掉 I O S V Z 的字符集。**要求至少含一个字母**， +#: 纯数字的 18 位串交给身份证那条规则去认，认不出就不动它——那多半是银行账号或流水号， +#: 替换掉会把审核依据改坏。 +_USCC_RE = re.compile(r"(?]{1,6})") + +#: `联系人：` 后面跟的这些词是表格里的下一个字段名，不是人名。OCR 把表格拍平之后很常见。 +_CONTACT_FIELD_LABELS = frozenset( + {"电话", "传真", "邮箱", "地址", "手机", "邮编", "单位", "姓名", "方式"} +) + +#: 假名里用的类别词，非机构那几类。 +_PLAIN_CATEGORY_WORD: Mapping[str, str] = { + "email": "邮箱", + "phone": "电话", + "mobile": "手机", + "uscc": "统一社会信用代码", + "id_card": "身份证号", + "person": "联系人", +} + +#: 脱敏最多迭代几轮。一轮替换之后可能长出新的候选——`广州市和示例甲医院` 这种，是切边界时 +#: 把真名字的前半截留在了原地，它与刚放进去的假名连成了一个新的候选串。再跑一轮就把整串一起 +#: 换掉。收敛是保证的：假名以 `示例` 开头，检测器跳过它们。 +_MAX_REDACTION_PASSES = 5 + + +@dataclass(frozen=True, slots=True, kw_only=True) +class RedactionResult: + """一次脱敏的结果与它的账。 + + `counts` 是**替换发生的次数**（同一个名字出现十次算十次），`distinct` 是**不同原名的 + 个数**。两个数都要，因为它们答的是不同的问题：前者答「文本被改动了多少处」，后者答 + 「文档里有多少个可识别主体」。 + """ + + text: str + counts: Mapping[str, int] + distinct: Mapping[str, int] + + +@dataclass(frozen=True, slots=True) +class Identifier: + """文本里一段还能识别到具体主体的片段：半开区间加它属于哪一类。 + + 带上区间而不只是那段文本，是因为替换要按位置一次扫完。按文本反复 `str.replace` 的话， + 前一次替换的结果会被后一次再匹配一遍，而且短名字会把长名字的一部分先换掉。 + """ + + start: int + end: int + category: str + + +def _mask(value: str) -> str: + """把一个命中掩码成「首字 + 星号 + 尾字」。异常消息里只出现掩码后的形态。""" + if len(value) <= 2: + return value[0] + "*" + return value[0] + "*" * (len(value) - 2) + value[-1] + + +def _is_generic_org(name: str) -> bool: + """这个候选是不是通用机构名词，不是某个具体主体。""" + if name in _GENERIC_ORGS: + return True + for generic in _GENERIC_ORGS: + if not name.endswith(generic) or name == generic: + continue + head = name[: -len(generic)] + # 前半截以区划字收尾（`广州市` + `人民政府`）说明这是一个具体主体，不是通用词。 + if head[-1] not in _PLACE_SUFFIX_CHARS: + return True + return False + + +def _category_of(name: str) -> str | None: + for suffix in sorted(_ORG_SUFFIX_TO_CATEGORY, key=len, reverse=True): + if name.endswith(suffix): + return _ORG_SUFFIX_TO_CATEGORY[suffix] + return None + + +def _find_org_spans(text: str) -> list[Identifier]: + """机构名分两趟找。 + + 第一趟按后缀定位，向左吃到边界字为止，得到一批候选。第二趟用**候选之间的自洽关系**把粘在 + 名字前面的句子成分切掉：如果某个候选的一段尾巴本身也是这份文档里的候选，那段尾巴才是真正 + 的名字。`采用公开招标方式组织采购某某医院` 会被缩成 `某某医院`，因为后者在别处独立出现过。 + + 没有这一趟的话，同一个主体会因为前面粘的字不同而拿到好几个不同的假名，模型会把它们当成 + 几个不同的主体，任务难度跟着变。 + + **缩短永远不许把「要替换」变成「不替换」**：只在缩短后的名字仍然通过全部过滤时才缩短。 + 否则 `某某省监狱管理局` 会被缩成通用词 `监狱管理局`、然后整个跳过，那是漏检。 + """ + raw_hits: list[tuple[int, str, str]] = [] + for match in _ORG_RE.finditer(text): + raw = match.group() + cut = max((raw.rfind(char) for char in _ORG_BOUNDARY_CHARS), default=-1) + name = raw[cut + 1 :] + if not name or name.startswith(PSEUDONYM_PREFIX): + continue + category = _category_of(name) + if category is None or len(name) < _ORG_MIN_LEN[category]: + continue + raw_hits.append((match.end(), name, category)) + + known = {name for _, name, _ in raw_hits} + spans: list[Identifier] = [] + for end, name, category in raw_hits: + shorter = _shortest_known_suffix(name, known, category) + chosen = shorter if shorter is not None else name + if _is_generic_org(chosen): + continue + spans.append(Identifier(start=end - len(chosen), end=end, category=category)) + return spans + + +def _shortest_known_suffix(name: str, known: set[str], category: str) -> str | None: + """`name` 的最短那段尾巴，条件是它本身也是这份文档里的候选，且缩短后仍然要被替换。 + + **不在区划字后面切**：`某市某区某医院` 里的 `某区某医院` 就算在别处独立出现过，切在 + `市` 后面也只会把 `某市` 留在原地，与后面的假名连成一个新的候选串，下一轮再整串替换一次。 + 完整的正式名称本来就该整个换掉。 + """ + for length in range(_ORG_MIN_LEN[category], len(name)): + candidate = name[-length:] + if candidate not in known or _is_generic_org(candidate): + continue + if name[-length - 1] in _PLACE_SUFFIX_CHARS: + continue + return candidate + return None + + +def _find_group_spans(pattern: re.Pattern[str], text: str, category: str) -> list[Identifier]: + """按第一个捕获组取范围；没有捕获组就取整个命中。""" + spans: list[Identifier] = [] + for match in pattern.finditer(text): + index = 1 if match.re.groups else 0 + raw = match.group(index) + if raw is None or not raw.strip(): + continue + # 捕获组常把标签后面的空白与尾随的括号一起收进来。把它们留在原文里，替换只吃真正的 + # 那一段——否则「电话：020-1234567 传真」会连中间那个空格一起没掉。 + start, end = match.span(index) + while start < end and text[start].isspace(): + start += 1 + while end > start and ( + text[end - 1].isspace() or (category == "phone" and not text[end - 1].isdigit()) + ): + end -= 1 + if end <= start: + continue + value = text[start:end] + if value.startswith(PSEUDONYM_PREFIX): + continue + if category == "person" and value in _CONTACT_FIELD_LABELS: + continue + if category == "phone" and not any(char.isdigit() for char in value): + continue + spans.append(Identifier(start=start, end=end, category=category)) + return spans + + +def find_identifiers(text: str) -> list[Identifier]: + """找出文本里所有还能识别到具体主体的片段。 + + **脱敏器与校验函数用的是同一份检测**，不是两套正则。两套的话它们迟早漂移，而漂移的方向 + 只有一种能被发现：校验比脱敏严，压测启动不了，有人会去看；反过来校验比脱敏松，泄漏静默 + 通过，没有任何地方会报。 + """ + spans: list[Identifier] = [] + spans += _find_group_spans(_EMAIL_RE, text, "email") + spans += _find_group_spans(_USCC_RE, text, "uscc") + spans += _find_group_spans(_ID_CARD_RE, text, "id_card") + spans += _find_group_spans(_MOBILE_RE, text, "mobile") + spans += _find_group_spans(_LANDLINE_RE, text, "phone") + spans += _find_group_spans(_LABELLED_PHONE_RE, text, "phone") + spans += _find_group_spans(_CONTACT_RE, text, "person") + spans += _find_org_spans(text) + return _resolve_overlaps(spans) + + +def _resolve_overlaps(spans: list[Identifier]) -> list[Identifier]: + """重叠的命中只留一个：起点靠前的优先，起点相同则长的优先。 + + 重叠是常态——邮箱里有数字段，会同时被电话那条规则认领。按起点与长度定序而不是按规则 + 顺序定序，是为了让结果与规则的书写顺序无关。 + """ + ordered = sorted(spans, key=lambda span: (span.start, -(span.end - span.start))) + kept: list[Identifier] = [] + cursor = -1 + for span in ordered: + if span.start < cursor: + continue + kept.append(span) + cursor = span.end + return kept + + +class Redactor: + """把真实文书里的可识别信息换成稳定的假名。 + + **同一个原名在一个 `Redactor` 实例的生命周期里永远换成同一个假名**，跨文档也一样：招标 + 文件与合同里的同一家公司必须还是同一家，否则模型会当成两个主体，任务难度跟着变。 + + 原文与脱敏文本都不写盘，这个类只在内存里工作。 + """ + + __slots__ = ("_aliases", "_counters") + + def __init__(self) -> None: + self._aliases: dict[tuple[str, str], str] = {} + self._counters: dict[str, int] = {} + + def alias_for(self, category: str, original: str) -> str: + key = (category, original) + existing = self._aliases.get(key) + if existing is not None: + return existing + index = self._counters.get(category, 0) + self._counters[category] = index + 1 + tag = _TAGS[index] if index < len(_TAGS) else f"第{index + 1}" + word = _ORG_CATEGORY_WORD.get(category) or _PLAIN_CATEGORY_WORD[category] + alias = ( + f"{PSEUDONYM_PREFIX}{tag}{word}" + if category in _ORG_CATEGORY_WORD + else f"{PSEUDONYM_PREFIX}{word}{tag}" + ) + self._aliases[key] = alias + return alias + + def redact(self, text: str) -> RedactionResult: + counts: dict[str, int] = {} + distinct_keys: set[tuple[str, str]] = set() + current = text + for _ in range(_MAX_REDACTION_PASSES): + spans = find_identifiers(current) + if not spans: + break + pieces: list[str] = [] + cursor = 0 + for span in spans: + original = current[span.start : span.end] + pieces.append(current[cursor : span.start]) + pieces.append(self.alias_for(span.category, original)) + cursor = span.end + counts[span.category] = counts.get(span.category, 0) + 1 + distinct_keys.add((span.category, original)) + pieces.append(current[cursor:]) + current = "".join(pieces) + distinct: dict[str, int] = {} + for category, _original in distinct_keys: + distinct[category] = distinct.get(category, 0) + 1 + return RedactionResult(text=current, counts=counts, distinct=distinct) + + +def assert_no_residue(text: str, *, where: str) -> None: + """脱敏后的硬闸：检出任何可识别信息就抛异常。 + + `where` 是出错时告诉人「哪一份文档没干净」用的，因为语料装配会一次过好几份。 + """ + spans = find_identifiers(text) + if not spans: + return + samples = ", ".join( + f"{span.category}:{_mask(text[span.start : span.end])}" for span in spans[:5] + ) + raise RedactionResidueError( + f"{where} 脱敏后仍检出 {len(spans)} 处可识别信息（样本已掩码）：{samples}。" + "在这里拒绝启动，是因为发出去的请求收不回来，而漏掉一处的表现是压测正常跑完" + ) + + +# --------------------------------------------------------------------------- +# 二、语料：审核点与文档 +# --------------------------------------------------------------------------- + +#: 数据在工作副本里，不在 `reference/` 下——那边是 git 检出，`data/` 没入库。 +DEFAULT_DATA_ROOT = Path("/home/iomgaa/Projects/GovDoc_Editor/data") + +#: 主招标文件，2287 行、174,690 字符。选它是因为单份就远超十万字符，模型塞不进一次上下文， +#: 必然要靠 `grep_document` 与 `read_document` 分段读——那正是这个场景要压的东西。 +DEFAULT_TENDER_FILENAME = "8346d4f0e280c15731827339dc6495d1ab65a27c7856313d38c078e935bec20c.md" + +#: 语料文件名到逻辑名的映射。**工具只认逻辑名**，模型看不见磁盘上的文件名，也就没法用它拼路径。 +DEFAULT_DOCUMENTS: tuple[tuple[str, str], ...] = ((DEFAULT_TENDER_FILENAME, "tender.md"),) + +#: 照 `reference/GovDoc-Editor/govdoc/pipelines/audit_tender.py:498` 的 `verdict_levels`。 +VERDICT_LEVELS: tuple[str, ...] = ("合规", "不合规", "存疑") + + +@dataclass(frozen=True, slots=True, kw_only=True) +class Checkpoint: + """一条审核点：违规情形的判定标准加法条依据，不是问句也不是检查动作。""" + + checkpoint_id: str + category: str + title: str + description: str + legal_basis: tuple[str, ...] + severity: str + + def render(self) -> str: + basis = "、".join(self.legal_basis) if self.legal_basis else "（未拆解）" + return ( + f"审核点编号：{self.checkpoint_id}\n" + f"类别：{self.category}\n" + f"标题：{self.title}\n" + f"判定标准：{self.description}\n" + f"法条依据：{basis}\n" + f"严重程度：{self.severity}" + ) + + +@dataclass(frozen=True, slots=True, kw_only=True) +class CorpusDocument: + """一份脱敏后的文档。正文只在内存里，工具按行区间往外给。""" + + logical_name: str + lines: tuple[str, ...] + + @property + def line_count(self) -> int: + return len(self.lines) + + @property + def char_count(self) -> int: + return sum(len(line) for line in self.lines) + max(0, len(self.lines) - 1) + + @classmethod + def from_text(cls, *, logical_name: str, text: str) -> CorpusDocument: + return cls(logical_name=logical_name, lines=tuple(text.split("\n"))) + + +@dataclass(frozen=True, slots=True, kw_only=True) +class AuditTask: + """一个任务：一条审核点加一份（或几份）文书。三个阶段共用它。""" + + index: int + checkpoint: Checkpoint + documents: tuple[CorpusDocument, ...] + + +def load_checkpoints(*, db_path: Path, limit: int) -> tuple[Checkpoint, ...]: + """从 `checkpointfinal` 读前 `limit` 条已批准的审核点。 + + **只读打开、只跑 SELECT**：这个库是别人项目的生产数据，压测没有任何理由写它。只读是用 + `file:...?mode=ro` 的 URI 形式表达的，不是靠自觉不写。 + + 审核点那侧本来就是纯法规文本、不含机构名，但读进来照样过一遍校验函数——多花的是一次正则 + 扫描，省掉的是「那份数据后来变了、而这条路径上没有闸」这种情形。 + """ + if limit < 1: + raise GovDocScenarioError(f"要取的审核点条数必须 ≥ 1，收到 {limit}") + if not db_path.is_file(): + raise GovDocScenarioError(f"审核点库不在：{db_path}") + uri = f"file:{db_path}?mode=ro" + with closing(sqlite3.connect(uri, uri=True)) as conn: + rows = conn.execute( + "SELECT id, payload_json FROM checkpointfinal WHERE status = 'active' " + "ORDER BY id LIMIT ?", + (limit,), + ).fetchall() + if not rows: + raise GovDocScenarioError(f"{db_path} 的 checkpointfinal 里没有 status='active' 的行") + + checkpoints: list[Checkpoint] = [] + for row_id, payload_json in rows: + try: + payload = json.loads(payload_json) + except json.JSONDecodeError as exc: + raise GovDocScenarioError( + f"审核点 {row_id} 的 payload_json 不是合法 JSON：{exc}" + ) from exc + if not isinstance(payload, dict): + raise GovDocScenarioError(f"审核点 {row_id} 的 payload_json 不是对象") + basis = tuple( + str(item.get("law_name", "")).strip() + for item in payload.get("legal_basis", []) + if isinstance(item, dict) and str(item.get("law_name", "")).strip() + ) + checkpoint = Checkpoint( + checkpoint_id=str(row_id), + category=str(payload.get("category", "")), + title=str(payload.get("title", "")), + description=str(payload.get("description", "")), + legal_basis=basis, + severity=str(payload.get("severity", "")), + ) + assert_no_residue(checkpoint.render(), where=f"审核点 {row_id}") + checkpoints.append(checkpoint) + return tuple(checkpoints) + + +def load_documents( + *, + prepared_dir: Path, + redactor: Redactor, + selection: tuple[tuple[str, str], ...] = DEFAULT_DOCUMENTS, +) -> tuple[tuple[CorpusDocument, RedactionResult], ...]: + """读语料文件，脱敏，过闸，返回文档与它的脱敏账。 + + 返回值带上脱敏账是为了让调用方能报「替换了哪几类各多少处」。原文不在返回值里，也不写盘。 + """ + if not prepared_dir.is_dir(): + raise GovDocScenarioError(f"语料目录不在：{prepared_dir}") + loaded: list[tuple[CorpusDocument, RedactionResult]] = [] + for filename, logical_name in selection: + path = prepared_dir / filename + if not path.is_file(): + raise GovDocScenarioError(f"语料文件不在：{path}") + raw = path.read_text(encoding="utf-8", errors="replace") + result = redactor.redact(raw) + assert_no_residue(result.text, where=f"语料 {logical_name}") + loaded.append( + (CorpusDocument.from_text(logical_name=logical_name, text=result.text), result) + ) + return tuple(loaded) + + +def build_audit_tasks( + *, + data_root: Path = DEFAULT_DATA_ROOT, + checkpoint_count: int, + selection: tuple[tuple[str, str], ...] = DEFAULT_DOCUMENTS, +) -> tuple[tuple[AuditTask, ...], tuple[RedactionResult, ...]]: + """装配任务列表：前 N 条审核点各配同一份（脱敏后的）文书。 + + 文档只脱敏一次、被所有任务共享——`CorpusDocument` 是不可变的，共享没有隔离问题，而每个 + 任务各脱敏一遍要在 174,690 字符上重跑 N 次正则。 + """ + redactor = Redactor() + checkpoints = load_checkpoints(db_path=data_root / "app.sqlite", limit=checkpoint_count) + loaded = load_documents( + prepared_dir=data_root / "storage" / "prepared", + redactor=redactor, + selection=selection, + ) + documents = tuple(document for document, _ in loaded) + reports = tuple(report for _, report in loaded) + tasks = tuple( + AuditTask(index=index, checkpoint=checkpoint, documents=documents) + for index, checkpoint in enumerate(checkpoints) + ) + return tasks, reports + + +# --------------------------------------------------------------------------- +# 三、决策解释：模型输出的 JSON 工具调用 +# --------------------------------------------------------------------------- + +#: 代码围栏：```json 或裸 ```，到配对的 ```。模型很难被劝住不写围栏，而围栏本身不是错误—— +#: 剥掉它再解析，比给一条「不要写围栏」的纠错说明白烧一步划算。 +_FENCE_RE = re.compile(r"```(?:json|JSON)?[ \t\r]*\n(.*?)```", re.DOTALL) + +_NO_JSON = ( + "你的回复里找不到 JSON 对象。这一步只能输出一个 JSON 对象，形如 " + '{"tool": "工具名", "arguments": {"参数名": "参数值"}}，不要写任何解释文字。' +) +_NOT_OBJECT = ( + "解析出来的是 {kind}，不是 JSON 对象。最外层必须是一对花括号，" + '形如 {{"tool": "工具名", "arguments": {{}}}}。' +) +_MISSING_TOOL = ( + '缺少 "tool" 键。JSON 对象里必须有一个 "tool"，它的值是工具清单里的一个工具名（字符串）。' +) +_BAD_ARGUMENTS = ( + '"arguments" 的值是 {kind}，不是对象。它必须是一对花括号包起来的参数表；' + "没有参数就写空对象 {{}}。" +) + + +def _fence_stripped_candidates(content: str) -> list[str]: + """按「最可能是 JSON 的那一段」的顺序给出候选文本。""" + candidates: list[str] = [] + for match in _FENCE_RE.finditer(content): + inner = match.group(1).strip() + if inner: + candidates.append(inner) + stripped = content.strip() + if stripped: + candidates.append(stripped) + # 花括号跨度：模型在 JSON 前后各写了一句话时靠它捞回来。 + start = content.find("{") + end = content.rfind("}") + if start != -1 and end > start: + candidates.append(content[start : end + 1]) + return candidates + + +class GovDocParser: + """把模型输出解释成一次工具调用。 + + 协议是一个 JSON 对象 `{"tool": ..., "arguments": {...}}`。容错两种偏差：被代码围栏包住， + 以及参数平铺在顶层而不是嵌在 `arguments` 里。这两种是实测里最常见的，且都不影响意图—— + 判成失败只会白烧一步。 + + **五种失败各给各的说明**，不合成一句：这段文本就是回喂给模型的那条观察，压成一句会改掉 + 模型收到的纠错信息（`src/polyloop/ports/__init__.py:86`）。 + """ + + def parse(self, reply: ModelReply) -> ParsedReply: + # 契约要求 `len(history_text) <= len(reply.content)`（`tests/contract/ + # test_decision_parser.py:30`），所以直接用原文，不拼接任何东西。 + history = reply.content + decision = self._decide(reply.content) + return ParsedReply(history_text=history, decision=decision) + + def parameters(self) -> Mapping[str, str]: + return {"kind": "govdoc_json_tool_call"} + + def _decide(self, content: str) -> Action | InvalidDecision: + candidates = _fence_stripped_candidates(content) + if not candidates: + return InvalidDecision(explanation=_NO_JSON) + + first_syntax_error: str | None = None + for candidate in candidates: + try: + payload = json.loads(candidate) + except json.JSONDecodeError as exc: + if first_syntax_error is None: + first_syntax_error = ( + f"JSON 语法错误：{exc.msg}（第 {exc.lineno} 行第 {exc.colno} 列）。" + "请重新输出一个语法正确的 JSON 对象，所有标点用英文半角，" + "字符串里的双引号要转义。" + ) + continue + except (TypeError, ValueError, RecursionError) as exc: + # 兜底：`parse` 对任何输入都不许抛异常（契约第 5 条）。这里不是吞错误—— + # 错误原样进了回喂给模型的说明里。 + if first_syntax_error is None: + first_syntax_error = f"JSON 解析失败：{type(exc).__name__}: {exc}" + continue + return self._from_payload(payload) + + if first_syntax_error is not None: + return InvalidDecision(explanation=first_syntax_error) + return InvalidDecision(explanation=_NO_JSON) + + def _from_payload(self, payload: object) -> Action | InvalidDecision: + if not isinstance(payload, dict): + return InvalidDecision(explanation=_NOT_OBJECT.format(kind=type(payload).__name__)) + name = payload.get("tool") + if not isinstance(name, str) or not name.strip(): + return InvalidDecision(explanation=_MISSING_TOOL) + + if "arguments" in payload: + raw_arguments = payload["arguments"] + if not isinstance(raw_arguments, Mapping): + return InvalidDecision( + explanation=_BAD_ARGUMENTS.format(kind=type(raw_arguments).__name__) + ) + arguments = {str(key): value for key, value in raw_arguments.items()} + else: + # 参数平铺：模型把参数写在顶层而不是嵌在 `arguments` 里。除 `tool` 之外的键全当 + # 参数收下——意图是清楚的，判成失败只会白烧一步。 + arguments = {str(key): value for key, value in payload.items() if key != "tool"} + + summary = json.dumps( + {"tool": name.strip(), "arguments": arguments}, + ensure_ascii=False, + separators=(",", ":"), + default=str, + ) + return Action( + text=summary, + tool_call=ToolCall(name=name.strip(), arguments=arguments), + ) + + +# --------------------------------------------------------------------------- +# 四、工具与工作区 +# --------------------------------------------------------------------------- + +#: 一次 `read_document` 最多返回多少行。上限存在是因为主招标文件有 2287 行，模型一句 +#: `end_line: 99999` 就能把整份文书拉进上下文，而这个场景要压的恰恰是「读不完，得分段读」。 +MAX_READ_LINES = 400 + +#: 一次 `grep_document` 最多返回多少条命中。 +MAX_GREP_MATCHES = 50 + +#: 工作区里的审计文件名。**它是环境侧「这个动作真的被执行了几次」的证据**：故障注入那一步要 +#: 验「声明绝不重放的动作不许被执行两次」，模型的说法与轨迹都不算数，只有环境自己记的账算。 +#: 以点开头，是为了让它不被 `read_document` 的工作区文件名规则误当成模型该读的笔记。 +AUDIT_LOG_NAME = ".write_audit.log" + +#: 提交结论落在这个文件里。 +FINDING_NAME = "finding.json" + + +def read_audit_lines(workspace: Path) -> tuple[str, ...]: + """读工作区的审计账。故障注入那一步数它。""" + path = Path(workspace) / AUDIT_LOG_NAME + if not path.is_file(): + return () + return tuple(line for line in path.read_text(encoding="utf-8").splitlines() if line) + + +def _digest(content: str) -> str: + return hashlib.sha256(content.encode("utf-8")).hexdigest()[:12] + + +def _require_str(arguments: Mapping[str, object], key: str) -> str: + value = arguments.get(key) + if not isinstance(value, str): + raise ValueError(f"参数 {key} 必须是字符串，收到 {type(value).__name__}") + return value + + +def _require_int(arguments: Mapping[str, object], key: str) -> int: + value = arguments.get(key) + # bool 是 int 的子类，单独挡掉：`{"start_line": true}` 会被静默当成第 1 行。 + if isinstance(value, bool) or not isinstance(value, int): + raise ValueError(f"参数 {key} 必须是整数，收到 {type(value).__name__}") + return value + + +class GovDocTools: + """四个工具的实现，外加它们共用的工作区。 + + 工作区是一个目录，三个阶段共用同一个——阶段之间靠里面的文件传状态。`read_document` 既读 + 语料里的逻辑名，也读工作区里的纯文件名：没有这条读回路的话，plan 写的 `plan.md` 在 + execute 那次运行里没有任何办法被读到，「工作区传状态」就是句空话。 + """ + + __slots__ = ("_documents", "_workspace") + + def __init__(self, *, documents: tuple[CorpusDocument, ...], workspace: Path) -> None: + self._documents = {document.logical_name: document for document in documents} + self._workspace = Path(workspace) + + @property + def workspace(self) -> Path: + return self._workspace + + # -- 路径解析 --------------------------------------------------------- + + def _lines_for(self, path: str) -> tuple[str, ...]: + """把一个 `path` 解析成行序列。解析不了抛普通 `ValueError`。 + + **抛普通异常而不是 `ToolEnvironmentError`**：库把普通异常判成「动作已执行 + 报错文本 + 回喂」，那正是这里要的——路径写错是模型自己能纠正的事，不该把整次运行判成环境故障 + （`src/polyloop/tools/__init__.py` 的 `RegistryExecutor` 那张表）。 + """ + document = self._documents.get(path) + if document is not None: + return document.lines + if "/" in path or "\\" in path or ".." in path or path.startswith("."): + raise ValueError( + f"path 只能是语料的逻辑名或工作区里的纯文件名，不能是路径：{path!r}。" + f"语料里有：{sorted(self._documents)}" + ) + candidate = self._workspace / path + if candidate.is_file(): + return tuple(candidate.read_text(encoding="utf-8").split("\n")) + raise ValueError( + f"没有这份文档：{path!r}。语料里有：{sorted(self._documents)}；" + f"工作区里有：{sorted(item.name for item in self._workspace.glob('*') if item.is_file() and not item.name.startswith('.'))}" + ) + + def _safe_filename(self, filename: str) -> Path: + if not filename or "/" in filename or "\\" in filename or ".." in filename: + raise ValueError(f"filename 必须是纯文件名，不能含 / 或 ..：{filename!r}") + if filename.startswith("."): + raise ValueError(f"filename 不能以点开头：{filename!r}") + return self._workspace / filename + + def _append_audit(self, tool: str, name: str, content: str) -> None: + """把一次有副作用的调用记进工作区的账，追加写。 + + 记的是「环境侧真的执行了几次」。轨迹里的步记录记的是「库以为执行了几次」，两者在 + 重放判定出错时会分叉，而分叉正是要验的东西——所以证据不能取自轨迹。 + """ + self._workspace.mkdir(parents=True, exist_ok=True) + line = f"{tool}\t{name}\t{_digest(content)}\n" + with (self._workspace / AUDIT_LOG_NAME).open("a", encoding="utf-8") as handle: + handle.write(line) + + # -- 四个 handler ----------------------------------------------------- + + async def read_document(self, arguments: Mapping[str, object]) -> str: + path = _require_str(arguments, "path") + start_line = _require_int(arguments, "start_line") + end_line = _require_int(arguments, "end_line") + lines = self._lines_for(path) + total = len(lines) + if start_line < 1: + raise ValueError(f"start_line 从 1 开始计，收到 {start_line}") + if end_line < start_line: + raise ValueError(f"end_line 不能小于 start_line：{end_line} < {start_line}") + if start_line > total: + raise ValueError(f"{path} 只有 {total} 行，start_line={start_line} 已经越过末尾") + + last = min(end_line, total) + truncated_to = min(last, start_line + MAX_READ_LINES - 1) + body = "\n".join( + f"{number}: {lines[number - 1]}" for number in range(start_line, truncated_to + 1) + ) + if truncated_to < last: + remaining = last - truncated_to + body += ( + f"\n[截断：本次只返回第 {start_line} 到第 {truncated_to} 行；" + f"请求的区间还剩 {remaining} 行未返回，{path} 共 {total} 行]" + ) + return body + + async def grep_document(self, arguments: Mapping[str, object]) -> str: + pattern = _require_str(arguments, "pattern") + path = _require_str(arguments, "path") + # max_matches 是可选的。**先判「传没传」再判类型**：直接 `_require_int` 会把 + # 「没传」当成「传了个 None」报错，而这个参数的 schema 里根本没把它写进 required。 + if arguments.get("max_matches") is None: + limit = MAX_GREP_MATCHES + else: + limit = max(1, min(_require_int(arguments, "max_matches"), MAX_GREP_MATCHES)) + lines = self._lines_for(path) + try: + compiled = re.compile(pattern) + except re.error as exc: + # 正则写错是模型自己能纠正的事，回喂原始报错让它改。 + raise ValueError(f"pattern 不是合法正则：{exc}") from exc + + hits = [ + f"{number}: {line}" + for number, line in enumerate(lines, start=1) + if compiled.search(line) + ] + if not hits: + return f"{path} 里没有命中 {pattern!r}，共 {len(lines)} 行。换个关键词再试。" + shown = hits[:limit] + body = "\n".join(shown) + if len(hits) > len(shown): + body += f"\n[共 {len(hits)} 处命中，只返回前 {len(shown)} 处。缩小 pattern 再试]" + return body + + async def write_note(self, arguments: Mapping[str, object]) -> str: + filename = _require_str(arguments, "filename") + content = _require_str(arguments, "content") + target = self._safe_filename(filename) + self._workspace.mkdir(parents=True, exist_ok=True) + target.write_text(content, encoding="utf-8") + self._append_audit("write_note", filename, content) + return f"已写入 {filename}（{len(content)} 字符）。" + + async def submit_finding(self, arguments: Mapping[str, object]) -> str: + verdict = _require_str(arguments, "verdict") + evidence = _require_str(arguments, "evidence") + reasoning = _require_str(arguments, "reasoning") + if verdict not in VERDICT_LEVELS: + raise ValueError(f"verdict 只能取 {list(VERDICT_LEVELS)}，收到 {verdict!r}") + payload = json.dumps( + {"verdict": verdict, "evidence": evidence, "reasoning": reasoning}, + ensure_ascii=False, + indent=2, + ) + self._workspace.mkdir(parents=True, exist_ok=True) + (self._workspace / FINDING_NAME).write_text(payload, encoding="utf-8") + self._append_audit("submit_finding", FINDING_NAME, payload) + return f"结论已提交：{verdict}。本次运行到此结束。" + + # -- 注册表 ----------------------------------------------------------- + + def registry(self) -> ToolRegistry: + """全部四个工具。阶段收窄由 `restrict_to` 在装配时做。""" + return ToolRegistry( + ( + ToolSpec( + name="read_document", + description=( + "按行区间读一份文档。path 是语料的逻辑名（如 tender.md）或工作区里的" + "纯文件名（如 plan.md），不接受任何路径。返回的每行前面带行号。" + f"单次最多返回 {MAX_READ_LINES} 行，超出会截断并注明还剩多少行。" + ), + parameters={ + "type": "object", + "properties": { + "path": {"type": "string", "description": "文档的逻辑名或工作区文件名"}, + "start_line": { + "type": "integer", + "description": "起始行，从 1 开始，含", + }, + "end_line": {"type": "integer", "description": "结束行，含"}, + }, + "required": ["path", "start_line", "end_line"], + "additionalProperties": False, + }, + # 只读且幂等：状态未知时重放它不会产生任何副作用。 + replay_policy=ReplayPolicy.SAFE, + handler=self.read_document, + ), + ToolSpec( + name="grep_document", + description=( + "在一份文档里做正则搜索，返回「行号: 行内容」。" + f"最多返回 {MAX_GREP_MATCHES} 条。正则语法是 Python 的 re。" + ), + parameters={ + "type": "object", + "properties": { + "pattern": {"type": "string", "description": "Python re 正则"}, + "path": {"type": "string", "description": "文档的逻辑名或工作区文件名"}, + "max_matches": { + "type": "integer", + "description": f"最多返回几条，上限 {MAX_GREP_MATCHES}", + }, + }, + "required": ["pattern", "path"], + "additionalProperties": False, + }, + replay_policy=ReplayPolicy.SAFE, + handler=self.grep_document, + ), + ToolSpec( + name="write_note", + description=( + "往工作区写一个文件，同名覆盖。filename 必须是纯文件名。" + "写下的文件在后续阶段可以用 read_document 读回来。" + ), + parameters={ + "type": "object", + "properties": { + "filename": {"type": "string", "description": "纯文件名，如 plan.md"}, + "content": {"type": "string", "description": "文件全文"}, + }, + "required": ["filename", "content"], + "additionalProperties": False, + }, + # **绝不重放。** 它覆盖工作区里的文件，重放一次就把上一次的内容盖掉，而 + # 库在「状态未知」时无法分辨这次写有没有落盘。故障注入验「声明绝不重放的 + # 动作不许被执行两次」时，数的就是这个工具在审计账里的行数。 + replay_policy=ReplayPolicy.NEVER, + handler=self.write_note, + ), + ToolSpec( + name="submit_finding", + description=( + f"提交本次审核的唯一结论。verdict 只能取 {list(VERDICT_LEVELS)}；" + "evidence 必须是从文书里逐字摘录的原文，注明文档名与行号；" + "reasoning 说明判断依据。提交成功即本次运行结束。" + ), + parameters={ + "type": "object", + "properties": { + "verdict": { + "type": "string", + "description": "合规 / 不合规 / 存疑，三选一", + }, + "evidence": {"type": "string", "description": "逐字摘录的原文与出处"}, + "reasoning": {"type": "string", "description": "判断依据"}, + }, + "required": ["verdict", "evidence", "reasoning"], + "additionalProperties": False, + }, + replay_policy=ReplayPolicy.NEVER, + # agent 自报的完成通路：它调这个工具来宣布做完，环境状态没有被独立验证过。 + # 与 `ActionOutcome.env_reported_completion` 是两条不同的通路，可信度不同。 + completes_run=True, + handler=self.submit_finding, + ), + ) + ) + + +# --------------------------------------------------------------------------- +# 五、阶段：工具集、预算、提示词 +# --------------------------------------------------------------------------- + +PHASES: tuple[str, ...] = ("plan", "execute", "summarize") + +#: 每个阶段模型能看见的工具。**summarize 那次收窄是重点**：检索工具被收走，照 +#: `gov-auditor.yaml` L22-27 的 summarize 阶段收走 Grep。plan 与 execute 的工具集相同， +#: 它们的区别在提示词与预算——这与 `gov-auditor.yaml` L9-20 一致（那边两个阶段只差一个 Skill）。 +PHASE_TOOLS: Mapping[str, tuple[str, ...]] = { + "plan": ("read_document", "grep_document", "write_note"), + "execute": ("read_document", "grep_document", "write_note"), + "summarize": ("read_document", "submit_finding"), +} + +#: 步数与动作上限照 `gov-auditor.yaml` L9-25 的 `max_turns`：plan 50 / execute 50 / +#: summarize 16。那三个数是 GovDoc 跑了 264 次真实审核之后定下来的，实测均值是 plan 23、 +#: execute 26、summarize 7 轮，上限留了约一倍余量。 +#: +#: `max_prompt_chars` 取 400,000：主招标文件是 174,690 字符，一次运行里模型会分段读进来 +#: 相当一部分，再加上工具清单与历史，四十万给的是「读得进去但读不完」的空间——那正是这个 +#: 场景要压的形态。 +#: +#: `max_consecutive_parse_failures` 取 3：JSON 协议比代码围栏协议好写得多，连错三次说明模型 +#: 没在按协议输出，再给机会只是烧钱。 +PHASE_BUDGETS: Mapping[str, Budget] = { + "plan": Budget( + max_steps=50, max_actions=50, max_consecutive_parse_failures=3, max_prompt_chars=400_000 + ), + "execute": Budget( + max_steps=50, max_actions=50, max_consecutive_parse_failures=3, max_prompt_chars=400_000 + ), + "summarize": Budget( + max_steps=16, max_actions=16, max_consecutive_parse_failures=3, max_prompt_chars=400_000 + ), +} + +OBSERVATION_TEMPLATE = "工具返回：\n{observation}\n\n" + +CANCEL_GRACE_SECONDS = 5.0 + +SYNTHETIC_OBSERVATIONS = SyntheticObservations( + action_rejected=( + "[动作被拒绝，这一步没有执行任何工具]\n" + "请对照工具清单检查工具名与参数，然后重新输出一个 JSON 对象。" + ), + env_failed=( + "[环境故障，这一步的动作没有被执行]\n" + "换一种方式继续；如果同一个动作连续故障，改用别的工具推进。" + ), + model_call_failed=("[模型调用失败，这一步没有产生任何输出]\n请重新给出这一步的 JSON 对象。"), +) + +_PROTOCOL_BLOCK = """输出格式（每一步只输出一个 JSON 对象，不要写解释文字）： +{"tool": "工具名", "arguments": {"参数名": "参数值"}} + +例： +{"tool": "grep_document", "arguments": {"pattern": "注册地|所在地|分支机构", "path": "tender.md", "max_matches": 20}} + +绝对禁止（违反会浪费这一步）： +- 禁止在 JSON 之外写任何文字、标题或思考过程。 +- 禁止一次输出多个 JSON 对象。 +- 禁止调用工具清单以外的工具，或传清单里没声明的参数。 +- 禁止凭印象断言文书内容；每一条结论都必须先用工具读到原文。 +- 所有 JSON 标点用英文半角。""" + +_PHASE_INSTRUCTIONS: Mapping[str, str] = { + "plan": """你是政府采购合规审核专家。本阶段是 plan：针对一条审核点，在招标文书里定位可能相关的段落，产出审核计划。 + +操作步骤（严格按序，不跳步、不加步）： +1. 用 grep_document 按审核点判定标准里的关键词检索文书，拿到候选行号。关键词要换几组试，不要只搜一次。 +2. 用 read_document 读候选行号前后的上下文，确认这一处是不是真的与审核点相关。 +3. 用 write_note 写 plan.md：逐条列出候选证据的文档名、行号区间、以及它与审核点的关系。 +4. plan.md 写完就停下，不要重复步骤 1 到 3。 + +本阶段不下结论，也没有提交工具；结论在后面的阶段提交。""", + "execute": """你是政府采购合规审核专家。本阶段是 execute：按上一阶段的计划逐条取证。 + +操作步骤（严格按序，不跳步、不加步）： +1. 用 read_document 读 plan.md（它在工作区里，直接写文件名）。 +2. 按 plan.md 给的行号区间逐条 read_document 核实原文；不够就用 grep_document 补检索。 +3. 用 write_note 写 evidence.md：每条证据一段，含文档名、行号、逐字摘录的原文（不要改写）、以及它支持还是反对「不合规」这个判断。 +4. evidence.md 写完就停下。 + +本阶段不下结论，也没有提交工具。""", + "summarize": """你是政府采购合规审核专家。本阶段是 summarize：汇总证据并提交唯一一条结论。 + +操作步骤（严格按序，不跳步、不加步）： +1. 用 read_document 读 evidence.md（它在工作区里，直接写文件名）。 +2. 需要复核原文时用 read_document 读语料文档。本阶段没有检索工具，不要尝试调用 grep_document。 +3. 用 submit_finding 提交结论。verdict 只能取 合规 / 不合规 / 存疑。 +4. 提交成功即本次运行结束，不要再调用任何工具。 + +证据不足时给「存疑」，不要猜测。""", +} + + +def build_context(*, task: AuditTask, phase: str, tools: ToolRegistry) -> Context: + """装配这一次运行的上下文。 + + **按变化频率从低到高分段**：`run_level` 放阶段提示词与工具清单（同一阶段的每个任务都一样）， + `goal_level` 放审核点与文档清单（每个任务不同）。这个分法是为了前缀缓存——稳定的那一段 + 排在前面，才有可能被命中。 + + **公文正文不进上下文。** 文书要靠 `read_document` 与 `grep_document` 一点点读出来， + `goal_level` 里只给逻辑名、总行数、总字符数。整篇塞进去的话，第一步的提示词就是十七万 + 字符，之后每一步都在重复付这份钱，而且模型再也不需要检索——这个场景要压的正是「读不完， + 得自己找」。 + """ + if phase not in _PHASE_INSTRUCTIONS: + raise GovDocScenarioError(f"没有这个阶段：{phase!r}，只有 {list(PHASES)}") + schema = json.dumps(list(tools.schema_for_model()), ensure_ascii=False, indent=2) + system = ( + f"{_PHASE_INSTRUCTIONS[phase]}\n\n" + f"本阶段可用的工具（JSON Schema，多一个参数都不认）：\n{schema}\n\n" + f"{_PROTOCOL_BLOCK}" + ) + manifest = "\n".join( + f"- {document.logical_name}：共 {document.line_count} 行，{document.char_count} 字符" + for document in task.documents + ) + goal = ( + f"本次审核点（原文）：\n{task.checkpoint.render()}\n\n" + f"可读的语料文档（正文不在上下文里，必须用工具读）：\n{manifest}\n\n" + "工作区里的文件用纯文件名读写（如 plan.md、evidence.md）。" + ) + return Context( + run_level=(Message(role=Role.SYSTEM, content=(TextBlock(text=system),)),), + goal_level=(Message(role=Role.USER, content=(TextBlock(text=goal),)),), + ) + + +def make_run_id(*, task_index: int, phase: str) -> str: + """`govdoc-<任务序号>-<阶段>`。必须匹配 `[A-Za-z0-9._-]+`（`JsonlRunStore` 的约束）。""" + if phase not in _PHASE_INSTRUCTIONS: + raise GovDocScenarioError(f"没有这个阶段：{phase!r}，只有 {list(PHASES)}") + return f"govdoc-{task_index}-{phase}" + + +def build_run_request( + *, + task: AuditTask, + phase: str, + run_id: str, + workspace: Path, + model_binding: Mapping[str, str], +) -> RunRequest: + """装好一次运行的请求。三个阶段各调一次，工作区传同一个。 + + **动作执行器必须从收窄之后的注册表派生**：`RunRequest` 会比对 + `action_executor.registry == tools`（`src/polyloop/session/__init__.py:156-170`）， + 从全量注册表派生再传收窄注册表会当场报错。这条校验挡的是「模型看见的 schema 与实际分发 + 来自两份不同的工具集」。 + """ + if phase not in _PHASE_INSTRUCTIONS: + raise GovDocScenarioError(f"没有这个阶段：{phase!r}，只有 {list(PHASES)}") + handlers = GovDocTools(documents=task.documents, workspace=workspace) + tools = handlers.registry().restrict_to(PHASE_TOOLS[phase]) + return RunRequest( + run_id=run_id, + budget=PHASE_BUDGETS[phase], + action_executor=tools.executor(), + tools=tools, + context=build_context(task=task, phase=phase, tools=tools), + injections={}, + model_binding=dict(model_binding), + # 模型调用绝不重放：一次调用的钱已经花了，重放会再花一次，而且两次的输出不保证相同。 + model_replay_policy=ReplayPolicy.NEVER, + observation_template=OBSERVATION_TEMPLATE, + cancel_grace_seconds=CANCEL_GRACE_SECONDS, + ) diff --git a/tools/soak/tests/test_govdoc_scenario.py b/tools/soak/tests/test_govdoc_scenario.py new file mode 100644 index 0000000..771daa5 --- /dev/null +++ b/tools/soak/tests/test_govdoc_scenario.py @@ -0,0 +1,568 @@ +"""GovDoc 公文审核场景适配器的测试。 + +分五块：脱敏、解析器、工具、阶段收窄、装配。 + +脱敏那块里最重要的一条是「校验函数对未脱敏文本确实会抛异常」。一个永远返回通过的校验函数比 +没有校验更糟：它会让所有人以为这道闸在守着，而它什么都没守。 + +解析器那块的前五条是 `tests/contract/test_decision_parser.py` 那份公共契约的逐条复刻。契约 +套件本身是给下游接自己的实现用的（在自己的 `conftest.py` 里覆盖 fixture），压测这边不接那套 +装配、只把五条断言照着写一遍——它是任何新适配器的准入标准，压测的适配器也是适配器。 + +用真实数据的那几条在数据目录不在时跳过而不是失败：那份数据是另一个项目的工作副本，不在本仓库 +里，换一台机器就没有。 +""" + +from __future__ import annotations + +import json +import re +from typing import TYPE_CHECKING + +import pytest + +from polyloop.ports import Action, InvalidDecision +from polyloop.tools import ToolRegistry +from polyloop.types import ModelReply, ReplayPolicy +from tools.soak.scenarios.govdoc import ( + AUDIT_LOG_NAME, + DEFAULT_DATA_ROOT, + FINDING_NAME, + MAX_READ_LINES, + PHASE_TOOLS, + VERDICT_LEVELS, + AuditTask, + Checkpoint, + CorpusDocument, + GovDocParser, + GovDocScenarioError, + GovDocTools, + RedactionResidueError, + Redactor, + assert_no_residue, + build_audit_tasks, + build_context, + build_run_request, + make_run_id, + read_audit_lines, +) + +if TYPE_CHECKING: + from pathlib import Path + +#: 一段自造的「像真的」文本：机构全称、医院、财政局、固定电话、统一社会信用代码、邮箱、 +#: 联系人各一处，同一家公司出现两次。 +#: +#: **全部是编造的，一个字都不取自真实文书**：机构名前面带「虚构」两字，邮箱域名带 example， +#: 号码是连号。测试数据本身要是可识别的，那这份测试就成了它自己要挡的那种泄漏。 +DIRTY_TEXT = """项目名称：某设备采购 +采购人：虚构市第三人民医院 +采购代理机构：虚构鸿远工程咨询有限公司 +监督部门：虚构市财政局 +代理机构地址：虚构市朝阳街道 88 号 +联系人：赵明 +电话：0768-12345678 +邮箱：zhaoming@example-invalid.cn +统一社会信用代码：91445102MA4XK7YQ3B +中标供应商：虚构鸿远工程咨询有限公司 +预算金额：8,736,100.00 元 +项目编号：440513-2023-03374 +""" + + +def _reply(content: str) -> ModelReply: + return ModelReply(call_id="call-1", content=content, thinking="") + + +def _sample_task() -> AuditTask: + checkpoint = Checkpoint( + checkpoint_id="cp-1", + category="不合理条件限制或排斥供应商", + title="1.直接或变相对外地企业进入本地市场设置阻碍。", + description="采购文件设置供应商注册地等不合理的资格条件、评审因素。", + legal_basis=("政府采购法第5条", "第22条第二款"), + severity="major", + ) + document = CorpusDocument.from_text( + logical_name="tender.md", + text="\n".join(f"第 {number} 行：投标人须在本地注册。" for number in range(1, 1001)), + ) + return AuditTask(index=0, checkpoint=checkpoint, documents=(document,)) + + +def _tools(tmp_path: Path) -> GovDocTools: + return GovDocTools(documents=_sample_task().documents, workspace=tmp_path) + + +# --------------------------------------------------------------------------- +# 一、脱敏 +# --------------------------------------------------------------------------- + + +def test_validator_rejects_unredacted_text(): + """这条是这份测试里最要紧的一条：校验函数必须真的会拒。""" + with pytest.raises(RedactionResidueError): + assert_no_residue(DIRTY_TEXT, where="自造样本") + + +def test_validator_accepts_redacted_text(): + result = Redactor().redact(DIRTY_TEXT) + assert_no_residue(result.text, where="自造样本") + + +def test_every_identifier_category_is_replaced(): + result = Redactor().redact(DIRTY_TEXT) + for category in ("company", "hospital", "bureau", "phone", "email", "uscc", "person"): + assert result.counts.get(category, 0) >= 1, f"{category} 一处都没替换：{result.counts}" + for leaked in ( + "虚构鸿远工程咨询有限公司", + "虚构市第三人民医院", + "虚构市财政局", + "0768-12345678", + "zhaoming@example-invalid.cn", + "91445102MA4XK7YQ3B", + "赵明", + ): + assert leaked not in result.text + + +def test_same_original_gets_one_stable_alias(): + result = Redactor().redact(DIRTY_TEXT) + # 那家公司在原文里出现两次（第 3 行的代理机构、倒数第 3 行的中标供应商）， + # 替换后必须还是同一个假名、同样两次。 + company_alias = result.text.splitlines()[2].split("：", 1)[1] + assert company_alias.startswith("示例") + assert result.text.count(company_alias) == 2 + assert result.distinct["company"] == 1 + + +def test_alias_is_stable_across_documents(): + redactor = Redactor() + first = redactor.redact(DIRTY_TEXT).text + second = redactor.redact("中标人是虚构鸿远工程咨询有限公司。").text + alias = first.splitlines()[2].split("：", 1)[1] + assert alias in second + + +def test_amounts_and_project_numbers_survive(): + result = Redactor().redact(DIRTY_TEXT) + assert "8,736,100.00" in result.text + assert "440513-2023-03374" in result.text + + +def test_generic_institution_words_survive(): + text = "评标委员会依法组建，投标人可由总公司授权分公司投标。" + result = Redactor().redact(text) + assert result.text == text + + +def test_redaction_is_idempotent(): + once = Redactor().redact(DIRTY_TEXT).text + twice = Redactor().redact(once).text + assert twice == once + + +# --------------------------------------------------------------------------- +# 二、决策解释器：五条公共契约 + 两种容错 + 五种失败 +# --------------------------------------------------------------------------- + +_ACTION_REPLY = ( + '{"tool": "read_document", "arguments": {"path": "tender.md", "start_line": 1, "end_line": 20}}' +) +_INVALID_REPLY = "我先想想应该从哪里开始查。" + + +def test_contract_1_parse_is_synchronous(): + parsed = GovDocParser().parse(_reply(_ACTION_REPLY)) + assert not hasattr(parsed, "__await__") + + +@pytest.mark.parametrize("content", [_ACTION_REPLY, _INVALID_REPLY, "", "```json\n{}\n```"]) +def test_contract_2_history_text_is_not_longer(content: str): + parsed = GovDocParser().parse(_reply(content)) + assert isinstance(parsed.history_text, str) + assert len(parsed.history_text) <= len(content) + + +def test_contract_3_invalid_decision_explains_itself(): + parsed = GovDocParser().parse(_reply(_INVALID_REPLY)) + assert isinstance(parsed.decision, InvalidDecision) + assert parsed.decision.explanation.strip() + + +def test_contract_4_action_carries_text(): + parsed = GovDocParser().parse(_reply(_ACTION_REPLY)) + assert isinstance(parsed.decision, Action) + assert isinstance(parsed.decision.text, str) + assert parsed.decision.text + assert parsed.decision.tool_call is not None + assert parsed.decision.tool_call.name == "read_document" + assert parsed.decision.tool_call.arguments["path"] == "tender.md" + + +@pytest.mark.parametrize( + "content", + [ + "", + " ", + "{", + "```json\n", + "[1, 2, 3]", + '{"tool": null}', + '{"tool": "x", "arguments": 5}', + "```\n```", + "\x00\x01", + "{" * 500, + ], +) +def test_contract_5_parse_never_raises(content: str): + parsed = GovDocParser().parse(_reply(content)) + assert isinstance(parsed.decision, Action | InvalidDecision) + + +def test_tolerates_code_fences(): + for fenced in ( + f"```json\n{_ACTION_REPLY}\n```", + f"```\n{_ACTION_REPLY}\n```", + f"我打算先读一段：\n```json\n{_ACTION_REPLY}\n```\n读完再说。", + ): + parsed = GovDocParser().parse(_reply(fenced)) + assert isinstance(parsed.decision, Action), fenced + assert parsed.decision.tool_call is not None + assert parsed.decision.tool_call.name == "read_document" + + +def test_tolerates_flattened_arguments(): + parsed = GovDocParser().parse( + _reply('{"tool": "read_document", "path": "tender.md", "start_line": 1, "end_line": 20}') + ) + assert isinstance(parsed.decision, Action) + assert parsed.decision.tool_call is not None + assert parsed.decision.tool_call.arguments == { + "path": "tender.md", + "start_line": 1, + "end_line": 20, + } + + +def test_five_parse_failures_get_five_explanations(): + parser = GovDocParser() + explanations = {} + for label, content in { + "没有 JSON": "我准备开始审核了。", + "语法错": '{"tool": "read_document", "arguments": {,}}', + "不是对象": "[1, 2, 3]", + "缺 tool": '{"arguments": {"path": "tender.md"}}', + "arguments 不是对象": '{"tool": "read_document", "arguments": "tender.md"}', + }.items(): + decision = parser.parse(_reply(content)).decision + assert isinstance(decision, InvalidDecision), label + explanations[label] = decision.explanation + assert len(set(explanations.values())) == 5, explanations + assert "语法" in explanations["语法错"] + assert "tool" in explanations["缺 tool"] + assert "arguments" in explanations["arguments 不是对象"] + + +def test_parser_reports_its_parameters(): + assert GovDocParser().parameters() == {"kind": "govdoc_json_tool_call"} + + +# --------------------------------------------------------------------------- +# 三、四个工具 +# --------------------------------------------------------------------------- + + +async def test_read_document_returns_numbered_lines(tmp_path: Path): + body = await _tools(tmp_path).read_document( + {"path": "tender.md", "start_line": 3, "end_line": 5} + ) + assert body.splitlines() == [ + "3: 第 3 行：投标人须在本地注册。", + "4: 第 4 行：投标人须在本地注册。", + "5: 第 5 行：投标人须在本地注册。", + ] + + +async def test_read_document_truncates_and_says_so(tmp_path: Path): + body = await _tools(tmp_path).read_document( + {"path": "tender.md", "start_line": 1, "end_line": 1000} + ) + lines = body.splitlines() + assert len(lines) == MAX_READ_LINES + 1 + assert lines[MAX_READ_LINES - 1].startswith(f"{MAX_READ_LINES}: ") + assert f"第 1 到第 {MAX_READ_LINES} 行" in lines[-1] + assert f"还剩 {1000 - MAX_READ_LINES} 行未返回" in lines[-1] + assert "共 1000 行" in lines[-1] + + +@pytest.mark.parametrize( + "path", ["../etc/passwd", "/etc/passwd", "notes/plan.md", "nope.md", ".write_audit.log"] +) +async def test_read_document_rejects_unregistered_path(tmp_path: Path, path: str): + with pytest.raises(ValueError): + await _tools(tmp_path).read_document({"path": path, "start_line": 1, "end_line": 2}) + + +async def test_read_document_reads_workspace_notes(tmp_path: Path): + tools = _tools(tmp_path) + await tools.write_note({"filename": "plan.md", "content": "第一条：查注册地要求"}) + body = await tools.read_document({"path": "plan.md", "start_line": 1, "end_line": 10}) + assert "第一条：查注册地要求" in body + + +async def test_grep_document_caps_matches(tmp_path: Path): + tools = _tools(tmp_path) + body = await tools.grep_document({"pattern": "本地注册", "path": "tender.md", "max_matches": 3}) + lines = body.splitlines() + assert lines[0].startswith("1: ") + assert len(lines) == 4 + assert "共 1000 处命中" in lines[-1] + + +async def test_grep_document_raises_on_bad_pattern(tmp_path: Path): + # 报错文本要指向正则本身。断言这一句是为了挡住「因为别的参数报错而恰好也抛了 ValueError」 + # 那种假绿——这条曾经真的因为可选参数 max_matches 没传而在别处先炸掉。 + with pytest.raises(ValueError, match="正则"): + await _tools(tmp_path).grep_document({"pattern": "([", "path": "tender.md"}) + + +async def test_grep_document_max_matches_is_optional(tmp_path: Path): + body = await _tools(tmp_path).grep_document({"pattern": "本地注册", "path": "tender.md"}) + assert len(body.splitlines()) == 51 + + +@pytest.mark.parametrize("filename", ["../escape.md", "sub/plan.md", "", ".hidden"]) +async def test_write_note_rejects_paths(tmp_path: Path, filename: str): + with pytest.raises(ValueError): + await _tools(tmp_path).write_note({"filename": filename, "content": "x"}) + + +async def test_write_note_writes_and_audits(tmp_path: Path): + tools = _tools(tmp_path) + await tools.write_note({"filename": "plan.md", "content": "第一版"}) + await tools.write_note({"filename": "plan.md", "content": "第二版"}) + assert (tmp_path / "plan.md").read_text(encoding="utf-8") == "第二版" + audit = read_audit_lines(tmp_path) + assert len(audit) == 2 + assert all(line.startswith("write_note\tplan.md\t") for line in audit) + # 两次内容不同，摘要也要不同——审计要能区分「同一个动作被执行了两次」与「两次写的是同一份」。 + assert audit[0] != audit[1] + + +@pytest.mark.parametrize("verdict", ["合格", "compliant", "", "合规 "]) +async def test_submit_finding_rejects_bad_verdict(tmp_path: Path, verdict: str): + with pytest.raises(ValueError): + await _tools(tmp_path).submit_finding( + {"verdict": verdict, "evidence": "第 3 行", "reasoning": "无"} + ) + + +async def test_submit_finding_writes_and_audits(tmp_path: Path): + tools = _tools(tmp_path) + for verdict in VERDICT_LEVELS: + await tools.submit_finding( + {"verdict": verdict, "evidence": "tender.md 第 3 行", "reasoning": "见证据"} + ) + payload = json.loads((tmp_path / FINDING_NAME).read_text(encoding="utf-8")) + assert payload["verdict"] == VERDICT_LEVELS[-1] + audit = read_audit_lines(tmp_path) + assert len(audit) == len(VERDICT_LEVELS) + assert all(line.startswith(f"submit_finding\t{FINDING_NAME}\t") for line in audit) + + +async def test_audit_log_is_not_readable_by_the_model(tmp_path: Path): + """审计是环境侧的账，不是给模型看的材料。""" + tools = _tools(tmp_path) + await tools.write_note({"filename": "plan.md", "content": "x"}) + assert (tmp_path / AUDIT_LOG_NAME).is_file() + with pytest.raises(ValueError): + await tools.read_document({"path": AUDIT_LOG_NAME, "start_line": 1, "end_line": 2}) + + +async def test_wrong_argument_types_raise_plain_errors(tmp_path: Path): + tools = _tools(tmp_path) + with pytest.raises(ValueError): + await tools.read_document({"path": "tender.md", "start_line": True, "end_line": 2}) + with pytest.raises(ValueError): + await tools.read_document({"path": 3, "start_line": 1, "end_line": 2}) + + +def test_replay_policies_and_completion_flag(tmp_path: Path): + registry = _tools(tmp_path).registry() + assert registry.names() == ( + "read_document", + "grep_document", + "write_note", + "submit_finding", + ) + policies = {name: registry.spec_for(name).replay_policy for name in registry.names()} + assert policies == { + "read_document": ReplayPolicy.SAFE, + "grep_document": ReplayPolicy.SAFE, + "write_note": ReplayPolicy.NEVER, + "submit_finding": ReplayPolicy.NEVER, + } + completes = [name for name in registry.names() if registry.spec_for(name).completes_run] + assert completes == ["submit_finding"] + + +def test_tool_parameters_are_closed_json_schema(tmp_path: Path): + for spec in _tools(tmp_path).registry().schema_for_model(): + parameters = spec["parameters"] + assert parameters["type"] == "object" + assert parameters["additionalProperties"] is False + assert parameters["required"] + json.dumps(spec, ensure_ascii=False) + + +# --------------------------------------------------------------------------- +# 四、阶段收窄 +# --------------------------------------------------------------------------- + + +def test_summarize_drops_grep_keeps_submit(tmp_path: Path): + full = _tools(tmp_path).registry() + narrowed = full.restrict_to(PHASE_TOOLS["summarize"]) + assert "grep_document" not in narrowed.names() + assert "write_note" not in narrowed.names() + assert "submit_finding" in narrowed.names() + assert "read_document" in narrowed.names() + + +def test_restrict_to_leaves_the_source_registry_alone(tmp_path: Path): + full = _tools(tmp_path).registry() + before = full.names() + full.restrict_to(PHASE_TOOLS["summarize"]) + assert full.names() == before + assert len(before) == 4 + + +def test_plan_and_execute_have_no_submit_tool(tmp_path: Path): + full = _tools(tmp_path).registry() + for phase in ("plan", "execute"): + narrowed = full.restrict_to(PHASE_TOOLS[phase]) + assert "submit_finding" not in narrowed.names() + assert narrowed.names() == ("read_document", "grep_document", "write_note") + + +# --------------------------------------------------------------------------- +# 五、装配 +# --------------------------------------------------------------------------- + + +def test_run_id_shape(): + assert make_run_id(task_index=7, phase="plan") == "govdoc-7-plan" + for phase in ("plan", "execute", "summarize"): + assert re.fullmatch(r"[A-Za-z0-9._\-]+", make_run_id(task_index=12, phase=phase)) + with pytest.raises(GovDocScenarioError): + make_run_id(task_index=7, phase="finalize") + + +def test_all_three_phases_assemble(tmp_path: Path): + task = _sample_task() + for phase in ("plan", "execute", "summarize"): + request = build_run_request( + task=task, + phase=phase, + run_id=make_run_id(task_index=task.index, phase=phase), + workspace=tmp_path, + model_binding={"session_id": "soak-1"}, + ) + # 库会校验执行器与本次可见注册表同源（session/__init__.py:156-170）。 + assert request.action_executor.registry == request.tools + assert request.tools.names() == PHASE_TOOLS[phase] + assert "{observation}" in request.observation_template + assert request.model_replay_policy is ReplayPolicy.NEVER + assert request.cancel_grace_seconds == 5.0 + assert request.injections == {} + + +def test_budgets_follow_gov_auditor_turn_limits(tmp_path: Path): + task = _sample_task() + budgets = { + phase: build_run_request( + task=task, + phase=phase, + run_id=make_run_id(task_index=0, phase=phase), + workspace=tmp_path, + model_binding={}, + ).budget + for phase in ("plan", "execute", "summarize") + } + assert (budgets["plan"].max_steps, budgets["plan"].max_actions) == (50, 50) + assert (budgets["execute"].max_steps, budgets["execute"].max_actions) == (50, 50) + assert (budgets["summarize"].max_steps, budgets["summarize"].max_actions) == (16, 16) + for budget in budgets.values(): + assert budget.max_consecutive_parse_failures == 3 + assert budget.max_prompt_chars == 400_000 + + +def test_context_carries_tools_but_not_the_document_body(tmp_path: Path): + task = _sample_task() + registry = _tools(tmp_path).registry().restrict_to(PHASE_TOOLS["summarize"]) + context = build_context(task=task, phase="summarize", tools=registry) + system = context.run_level[0].content[0].text + goal = context.goal_level[0].content[0].text + # 提示词正文里会点名说「本阶段没有检索工具」，所以工具清单要在 schema 那一段里查。 + assert '"name": "submit_finding"' in system + assert '"name": "grep_document"' not in system + assert task.checkpoint.title in goal + assert "共 1000 行" in goal + # 公文正文必须靠工具读，不能整篇塞进上下文。 + assert task.documents[0].lines[0] not in system + goal + + +async def test_phases_share_state_through_the_workspace(tmp_path: Path): + task = _sample_task() + plan_tools = GovDocTools(documents=task.documents, workspace=tmp_path) + await plan_tools.write_note({"filename": "evidence.md", "content": "证据一：第 3 行"}) + summarize_tools = GovDocTools(documents=task.documents, workspace=tmp_path) + body = await summarize_tools.read_document( + {"path": "evidence.md", "start_line": 1, "end_line": 5} + ) + assert "证据一" in body + + +def test_empty_registry_restricts_to_empty(tmp_path: Path): + assert ToolRegistry(()).restrict_to(()).names() == () + + +# --------------------------------------------------------------------------- +# 六、真实数据（数据目录不在就跳过） +# --------------------------------------------------------------------------- + +_HAS_REAL_DATA = DEFAULT_DATA_ROOT.is_dir() +_needs_real_data = pytest.mark.skipif( + not _HAS_REAL_DATA, reason=f"数据源不在这台机器上：{DEFAULT_DATA_ROOT}" +) + + +@_needs_real_data +def test_real_corpus_assembles_and_passes_the_gate(): + tasks, reports = build_audit_tasks(checkpoint_count=3) + assert len(tasks) == 3 + assert reports + document = tasks[0].documents[0] + assert document.logical_name == "tender.md" + assert document.line_count > 2000 + assert document.char_count > 100_000 + # 装配路径上已经调过校验函数，这里再自己确认一遍：这道闸是压测能不能启动的判据。 + assert_no_residue("\n".join(document.lines), where="真实语料") + assert_no_residue(tasks[0].checkpoint.render(), where="真实审核点") + replaced = sum(count for report in reports for count in report.counts.values()) + assert replaced > 0 + + +@_needs_real_data +def test_real_task_assembles_three_requests(tmp_path: Path): + tasks, _ = build_audit_tasks(checkpoint_count=1) + for phase in ("plan", "execute", "summarize"): + request = build_run_request( + task=tasks[0], + phase=phase, + run_id=make_run_id(task_index=0, phase=phase), + workspace=tmp_path, + model_binding={}, + ) + assert request.action_executor.registry == request.tools