From d4aa5e117ee7a87ebd42bba1bf831d380e093de6 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Mon, 10 Aug 2026 23:49:11 -0400 Subject: [PATCH] =?UTF-8?q?test(e2e):=20=E8=A1=A5=E4=B8=8A=E6=89=93?= =?UTF-8?q?=E7=9C=9F=E5=AE=9E=E7=BD=91=E5=85=B3=E7=9A=84=E9=82=A3=E4=B8=80?= =?UTF-8?q?=E5=B1=82=EF=BC=8C=E5=9B=9B=E5=B1=82=E6=B5=8B=E8=AF=95=E5=88=B0?= =?UTF-8?q?=E9=BD=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 两个用例：一次运行走通「模型→解释→工具→观察回填→再问模型→收尾」，以及取消能穿过真实的 HTTP 请求并留下已取消的结束记录。测试自带解释器、工具注册表与事件出口——库故意不带它们， 带了就等于替某一家定了动作语言。 断言只绑结构不变量，一条都不绑模型输出的文字：停止原因落在两个正常终态里、至少两步且至少 一步真的执行过动作、每步 call_id 非空（网关那边生成的，替身给不出来）、日志读回来与返回值 逐字段相等、事件条数等于步数。绑内容的测试会随机红，而随机红的测试很快没人看。 两道跳过闸：网关装没装，以及 POLYLOOP_E2E 是不是 1。分开是因为填好密钥不等于同意花钱。 实测跑一次三次真实模型调用、约 12 秒；开关关着时整层跳过；make ci 283 passed / 16 skipped。 Co-Authored-By: Claude Opus 5 (1M context) --- README.md | 7 +- tests/e2e/test_real_gateway_run.py | 421 +++++++++++++++++++++++++++++ 2 files changed, 425 insertions(+), 3 deletions(-) create mode 100644 tests/e2e/test_real_gateway_run.py diff --git a/README.md b/README.md index 8509618..d882a4d 100644 --- a/README.md +++ b/README.md @@ -39,9 +39,10 @@ 所以它排在架构前面：边界画错，后面每一份架构文档都要重写 - [x] ③ 架构 —— `research-wiki/explanation/architecture.md` 与 `pyproject.toml` 的 import-linter 契约。 架构文档先于代码存在，此期间它是一份规格而不是描述，文档开头须写明这一点 -- [x] ④ 测试框架 —— unit / integration / contract 三层已在跑；**e2e 还是空的**，它要打真实 - 模型网关（划分判据是「依赖什么」，见 [CLAUDE.md](CLAUDE.md) §1.9）。`tests/contract/` - 那套公共行为一致性用例接上了自带的存储实现，解释器与执行器那几条仍等下游把实现接进来 +- [x] ④ 测试框架 —— 四层都在跑（划分判据是「依赖什么」，见 [CLAUDE.md](CLAUDE.md) §1.9）。 + e2e 打真实模型网关、会产生真实费用，所以默认不跑：要 `POLYLOOP_E2E=1` 加显式 + `pytest -m e2e`，配置见 [.env.example](.env.example)。`tests/contract/` 那套公共行为 + 一致性用例接上了自带的存储实现，解释器与执行器那几条仍等下游把实现接进来 - [x] ⑤ 实现 —— 十个模块全部落地，五个接缝都有调用点。**一处已知欠账**：`stores` 只有逐行 追加那一种形态，关系数据库那种由下游自己实现，契约套件是它的准入标准 - [ ] ⑥ 迁移验收 —— 真的把 dissect 与 GovDoc-SaaS 迁过来，以两边测试全绿为准 diff --git a/tests/e2e/test_real_gateway_run.py b/tests/e2e/test_real_gateway_run.py new file mode 100644 index 0000000..e603a57 --- /dev/null +++ b/tests/e2e/test_real_gateway_run.py @@ -0,0 +1,421 @@ +"""打真实模型网关的那一层：一次运行真的从模型走到工具再走回来。 + +**这一层叫 e2e，因为它连的是真实模型网关**（`CLAUDE.md` §1.9 的分层判据是「依赖什么」）。 +它会产生真实的模型调用与真实的费用，所以它是唯一一层默认不跑的测试——`pyproject.toml` 的 +`addopts` 里有 `-m 'not e2e'`，`make ci` 因此跑不到这里。要跑它得显式写 `pytest -m e2e`。 + +**两道跳过闸，缺一不可。** 第一道是网关装没装（没装 `polyloop[gateway]` 就整份文件跳过）； +第二道是 `POLYLOOP_E2E` 这个开关等不等于 `"1"`。分成两件事是因为**填好密钥不等于同意花钱**： +只看密钥的话，任何人配好 `.env` 之后随手跑一次全套测试就会打出去一串真实调用并产生真实账单， +而他本来只是想看看测试过不过。开关的读法与网关一致——先读当前工作目录下的 `.env`，再让环境 +变量覆盖它，所以临时开一次不必改文件。 + +**这里自带三样真实实现：决策解释器、工具注册表、事件出口。** 库故意不带它们（带了就等于替 +某一家定了动作语言），而没有它们循环就走不起来。它们住在测试里，不是库的一部分。 + +**断言只绑结构不变量，一条都不绑模型输出的文字内容。** 模型是不确定的，绑内容的测试会随机 +红，而随机红的测试很快就会被所有人忽略，然后这一层就不再拦得住任何东西。 +""" + +import asyncio +import json +import os +import re +from collections.abc import Mapping + +import pytest + +polygateway = pytest.importorskip( + "polygateway", reason="没装 polyloop[gateway]，打真实网关这一层跳过" +) + +from dotenv import dotenv_values # noqa: E402 +from polygateway import GatewayClient, GatewaySettings # noqa: E402 + +from polyloop.adapters import GatewayModelClient # noqa: E402 +from polyloop.ports import ( # noqa: E402 + Action, + Event, + FinalAnswer, + InvalidDecision, + ModelCall, + ParsedReply, + ToolCall, +) +from polyloop.session import AgentDefinition, RunRequest, run # noqa: E402 +from polyloop.stores import JsonlRunStore # noqa: E402 +from polyloop.tools import ToolRegistry, ToolSpec # noqa: E402 +from polyloop.types import ( # noqa: E402 + ActionStatus, + Budget, + Context, + Message, + ModelReply, + ReplayPolicy, + Role, + StopReason, + SyntheticObservations, + TextBlock, +) + +pytestmark = pytest.mark.e2e + +#: 开关的读法与网关一致：`.env` 在下、环境变量在上。两份都读是因为密钥本来就在 `.env` 里， +#: 而临时开一次 e2e 不该逼人去改那个文件。 +_ENV = {**dotenv_values(".env"), **os.environ} + +if _ENV.get("POLYLOOP_E2E") != "1": + pytest.skip( + "POLYLOOP_E2E 不是 1：这一层会打真实模型网关并产生真实费用，默认不跑", + allow_module_level=True, + ) + + +# --------------------------------------------------------------------------- +# 动作协议：模型每一轮只输出一个 JSON 对象 +# --------------------------------------------------------------------------- + +#: 讲给模型听的那份协议。**写得极其明确并给一个例子**——模型只有照这个格式输出，循环才走得 +#: 下去。指望它猜的话，第一轮就会得到一段散文，然后这次运行以连续解析失败收尾，而那个红叉 +#: 看起来像是库坏了。 +_PROTOCOL = """你在一个自动循环里工作。每一轮你**只能输出一个 JSON 对象**，前后不许有任何别的文字、说明或标点。 + +要调用工具，输出： +{"tool": "工具名", "args": {"参数名": 参数值}} + +要给出最终回答，输出： +{"final": "你的回答"} + +可用的工具只有两个： +- add：把两个数相加。参数 a 和 b 都是数字。 +- submit：提交结果。参数 answer 是一个字符串。调用它就表示这次工作做完了。 + +例子——要算 3 加 4，你这一轮就输出： +{"tool": "add", "args": {"a": 3, "b": 4}} + +每一轮之后你会收到一条以「观察：」开头的消息，那是上一次工具调用返回的内容。""" + +_GOAL = "请先用 add 算出 17 加 25，拿到结果之后用 submit 把那个结果提交上去。" + +_OBSERVATION_TEMPLATE = "观察：{observation}" + +#: 代码围栏。模型很常把 JSON 包在 ```json ... ``` 里，剥不掉的话每一轮都会解析失败。 +_FENCE = re.compile(r"```[A-Za-z0-9_+-]*\n(?P.*?)```", re.DOTALL) + + +def _json_payload(text: str) -> str: + """把模型这一轮的输出削到只剩那个 JSON 对象。 + + 两步都是必要的：先剥围栏，再取最外层花括号之间的那一段。只剥围栏的话，模型在 JSON 前后 + 写一句「好的，我来算一下」就解析不了；只取花括号的话，围栏里带语言标签的那种输出会把 + ```json 一起吃进去。围栏没有闭合时第一步不匹配，第二步照样能把 JSON 捞出来。 + """ + body = text.strip() + fenced = _FENCE.search(body) + if fenced is not None: + body = fenced.group("body") + start = body.find("{") + end = body.rfind("}") + if start != -1 and end > start: + body = body[start : end + 1] + return body.strip() + + +class _JsonDecisionParser: + """按上面那份协议解释一次模型回复。 + + **对任何输入都返回 `ParsedReply`，绝不抛异常**（`design/0007` 决策三）。解释不出来是正常 + 路径的一部分——模型输出不合格式是每天都在发生的事，而抛异常会让库去替它编一个停止原因， + 于是「解释器有 bug」被伪装成「这次运行以某某原因结束」，然后进下游的统计。 + + 每一种失败给一条**对症**的说明，因为那段文本就是回喂给模型的观察。压成一句「格式错误」 + 的话，模型不知道自己错在哪，下一轮多半照错一遍。 + """ + + def parse(self, reply: ModelReply) -> ParsedReply: + payload = _json_payload(reply.content) + try: + decoded = json.loads(payload) + except json.JSONDecodeError as exc: + return self._invalid( + reply, + f"这一轮的输出不是一个 JSON 对象（{exc.msg}）。" + '只输出一个 JSON 对象，形如 {"tool": "add", "args": {"a": 1, "b": 2}}，前后不要有别的文字。', + ) + if not isinstance(decoded, dict): + return self._invalid( + reply, + f"解出来的是 {type(decoded).__name__} 而不是一个 JSON 对象。" + '只输出一个 JSON 对象，形如 {"tool": "add", "args": {"a": 1, "b": 2}}。', + ) + if "final" in decoded: + return ParsedReply( + history_text=reply.content, decision=FinalAnswer(text=str(decoded["final"])) + ) + if "tool" not in decoded: + return self._invalid( + reply, + '这个 JSON 对象里既没有 "tool" 也没有 "final"。调工具用 ' + '{"tool": ..., "args": {...}}，给最终回答用 {"final": "..."}。', + ) + name = decoded["tool"] + if not isinstance(name, str): + return self._invalid(reply, '"tool" 必须是一个字符串，也就是工具的名字。') + arguments = decoded.get("args", {}) + if not isinstance(arguments, dict): + return self._invalid( + reply, '"args" 必须是一个 JSON 对象，键是参数名，例如 {"a": 1, "b": 2}。' + ) + return ParsedReply( + history_text=reply.content, + decision=Action( + text=f"{name}({json.dumps(arguments, ensure_ascii=False, sort_keys=True)})", + tool_call=ToolCall(name=name, arguments=arguments), + ), + ) + + def parameters(self) -> Mapping[str, str]: + return {"kind": "json-tool-or-final"} + + @staticmethod + def _invalid(reply: ModelReply, explanation: str) -> ParsedReply: + return ParsedReply( + history_text=reply.content, decision=InvalidDecision(explanation=explanation) + ) + + +# --------------------------------------------------------------------------- +# 工具：纯计算，不碰文件系统、网络、子进程 +# --------------------------------------------------------------------------- + + +async def _add(arguments: Mapping[str, object]) -> str: + """两数相加。 + + 注册表已经按 schema 校验过类型，这里仍然自己判一次：校验的是 JSON Schema 的一个子集， + 而一个 `TypeError` 从这里抛出去会被判成「已执行」加一条正常观察，模型看不出该怎么改。 + """ + a, b = arguments.get("a"), arguments.get("b") + if not isinstance(a, int | float) or not isinstance(b, int | float): + return "a 和 b 都必须是数字。" + return str(a + b) + + +async def _submit(arguments: Mapping[str, object]) -> str: + return f"收到：{arguments.get('answer')}" + + +def _registry() -> ToolRegistry: + """本次运行可见的两个工具。 + + `submit` 带完成标记，这样运行有一条确定的收尾路径——没有它的话，这次运行只能靠模型自己 + 给最终回答或者撞上步数上限收尾，而那两条路一条不确定、一条要多花几次调用。 + """ + return ToolRegistry( + ( + ToolSpec( + name="add", + description="把两个数相加，返回它们的和。", + parameters={ + "type": "object", + "properties": {"a": {"type": "number"}, "b": {"type": "number"}}, + "required": ["a", "b"], + "additionalProperties": False, + }, + # 纯计算，重复算一次无害。 + replay_policy=ReplayPolicy.SAFE, + handler=_add, + ), + ToolSpec( + name="submit", + description="提交最终结果。调用它就表示这次工作做完了。", + parameters={ + "type": "object", + # 允许数字：模型很常把算出来的数原样填进来，只认字符串的话那次调用会被 + # 判成参数不合法，白花一次调用去纠正一个与本层无关的形式问题。 + "properties": {"answer": {"type": ["string", "number"]}}, + "required": ["answer"], + "additionalProperties": False, + }, + completes_run=True, + handler=_submit, + ), + ) + ) + + +# --------------------------------------------------------------------------- +# 事件出口 +# --------------------------------------------------------------------------- + + +class _RecordingEventSink: + """把收到的事件记进列表。一次运行一个实例，不跨运行复用。""" + + def __init__(self) -> None: + self.events: list[Event] = [] + + async def emit(self, event: Event) -> None: + self.events.append(event) + + def parameters(self) -> Mapping[str, str]: + return {"kind": "recording"} + + +# --------------------------------------------------------------------------- +# 装配 +# --------------------------------------------------------------------------- + + +@pytest.fixture +async def model_client(): + """一个连着真实网关的模型客户端。 + + 客户端持有连接池，用完必须 `aclose()`——不关的话每个用例漏一份连接池，而表现只是事件 + 循环关闭时的一串告警。 + """ + client = GatewayClient.from_env() + try: + yield GatewayModelClient(client=client, settings=GatewaySettings.from_env()) + finally: + await client.aclose() + + +def _text(role: Role, text: str) -> Message: + return Message(role=role, content=(TextBlock(text=text),)) + + +def _definition(model_client, store: JsonlRunStore, sink: _RecordingEventSink) -> AgentDefinition: + return AgentDefinition( + model_client=model_client, + decision_parser=_JsonDecisionParser(), + store=store, + event_sink=sink, + synthetic_observations=SyntheticObservations( + action_rejected="这次工具调用没有执行：工具名或参数不合法。改过之后重新输出一个 JSON 对象。", + env_failed="环境出错了，这次工具调用没有产生结果。", + model_call_failed="上一次模型调用失败了。", + ), + ) + + +def _request(run_id: str) -> RunRequest: + """一次运行的装配。 + + `max_steps` 取 3：这次运行正常走完是两步（算一次、提交一次），留一步的余量给模型偶尔多说 + 一轮。**上限压得这么低是为了控制费用**——这一层每跑一次都在花钱，而它要证明的事（链路通不通） + 两步就证明完了。 + """ + registry = _registry() + return RunRequest( + run_id=run_id, + budget=Budget( + max_steps=3, + max_actions=3, + max_consecutive_parse_failures=2, + max_prompt_chars=20_000, + ), + action_executor=registry.executor(), + tools=registry, + context=Context( + run_level=(_text(Role.SYSTEM, _PROTOCOL),), + goal_level=(_text(Role.USER, _GOAL),), + ), + injections={}, + model_binding={}, + model_replay_policy=ReplayPolicy.NEVER, + observation_template=_OBSERVATION_TEMPLATE, + cancel_grace_seconds=5.0, + ) + + +# --------------------------------------------------------------------------- +# 用例一：完整闭环 +# --------------------------------------------------------------------------- + +#: 这次运行走完之后可以落在的两个停止原因。 +#: +#: 两个都算正常终态，因为两者都意味着**循环自己走到了头**：`TASK_COMPLETED` 是模型调了带完成 +#: 标记的工具，`AGENT_FINISHED` 是模型算完之后直接给了最终回答。只写前一个的话，模型选了后 +#: 一条同样合法的路，这条测试就会红。预算耗尽、连续解析失败、模型调用失败都不在里面——那些 +#: 是循环没走通。 +_CLOSED_LOOP_STOP_REASONS = (StopReason.TASK_COMPLETED, StopReason.AGENT_FINISHED) + + +async def test_a_real_run_goes_from_model_through_a_tool_and_back(model_client, tmp_path) -> None: + """一次运行走通「模型 → 解释 → 工具执行 → 观察回填 → 再问模型 → 收尾」。""" + store = JsonlRunStore(directory=tmp_path) + sink = _RecordingEventSink() + request = _request("e2e-closed-loop") + + result = await run(_definition(model_client, store, sink), request) + + assert result.stop_reason in _CLOSED_LOOP_STOP_REASONS + # 只走一次模型调用不算闭环：那种运行证明的只是「请求发得出去」，证明不了观察回填之后模型 + # 还能接着往下走。 + assert len(result.steps) >= 2 + assert any(step.action_status is ActionStatus.EXECUTED for step in result.steps) + for step in result.steps: + # 调用标识是「真的打出去过」的硬证据：它由网关那边生成，替身给不出来。 + assert step.call_id, f"第 {step.step_idx} 步没有调用标识" + assert step.prompt_chars > 0, f"第 {step.step_idx} 步的提示词规模是 0" + assert step.step_wall_ms > 0, f"第 {step.step_idx} 步的墙钟是 0" + + # 往返等价：日志读回来的步序列与返回值逐字段相等。不等的话，下游拿轨迹做的分析和拿返回值 + # 做的分析会得出不同的结论，而两边都自称是这次运行。 + log = await store.read_log(request.run_id) + assert tuple(entry.step for entry in log.steps) == result.steps + assert log.finished is not None + assert log.finished.result.stop_reason is result.stop_reason + + assert len(sink.events) == len(result.steps) + + +# --------------------------------------------------------------------------- +# 用例二：取消穿透 +# --------------------------------------------------------------------------- + + +class _EntryAnnouncingClient: + """转发给真客户端，并在进入那次调用时打一个信号。 + + 取消要落在**真实的 HTTP 请求途中**才验得到东西。靠 `sleep` 猜时机的话，慢一点就落在解释 + 或工具执行上、快一点就落在调用发出之前，而两种落空都表现成一条绿的测试。这个信号把时机 + 收成确定的：它一亮，下一个 await 就是那次真实请求。 + """ + + def __init__(self, inner) -> None: + self._inner = inner + self.entered = asyncio.Event() + + async def call(self, call: ModelCall) -> ModelReply: + self.entered.set() + return await self._inner.call(call) + + def parameters(self) -> Mapping[str, str]: + return self._inner.parameters() + + +async def test_cancellation_passes_through_a_real_model_call(model_client, tmp_path) -> None: + """取消能穿过真实的网络调用，并且留下一条以「已取消」收尾的结束记录。 + + 结束记录是这条测试的另一半：没有它，恢复读到的是一次没有结束标记的运行，会被当成可以 + 续跑——而它其实是被人主动叫停的。 + """ + store = JsonlRunStore(directory=tmp_path) + sink = _RecordingEventSink() + request = _request("e2e-cancelled") + client = _EntryAnnouncingClient(model_client) + + task = asyncio.create_task(run(_definition(client, store, sink), request)) + await asyncio.wait_for(client.entered.wait(), timeout=30) + # 信号亮起时那次请求还没被 await。让出一小会儿，取消就确实落在请求飞在网上的那段。 + await asyncio.sleep(0.5) + task.cancel() + + with pytest.raises(asyncio.CancelledError): + await task + + log = await store.read_log(request.run_id) + assert log.finished is not None, "取消之后没有结束记录，这次运行看起来还能续跑" + assert log.finished.result.stop_reason is StopReason.CANCELLED