diff --git a/src/polyloop/_assembly/__init__.py b/src/polyloop/_assembly/__init__.py index 731d0b8..6278d58 100644 --- a/src/polyloop/_assembly/__init__.py +++ b/src/polyloop/_assembly/__init__.py @@ -86,13 +86,25 @@ def injection_messages(injections: Mapping[str, tuple[Injection, ...]]) -> tuple ) -def injected_entry_ids(injections: Mapping[str, tuple[Injection, ...]]) -> tuple[str, ...]: - """这次贴了哪几条，按与 `injection_messages` 相同的顺序。 +def injected_entry_ids( + injections: Mapping[str, tuple[Injection, ...]], +) -> Mapping[str, tuple[str, ...]]: + """这次每个通道贴了哪几条，通道之间按通道名字典序，通道内与 `injection_messages` 同序。 - **正文不进快照也不进轨迹，只有条目标识进。** 注入内容可能很大，进快照会让快照变成一份 - 数据副本；而「这次贴了哪几条」事后要查得到。 + **正文不进参数快照，只有条目标识进。** 注入内容可能很大，进快照会让快照变成一份数据 + 副本；而「这次贴了哪几条」事后要查得到。它进的是运行开始记录里的参数快照，不是轨迹—— + 轨迹是步记录的序列，里面没有这一列。 + + **通道维度保留，不拍平成一个扁平序列。** 拍平之后「声明了这个通道但一条都没选中」和 + 「压根没有这个通道」得出同一个结果，而有下游要比较的正是这两种情形（`0015` 决策二）。 + + 顺序必须与 `injection_messages` 一致，两个函数同住一个模块就是为了守住这一点：顺序对不上 + 的话，快照记的贴入顺序和模型真正看到的顺序是两回事，而续跑守卫照样全绿。 """ - return tuple(entry.entry_id for channel in sorted(injections) for entry in injections[channel]) + return { + channel: tuple(entry.entry_id for entry in injections[channel]) + for channel in sorted(injections) + } def history_messages(steps: Sequence[StepRecord], observation_template: str) -> tuple[Message, ...]: diff --git a/src/polyloop/ports/__init__.py b/src/polyloop/ports/__init__.py index a6379a5..06ce7d9 100644 --- a/src/polyloop/ports/__init__.py +++ b/src/polyloop/ports/__init__.py @@ -1,6 +1,6 @@ """五个 Protocol，以及只在一次调用往返之间存在的入参 / 返回壳。 -读者是写适配器的人和 `tests/contract/`。用 `typing.Protocol` 而不是抽象基类：下游的对象 +读者是写适配器的人和 `polyloop.testing`。用 `typing.Protocol` 而不是抽象基类：下游的对象 往往已经是它自己的类、还要同时满足项目自己更宽的接口，只有结构化子类型能让同一个对象同时 满足库的窄视图和项目的宽视图。 @@ -14,7 +14,7 @@ `research-wiki/design/0006-public-names-and-signatures.md` 决策四。 每个方法的行为契约在 `research-wiki/design/0007-seam-behaviour.md`，机器形式在 -`tests/contract/`。这里的 docstring 只写「读这段代码的人不知道就会写错什么」。 +`polyloop.testing` 那套契约套件里。这里的 docstring 只写「读这段代码的人不知道就会写错什么」。 """ from collections.abc import Mapping @@ -211,6 +211,25 @@ class ActionExecutor(Protocol): **动作本身报错算「已执行」**，不算环境故障：代码抛异常、命令返回非零都是正常观察，要 原样回喂让模型自己纠正。判成环境故障会让一次运行在模型本来能自我纠正的地方直接终止， 而轨迹上看不出它本可以继续。分界线是环境还能不能接着服务。 + + **环境自己坏了走返回值，不走异常**：连不上、协议不对、开好的会话没了，返回 + `ActionStatus.ENV_ERROR`。把可预期的环境异常翻译成这个状态是适配器的正常工作，不是 + catch-all——一个 HTTP 客户端的连接超时、一个容器会话的「会话已关闭」都是有名有姓的异常 + 类型，捕获它们并返回 `ENV_ERROR` 是在履行契约。 + + **真抛出来的异常库不捕获，原样穿出 `run()` 与 `resume()`。** 库替这一步编一个结算结果 + 就是在编造：异常抛出时副作用发生没发生是未知的，而 `StepCompleted` 要求有动作结果就得 + 连观察、完成信号、截断字符数一起造齐，而一条带着动作结果的完整步记录会被恢复读成「上一 + 步走完了」，那个未知状态就此被抹掉。不写反而是照实——日志停在「动作意图有、步记录无」， + 恢复照四态表把它读成状态未知。这也是为什么这个接缝和 `DecisionParser` 一样，抛出的异常 + 不会被伪装成一个停止原因送进下游的统计。 + + 模型调用接缝的异常反倒由库捕获，这个不对称的来由与被否掉的「库转成 `ENV_ERROR`」那条路 + 在 `research-wiki/design/0016-action-executor-failure.md`。 + + **`asyncio.CancelledError` 必须原样穿过，不许捕获吞没**（`CLAUDE.md` §1.6），in-flight + 资源在 `finally` 里放掉。上一条读快了容易读成「异常一概不用管」，而取消恰恰是那个必须 + 管的异常，管的方式是让它穿过去。 """ async def execute(self, action: Action) -> ActionOutcome: ... diff --git a/src/polyloop/session/__init__.py b/src/polyloop/session/__init__.py index c5fdeb3..a5e20cf 100644 --- a/src/polyloop/session/__init__.py +++ b/src/polyloop/session/__init__.py @@ -18,7 +18,7 @@ import json import logging import time from collections.abc import Mapping -from dataclasses import dataclass +from dataclasses import dataclass, field from polyloop._assembly import ( assemble, @@ -48,7 +48,7 @@ from polyloop.ports import ( RunLog, RunStore, ) -from polyloop.tools import RegistryExecutor, ToolRegistry +from polyloop.tools import RegistryExecutor, ToolRegistry, _frozen from polyloop.types import ( ActionOutcome, ActionStatus, @@ -90,6 +90,29 @@ class ParameterDriftError(Exception): """ +def _checked_str_mapping(mapping: Mapping[str, str], owner: str) -> Mapping[str, str]: + """快照的每一个键和每一个值都必须是字符串，不是就当场拒绝。 + + 取值类型已经是持久化契约的一部分：反序列化那一侧读到非字符串直接失败。只靠那一侧的话， + 失败发生在续跑读日志的时候——这次运行已经完整跑过一遍、钱花完了、日志也写下去了，才发现 + 里面有一项读不回来；而且发现它的前提是真的有人来续跑，没人续跑那份存坏了的日志就一直躺着 + 直到有人拿它做统计。 + + **用异常不用 `assert`**：`python -O` 会把断言整条移除，而这道校验守的正是一件静默出错的事。 + + `owner` 进错误信息，因为快照汇的是五个接缝加两个请求字段；只说「快照必须是字符串」的报错 + 在七个来源里指不出是谁。 + """ + for key, value in mapping.items(): + if not isinstance(key, str): + raise ValueError(f"{owner} 的键必须是字符串，收到 {type(key).__name__}：{key!r}") + if not isinstance(value, str): + raise ValueError( + f"{owner} 里 {key!r} 的取值必须是字符串，收到 {type(value).__name__}：{value!r}" + ) + return mapping + + @dataclass(frozen=True, slots=True, kw_only=True) class AgentDefinition: """跨运行不变的那一半装配，可以并发复用。 @@ -114,6 +137,11 @@ class AgentDefinition: 方法则每次现问，快照永远是从真实对象上读出来的**事实**而不是一份**声明**。 键带接缝名前缀，免得「哪一侧报的这个键」要靠约定记住。 + + **接缝上报的键值在这里校验类型，构造定义的时候不校验。** 构造时不向任何接缝发问， + 发问发生在首次算快照的时候，那时 `run()` 已经读过一次存储日志了。所以这道校验保证的 + 不是零代价，是它发生在写运行开始记录之前，也就是在任何一次模型调用之前——不会跑完 + 一整次运行、把钱花光，才在续跑时发现快照里有一项存不下去。 """ snapshot: dict[str, str] = {} for prefix, seam in ( @@ -122,14 +150,29 @@ class AgentDefinition: ("store", self.store), ("event_sink", self.event_sink), ): - for key, value in seam.parameters().items(): + for key, value in _checked_str_mapping( + seam.parameters(), f"{prefix}.parameters()" + ).items(): snapshot[f"{prefix}.{key}"] = value return snapshot @dataclass(frozen=True, slots=True, kw_only=True) class RunRequest: - """一次运行独有的那一半装配。构造廉价：无 I/O、无网络校验、无哈希计算。""" + """一次运行独有的那一半装配。构造廉价：无 I/O、无网络校验、无哈希计算。 + + 三个映射字段（`injections`、`model_binding`、`fingerprints`）在构造时被逐层冻成只读的 + 形状存下来，用的是 `ToolSpec.parameters` 那同一个函数。`frozen=True` 只挡住「把字段重新 + 绑到另一个对象上」，挡不住「原地改那个字段里的 dict」，而这三个字段全都进参数快照：不冻 + 的话，`__post_init__` 那道「键和值都得是字符串」的校验可以被构造完之后往 dict 里塞一个 + 整数绕过去，那个整数一路进到运行开始记录，要到续跑读日志反序列化时才炸——那时这次运行 + 已经完整跑过一遍、钱也花完了。`injections` 更凶一档：构造后往里加一个通道，会同时改掉 + 参数快照和装配出来的消息序列，而两者都是「这次运行是什么设置」的证据。 + + **三个字段的类型注解仍然是 `Mapping`，签名的形状没有变。** `Mapping` 本来就是只读接口， + 冻结没有对下游多要求什么：照旧传普通 dict 进来，变的只是「传进来之后再改那个 dict，这个 + 请求不跟着变」。 + """ #: 不透明字符串，库不解析。它同时是日志的主键。 run_id: str @@ -140,10 +183,27 @@ class RunRequest: #: 项目传一个空注册表加一个环境句柄，注册了工具的项目传 `tools.executor()`。 tools: ToolRegistry context: Context - #: 本次要贴进上下文的条目，按通道分组。 + #: 本次要贴进上下文的条目，按通道分组。构造时冻成只读映射（见类 docstring）。 injections: Mapping[str, tuple[Injection, ...]] #: 项目自己的标识，库不解释，原样透传给每次模型调用。它的全部键值都进参数快照。 + #: 构造时冻成只读映射（见类 docstring）。 model_binding: Mapping[str, str] + #: 这次运行用的材料是哪一版：提示词模板的哈希、技能库的版本这类。键名由项目自己定，库不 + #: 解释内容，全部键值以 `request.fingerprint.` 进参数快照，**不透传给模型调用**。 + #: + #: **和 `model_binding` 的分界是「坐标还是配方版本」**：那个记的是这次运行属于哪一格 + #: （哪个账本、第几轮、哪道题），这个记的是这次用的材料是哪一版。混在一个字段里事后分不 + #: 开——一组键值里既有「第 3 轮」又有一个 sha，要靠键名的命名约定去猜哪个是哪个，而命名 + #: 约定不在任何一处被断言。完整论证在 + #: `research-wiki/design/0015-parameter-snapshot-contract.md` 决策一。 + #: + #: 建议值里带上算法前缀，形如 `sha256:`。不带的话，换算法那天旧记录和新记录会以 + #: 「两个不同的十六进制串」的形式参与比对，报出来的漂移看不出是换了算法还是内容真的变了。 + #: 这是建议不是校验：库不解释这个值，也就没有立场规定下游能用哪几种哈希。 + #: + #: 构造时冻成只读映射（见类 docstring）。默认值那个空 dict 同样会被冻——不冻的话，一个 + #: 没传指纹的请求手上是一个改得动的空 dict，往里塞什么都不经过校验。 + fingerprints: Mapping[str, str] = field(default_factory=dict) #: 必填无默认。工具的重放策略能从注册表查到，模型调用的查不到——只有调用方知道这次调用 #: 能不能重来。 model_replay_policy: ReplayPolicy @@ -155,6 +215,16 @@ class RunRequest: def __post_init__(self) -> None: check_observation_template(self.observation_template) + # 这两个字段整个进快照，取值不是字符串的话这次运行照常跑完，续跑读日志时才炸。这里 + # 拒绝的代价是零：什么都还没发生，没有 I/O、没有日志、没有模型调用。两个一起校验是 + # 有意的——它们语义同族、形状相同，只校验其中一个会让两个看起来一样的字段行为不一样。 + _checked_str_mapping(self.model_binding, "RunRequest.model_binding") + _checked_str_mapping(self.fingerprints, "RunRequest.fingerprints") + # 校验完当场冻起来。校验和冻结是一对，缺了后半截前半截只在构造那一瞬间成立： + # `frozen=True` 拦不住 `request.fingerprints["x"] = 7`，而那之后快照里就有一个整数了。 + object.__setattr__(self, "model_binding", _frozen(dict(self.model_binding))) + object.__setattr__(self, "fingerprints", _frozen(dict(self.fingerprints))) + object.__setattr__(self, "injections", _frozen(dict(self.injections))) if self.cancel_grace_seconds < 0: raise ValueError(f"取消宽限期不能为负：{self.cancel_grace_seconds}") # 模型看见的 schema 来自一个注册表、实际分发走另一个，表现是「模型调了一个它看得见的 @@ -173,8 +243,9 @@ class RunRequest: """请求这一侧的快照，加上向动作执行接缝问的那一次。 **上下文与注入内容不进快照。** 它们是这次运行的输入数据不是参数，进快照会让快照变成 - 一份数据副本，而它们可能很大。注入的**条目标识**另行进轨迹，所以「这次贴了哪几条」 - 事后查得到，查不到的只是正文。 + 一份数据副本，而它们可能很大。注入的**条目标识**进快照，所以「这次贴了哪几条」事后 + 查得到，查不到的只是正文。生成上下文的东西（提示词模板这类）不是数据是参数，它的版本 + 走 `fingerprints`。 """ snapshot: dict[str, str] = { "request.max_steps": str(self.budget.max_steps), @@ -187,14 +258,24 @@ class RunRequest: "request.observation_template": self.observation_template, "request.cancel_grace_seconds": str(self.cancel_grace_seconds), "request.tools": ",".join(self.tools.names()), - "request.injected_entry_ids": ",".join(injected_entry_ids(self.injections)), } + # 一个通道一个键，不把所有通道拍平成一个。拍平之后「声明了这个通道但一条都没选中」 + # 与「压根没有这个通道」是同一个结果，而有下游要比较的正是这两种情形：前者是一个值为 + # 空串的键，后者是这个键不存在。 + for channel, entry_ids in injected_entry_ids(self.injections).items(): + snapshot[f"request.injected_entry_ids.{channel}"] = ",".join(entry_ids) # 模型绑定必须进快照，否则给它选字符串映射的那条理由就落空了。失败场景很具体：崩溃后 # 用同一个运行标识、换一组绑定续跑，后面每一次调用被记到另一套坐标上，而两段轨迹在 # 文件里看起来是同一次运行。 for key, value in self.model_binding.items(): snapshot[f"request.binding.{key}"] = value - for key, value in self.action_executor.parameters().items(): + # 一条指纹都没有时一个键都不写，不写一个值为空串的键：今天已经在跑的配置算出来的快照 + # 因此逐字节不变，只有真的传了指纹的运行才多出这几项。 + for key, value in self.fingerprints.items(): + snapshot[f"request.fingerprint.{key}"] = value + for key, value in _checked_str_mapping( + self.action_executor.parameters(), "action_executor.parameters()" + ).items(): snapshot[f"action_executor.{key}"] = value return snapshot @@ -676,6 +757,7 @@ class _Driver: replay_policy=ReplayPolicy.NEVER if spec is None else spec.replay_policy, ) ) + # 这里不包 try 是契约，不是漏了：执行器抛出的异常原样穿出去（`design/0016` 决策一）。 outcome = await self._request.action_executor.execute(action) return outcome, bool(spec is not None and spec.completes_run) diff --git a/src/polyloop/tools/__init__.py b/src/polyloop/tools/__init__.py index 62459c3..c347df4 100644 --- a/src/polyloop/tools/__init__.py +++ b/src/polyloop/tools/__init__.py @@ -69,7 +69,10 @@ class ToolHandler(Protocol): def _frozen(value: object) -> object: - """把一份 JSON Schema 逐层变成改不动的形状：映射变只读视图，列表变元组。 + """把一份嵌套结构逐层变成改不动的形状：映射变只读视图，列表变元组。 + + 工具的参数 schema 是第一个用它的地方，`polyloop.session` 的请求也用它冻自己那几个映射 + 字段——两处要防的是同一件事，所以共用这一个函数而不是各写一份。 只冻最外面一层不够。真正会发生的改法是从注册表里把规格取出来、往里伸一层去改 （`spec_for("read").parameters["properties"]["path"]["type"] = ...`），那一下同时改掉了 diff --git a/src/polyloop/types/__init__.py b/src/polyloop/types/__init__.py index 53d3e65..cf56bd3 100644 --- a/src/polyloop/types/__init__.py +++ b/src/polyloop/types/__init__.py @@ -88,8 +88,11 @@ class Injection: 库只负责贴和记录贴了什么，不负责生成、评测、挑选。 """ - #: 这条条目的标识，原样进轨迹。正文不进快照，只有它进——所以「这次贴了哪几条」事后 - #: 查得到，查不到的只是正文。 + #: 这条条目的标识，原样进运行开始记录里的参数快照。正文不进快照，只有它进——所以 + #: 「这次贴了哪几条」事后查得到，查不到的只是正文。**它不进轨迹**：轨迹是步记录的 + #: 序列，里面没有这一列，照着「进轨迹」去找会在步记录里翻一个不存在的东西。 + #: **标识里不要放逗号。** 快照把一个通道里的标识拼成逗号串，标识里有逗号的话两组不同的 + #: 注入可能拼出同一个串，于是一次该报的漂移没报（`design/0015` 决策二）。 entry_id: str content: str @@ -348,7 +351,7 @@ class Intent: **两种意图合成一个类型，用 `kind` 区分**，而不是两个类。它们字段完全相同，分成两个类 之后恢复逻辑要把同一段「有意图没结果」的判定写两遍，而那段判定是高危代码——同一个判断 写在两处，改的时候必然有一处漏掉。代价是类型检查器不再帮忙区分两种意图，这个补在 - `tests/contract/` 里。 + `polyloop.testing` 那套契约套件里。 """ run_id: str diff --git a/tests/unit/test_assembly.py b/tests/unit/test_assembly.py index 3c916f9..9165850 100644 --- a/tests/unit/test_assembly.py +++ b/tests/unit/test_assembly.py @@ -154,14 +154,45 @@ def test_channels_are_ordered_by_name_not_by_mapping_order() -> None: assert [block.text for m in injection_messages(entries) for block in m.content] == ["A", "Z"] -def test_entry_ids_come_back_in_the_same_order_as_the_messages() -> None: - """轨迹里记的是条目标识，正文不进——正文可能很大，而「这次贴了哪几条」事后要查得到。""" +def test_entry_ids_keep_the_channel_they_came_from() -> None: + """参数快照里记的是条目标识，正文不进——正文可能很大，而「这次贴了哪几条」事后要查得到。 + + 通道维度保留：拍平之后「声明了这个通道但一条都没选中」和「压根没有这个通道」是同一个 + 结果，而有下游要比较的正是这两种情形。 + """ entries = { "zeta": (Injection(entry_id="z", content="Z"),), - "alpha": (Injection(entry_id="a", content="A"),), + "alpha": (Injection(entry_id="a", content="A"), Injection(entry_id="b", content="B")), } - assert injected_entry_ids(entries) == ("a", "z") + assert injected_entry_ids(entries) == {"alpha": ("a", "b"), "zeta": ("z",)} + + +def test_a_declared_but_empty_channel_is_not_the_same_as_a_missing_one() -> None: + """一个是值为空元组的项，另一个是这个项不存在。两者拼进快照才分得开。""" + assert injected_entry_ids({"skill": ()}) == {"skill": ()} + assert injected_entry_ids({}) == {} + + +def test_entry_ids_come_back_in_the_same_order_as_the_messages() -> None: + """两个函数的顺序必须一致，它们同住一个模块就是为了守住这一点。 + + 对不上的话，快照记的贴入顺序和模型真正看到的顺序是两回事，而续跑守卫照样全绿。 + """ + entries = { + "zeta": (Injection(entry_id="z", content="Z"),), + "alpha": (Injection(entry_id="a", content="A"), Injection(entry_id="b", content="B")), + } + by_content = {"A": "a", "B": "b", "Z": "z"} + + flattened = [ + entry_id for entry_ids in injected_entry_ids(entries).values() for entry_id in entry_ids + ] + from_messages = [ + by_content[block.text] for m in injection_messages(entries) for block in m.content + ] + + assert flattened == from_messages # --------------------------------------------------------------------------- diff --git a/tests/unit/test_session.py b/tests/unit/test_session.py index 48078ea..2ebb482 100644 --- a/tests/unit/test_session.py +++ b/tests/unit/test_session.py @@ -661,7 +661,204 @@ def test_the_context_never_enters_the_snapshot() -> None: snapshot = request.parameter_snapshot() assert "很长的一段正文" not in "".join(snapshot.values()) - assert snapshot["request.injected_entry_ids"] == "e1" + assert snapshot["request.injected_entry_ids.skill"] == "e1" + + +def test_a_declared_but_empty_channel_still_gets_a_key() -> None: + """声明了通道却一条都没选中，是一个值为空串的键。 + + 有下游要比较「声明了通道而选中零条」和「连通道都不声明」这两组运行，拍平之后两者在记录 + 里长得一样，那一档就测不了。 + """ + request = dataclasses.replace(_request(FakeExecutor([])), injections={"skill": ()}) + + assert request.parameter_snapshot()["request.injected_entry_ids.skill"] == "" + + +def test_a_channel_that_was_never_declared_has_no_key_at_all() -> None: + request = dataclasses.replace(_request(FakeExecutor([])), injections={}) + + snapshot = request.parameter_snapshot() + + assert not [key for key in snapshot if key.startswith("request.injected_entry_ids")] + + +def test_two_channels_do_not_get_merged_into_one_key() -> None: + """通道名在拍平的写法里只用来定顺序，排完就没了，「哪几条来自哪个通道」事后查不到。""" + request = dataclasses.replace( + _request(FakeExecutor([])), + injections={ + "skill": ( + Injection(entry_id="s1", content="甲"), + Injection(entry_id="s2", content="乙"), + ), + "memo": (Injection(entry_id="m1", content="丙"),), + }, + ) + + snapshot = request.parameter_snapshot() + + assert snapshot["request.injected_entry_ids.skill"] == "s1,s2" + assert snapshot["request.injected_entry_ids.memo"] == "m1" + + +def test_no_fingerprints_means_no_fingerprint_keys() -> None: + """默认空映射时一个键都不写，不是写一个值为空串的键。 + + 今天已经在跑的配置算出来的快照因此逐字节不变，只有真的传了指纹的运行才多出那几项。 + """ + snapshot = _request(FakeExecutor([])).parameter_snapshot() + + assert not [key for key in snapshot if key.startswith("request.fingerprint.")] + + +def test_every_fingerprint_enters_the_snapshot() -> None: + """提示词模板不是数据是参数：它是一份跨运行复用的配方，每次运行拿它渲染出这一次的上下文。 + + 不记的话，「这次用的是哪一版提示词」就只剩下调用方的 git 提交这一个粒度，而同一个提交下 + 完全可以试好几份不同的模板。 + """ + request = dataclasses.replace( + _request(FakeExecutor([])), + fingerprints={"prompt_template": "sha256:abc", "skill_pack": "sha256:def"}, + ) + + snapshot = request.parameter_snapshot() + + assert snapshot["request.fingerprint.prompt_template"] == "sha256:abc" + assert snapshot["request.fingerprint.skill_pack"] == "sha256:def" + + +def test_a_non_string_binding_value_is_refused_at_construction() -> None: + """快照的取值类型已经是持久化契约的一部分，只靠反序列化那一侧的话，这次运行会照常跑完、 + 钱花光，才在续跑读日志时发现有一项读不回来。""" + with pytest.raises(ValueError, match="RunRequest.model_binding"): + dataclasses.replace(_request(FakeExecutor([])), model_binding={"round": 3}) + + +def test_a_non_string_fingerprint_is_refused_at_construction() -> None: + """`model_binding` 与 `fingerprints` 语义同族、形状相同，两个字段的行为必须一样——不一样 + 比两个都不校验更难查，查的人会先怀疑自己传错了字段。""" + with pytest.raises(ValueError, match="RunRequest.fingerprints"): + dataclasses.replace(_request(FakeExecutor([])), fingerprints={"template": 7}) + + with pytest.raises(ValueError, match="RunRequest.fingerprints"): + dataclasses.replace(_request(FakeExecutor([])), fingerprints={7: "sha256:abc"}) + + +@pytest.mark.parametrize( + ("field_name", "added_value"), + [("model_binding", "a"), ("fingerprints", "sha256:abc"), ("injections", ())], +) +def test_a_mapping_field_cannot_be_changed_after_construction( + field_name: str, added_value: object +) -> None: + """构造期那道校验拦不住构造之后原地改，所以三个映射字段构造时就被冻成只读的。 + + `frozen=True` 只挡住「把字段重新绑到另一个对象上」。往 `fingerprints` 里塞一个整数，它会 + 一路进到运行开始记录的参数快照，要到续跑读日志反序列化那一关才炸——而那时这次运行已经完整 + 跑过一遍、钱也花完了。`injections` 那一份还会同时改掉装配出来的消息序列。 + + `fingerprints` 这一档同时守住默认值：不传它的请求手上是一个 `default_factory` 造的空 + dict，不冻的话那个 dict 改得动，而往里塞什么都不经过校验。 + """ + request = _request(FakeExecutor([])) + + with pytest.raises(TypeError): + getattr(request, field_name)["新加的"] = added_value + + +def test_changing_the_dict_that_was_passed_in_does_not_change_the_request() -> None: + """调用方传进来的那个 dict 之后再被改，请求看见的仍是构造那一刻的形状。 + + 另一个方向：不拷一份的话，一个被复用的绑定 dict 改一个键，就悄悄改掉了一个已经在跑的运行 + 的参数快照。 + """ + binding = {"item": "a"} + request = dataclasses.replace(_request(FakeExecutor([])), model_binding=binding) + + binding["item"] = "b" + + assert request.parameter_snapshot()["request.binding.item"] == "a" + + +def test_a_request_derived_with_replace_still_constructs() -> None: + """`dataclasses.replace` 是推荐的派生范式，它会把已经冻过的那个只读映射再传进一次构造。 + + 冻结那一步不能因此炸——炸的话，派生一个请求就得先把三个映射字段各拆回普通 dict，而那正是 + 大多数调用方不会想到要做的一步。派生出来的那个照样是冻的。 + """ + request = dataclasses.replace( + _request(FakeExecutor([])), fingerprints={"prompt_template": "sha256:abc"} + ) + + derived = dataclasses.replace(request, run_id="run-2") + + assert derived.run_id == "run-2" + assert derived.parameter_snapshot()["request.fingerprint.prompt_template"] == "sha256:abc" + with pytest.raises(TypeError): + derived.fingerprints["prompt_template"] = "sha256:def" + + +async def test_the_frozen_binding_still_reaches_the_model_call_and_the_event() -> None: + """绑定冻成只读映射之后，模型调用与事件上那两份照样读得出来、内容一致。 + + 只读是这三个字段的全部改动：透传路径没有变，下游拿它查键、比对、序列化都和以前一样， + 只是拿它反过来改这次运行的配置不再行得通。 + """ + store = FakeStore() + sink = FakeSink() + model = FakeModel([_reply("go")]) + definition = _definition(store, model, FakeParser({"go": ACT}), sink) + + await run(definition, _request(FakeExecutor([_outcome(completed=True)]), binding={"item": "a"})) + + (event,) = sink.events + assert model.calls[0].binding == {"item": "a"} + assert event.model_binding == {"item": "a"} + with pytest.raises(TypeError): + model.calls[0].binding["item"] = "b" # type: ignore[index] + + +class _BadParameters: + """把一个接缝原样代理出去，只把 `parameters()` 换成一份取值不是字符串的映射。""" + + def __init__(self, seam: object, parameters: Mapping[object, object]) -> None: + self._seam = seam + self._parameters = parameters + + def __getattr__(self, name: str) -> object: + return getattr(self._seam, name) + + def parameters(self) -> Mapping[object, object]: + return self._parameters + + +@pytest.mark.parametrize("seam", ["model_client", "decision_parser", "store", "event_sink"]) +def test_a_definition_seam_reporting_a_non_string_is_refused(seam: str) -> None: + """接缝实现由下游写，它返回什么算外部输入。五个接缝里任何一个返回一个整数，症状都一样： + 这次运行照常跑完，续跑时才炸。 + + 报错要指得出是哪个接缝——只说「快照必须是字符串」的话，七个来源里找不出是谁。 + """ + definition = _definition(FakeStore(), FakeModel([]), FakeParser({})) + broken = dataclasses.replace( + definition, **{seam: _BadParameters(getattr(definition, seam), {"oops": 1})} + ) + + with pytest.raises(ValueError, match=rf"{seam}\.parameters"): + broken.parameter_snapshot() + + +def test_the_action_executor_reporting_a_non_string_is_refused() -> None: + """第五个接缝挂在请求上，聚合发生在另一处，所以它要单独验一遍。""" + request = dataclasses.replace( + _request(FakeExecutor([])), + action_executor=_BadParameters(FakeExecutor([]), {"session": 42}), # type: ignore[arg-type] + ) + + with pytest.raises(ValueError, match=r"action_executor\.parameters"): + request.parameter_snapshot() # --------------------------------------------------------------------------- @@ -699,6 +896,25 @@ async def test_resume_refuses_when_the_assembly_drifted() -> None: await resume(definition, _request(FakeExecutor([_outcome()]), budget=other_budget)) +async def test_resume_refuses_when_only_a_fingerprint_changed() -> None: + """换一份模板、用同一个运行标识续跑，前几步用 A、后几步用 B，全程零报错，那次运行的数据 + 已经废了却没有任何东西提示。守住这个失败场景是 `fingerprints` 存在的全部意义。 + + 别的东西一个字都没变：预算、绑定、接缝、注入都一样，只有指纹换了。 + """ + store = FakeStore() + definition = _definition(store, FakeModel([_reply("go")]), FakeParser({"go": ACT})) + first = dataclasses.replace( + _request(FakeExecutor([_outcome(completed=True)])), + fingerprints={"prompt_template": "sha256:aaa"}, + ) + await run(definition, first) + + second = dataclasses.replace(first, fingerprints={"prompt_template": "sha256:bbb"}) + with pytest.raises(ParameterDriftError, match="request.fingerprint.prompt_template"): + await resume(definition, second) + + async def test_resume_hands_back_a_finished_run_without_rerunning_it() -> None: store = FakeStore() model = FakeModel([_reply("go")]) @@ -788,6 +1004,81 @@ async def test_cancellation_propagates_and_still_writes_the_finished_marker() -> assert finished[0].result.stop_reason is StopReason.CANCELLED # type: ignore[attr-defined] +# --------------------------------------------------------------------------- +# 动作执行接缝抛异常 +# --------------------------------------------------------------------------- + + +class _RaisingExecutor(FakeExecutor): + """执行动作时抛出，而不是返回一个填好状态的结果。""" + + def __init__(self, error: BaseException) -> None: + super().__init__([]) + self._error = error + + async def execute(self, action: Action) -> ActionOutcome: + self.actions.append(action) + raise self._error + + +async def test_an_executor_exception_comes_straight_out_of_run() -> None: + """库不接管，异常原样穿出去，不被转成任何停止原因。 + + 接住就得给这一步编一个动作结果——状态、观察、完成信号、截断字符数四个字段全是库现造的， + 没有一个来自执行器。而一条带着动作结果的完整步记录会被恢复读成「上一步走完了」，那个未知 + 状态就被抹掉了。转成 `ENV_ERROR` 还会把实现方的 bug 伪装成环境故障送进下游的统计。 + """ + store = FakeStore() + definition = _definition(store, FakeModel([_reply("go")]), FakeParser({"go": ACT})) + executor = _RaisingExecutor(AttributeError("包装类自己的 bug")) + + with pytest.raises(AttributeError, match="包装类自己的 bug"): + await run(definition, _request(executor)) + + assert executor.actions != [] + # 日志停在「动作意图有、步记录无」，也没有结束记录：不写恰好是照实。 + assert [entry.kind for entry in store.of_type(Intent)] == [ # type: ignore[attr-defined] + IntentKind.MODEL_CALL, + IntentKind.ACTION, + ] + assert store.of_type(StepCompleted) == [] + assert store.of_type(RunFinished) == [] + + +async def test_resume_after_an_executor_exception_reads_the_state_as_unknown() -> None: + """副作用发生了没有，库无从知道，日志里也没有任何一处记着。 + + 这和进程崩在动作执行中途是同一种状态，恢复照四态表把它读成未知；那条动作意图记着「绝不 + 重放」，于是这次运行以「续跑时状态未知」收尾，而不是重新执行一次动作。 + """ + store = FakeStore() + definition = _definition(store, FakeModel([_reply("go")]), FakeParser({"go": ACT})) + request = _request(_RaisingExecutor(AttributeError("包装类自己的 bug"))) + with pytest.raises(AttributeError): + await run(definition, request) + + result = await resume(definition, request) + + assert result.stop_reason is StopReason.RESUME_STATE_UNKNOWN + assert result.steps == () + + +async def test_cancellation_from_the_executor_still_writes_the_finished_marker() -> None: + """取消要能穿过动作执行，`CancelledError` 原样重抛，但结束标记照常写下去。 + + 不写的话，恢复读到的是一次没有结束标记的运行，会被当成可以续跑，而它其实是被人主动叫停的。 + """ + store = FakeStore() + definition = _definition(store, FakeModel([_reply("go")]), FakeParser({"go": ACT})) + + with pytest.raises(asyncio.CancelledError): + await run(definition, _request(_RaisingExecutor(asyncio.CancelledError()))) + + finished = store.of_type(RunFinished) + assert len(finished) == 1 + assert finished[0].result.stop_reason is StopReason.CANCELLED # type: ignore[attr-defined] + + # --------------------------------------------------------------------------- # 并发隔离 # ---------------------------------------------------------------------------