"""五个 Protocol，以及只在一次调用往返之间存在的入参 / 返回壳。 读者是写适配器的人和 `tests/contract/`。用 `typing.Protocol` 而不是抽象基类：下游的对象 往往已经是它自己的类、还要同时满足项目自己更宽的接口，只有结构化子类型能让同一个对象同时 满足库的窄视图和项目的宽视图。 **签名上不许出现第三方类型**——出现了，那个包的 major 就是我们的 major。这一条由 `tests/unit/test_import_purity.py` 断言。 **什么住这里、什么住 `types`，判据是「它是不是一个值」**：值类型与持久化记录住 `types`， 只为一次调用打包入参或返回的壳住这里。判据不能写成「会不会被写进日志」——消息不出现在任何 一条日志记录里，照那条会判进这里，而上下文住 `types` 且字段就是消息序列，于是 `types` 反向 依赖本模块，分层契约当场违规。展开见 `research-wiki/design/0006-public-names-and-signatures.md` 决策四。 每个方法的行为契约在 `research-wiki/design/0007-seam-behaviour.md`，机器形式在 `tests/contract/`。这里的 docstring 只写「读这段代码的人不知道就会写错什么」。 """ from collections.abc import Mapping from dataclasses import dataclass from typing import Protocol, runtime_checkable from polyloop.types import ( ActionOutcome, Intent, Message, ModelCallResult, ModelReply, RunFinished, RunStarted, StepCompleted, ) # --------------------------------------------------------------------------- # 只在一次调用往返之间存在的壳 # --------------------------------------------------------------------------- @dataclass(frozen=True, slots=True) class ModelCall: """交给模型调用接缝的一次调用。""" messages: tuple[Message, ...] #: **本次运行内的第几次模型调用**，从 0 递增。它不是「同一个目标的第几次独立重做」—— #: 后者是项目自己的 metadata，库不理解、原样透传在 `binding` 里。两者撞过名，这里换个 #: 词就是为了不再撞。 call_index: int run_id: str #: 库预分配的结果 ID。 result_id: str #: 项目自己的标识，库不解释，原样透传。用字符串映射而不是不透明对象，是为了它能逐字段 #: 进参数快照——不透明对象是公共签名上一个永久的洞，洞里的东西快照看不见、契约测试也 #: 看不见。 binding: Mapping[str, str] @dataclass(frozen=True, slots=True) class ToolCall: """一次工具调用的名字与参数。""" name: str arguments: Mapping[str, object] @dataclass(frozen=True, slots=True) class Action: """决策解释出的动作。""" #: 这一步的动作在轨迹里长什么样，由实现方决定内容，库原样填进步记录的 `action`。 #: **不叫 `trace`**：那个词通常指整条执行轨迹，用它指单步的一个字符串会和「逐步轨迹」撞。 text: str #: 工具型动作才有；代码执行型的为 None。 tool_call: ToolCall | None @dataclass(frozen=True, slots=True) class FinalAnswer: """模型给出的最终回答。这一支不碰环境。""" text: str @dataclass(frozen=True, slots=True) class InvalidDecision: """解释不出有效动作。""" #: **这段文本就是回喂给模型的那段观察**，不是从一个固定串里取。不同的解析失败要给不同 #: 的对症说明——压成一句会改掉模型收到的纠错信息，它的纠错行为也就跟着变。 explanation: str #: 一次模型回复的三种解释结果。 Decision = Action | FinalAnswer | InvalidDecision @dataclass(frozen=True, slots=True) class ParsedReply: """决策解释接缝的返回。""" #: 这一步回填进历史的那段 assistant 文本。**它可以与模型原文不同**——解释器有权改写它， #: 比如把第一个代码围栏之后的内容整段丢掉（模型常在代码块后面编造执行结果）。库这边只有 #: 模型原文，照它回填，模型下一轮会看见自己编的那段。 history_text: str decision: Decision @dataclass(frozen=True, slots=True) class RunLog: """一份读回来的完整日志。恢复判定读它。 五个字段全部有默认值，所以「读一个从没写过的运行标识」返回的空日志就是 `RunLog()`。 """ started: RunStarted | None = None intents: tuple[Intent, ...] = () model_results: tuple[ModelCallResult, ...] = () steps: tuple[StepCompleted, ...] = () finished: RunFinished | None = None @dataclass(frozen=True, slots=True) class Event: """从事件出口发出去的一条事件。 **字段还没定。** 事件集与具名回调清单要独立成一份 design doc；在那之前这个类型只有名字， `EventSink.emit` 的签名不会因为它定下来而改变。 """ # --------------------------------------------------------------------------- # 五个接缝 # --------------------------------------------------------------------------- @runtime_checkable class ModelClient(Protocol): """模型调用接缝。挂在定义上。 **签名里不出现重试次数、退避时长、限流配额**——出现即意味着库在治理一次模型调用，而那 归下面那一层的共用库（`CLAUDE.md` §1.5）。 失败以异常表达，不以「返回一个内容为空的正常回复」表达：后者让库没有任何办法把基础设施 故障和「模型真的回了空字符串」分开，而这两者在分析里属于完全不同的类别。 """ async def call(self, call: ModelCall) -> ModelReply: ... def parameters(self) -> Mapping[str, str]: """上报这个实现的可复现参数，供参数快照聚合。 **同步、不许做 I/O。** 快照要能在装配之后立刻算出来，一个会发网络请求的实现会让 「构造廉价」这条承诺失效，也会让快照的取值依赖当时网络通不通。 """ ... @runtime_checkable class DecisionParser(Protocol): """决策解释接缝。挂在定义上。 **`parse` 是同步的**，因为解释一次模型回复是纯计算、没有等待点。写成协程会诱导实现方 在里面做 I/O，而这个接缝一旦做起 I/O，「恢复时重新解释被打断的那一步」就不再安全。 **不许抛异常**：对任何输入都要返回一个 `ParsedReply`，解释不出来就走 `InvalidDecision` 那一支。「无法解释」本来就是正常路径的一部分——模型输出不合格式是每天都在发生的事。 真抛了库也不接管：接住就得给它编一个停止原因，而编出来的原因会把「解释器有 bug」伪装成 「这次运行以某某原因结束」，然后进下游的统计。 """ def parse(self, reply: ModelReply) -> ParsedReply: ... def parameters(self) -> Mapping[str, str]: ... @runtime_checkable class ActionExecutor(Protocol): """动作执行接缝。**挂在请求上**，因为它每次运行都不同。 **动作本身报错算「已执行」**，不算环境故障：代码抛异常、命令返回非零都是正常观察，要 原样回喂让模型自己纠正。判成环境故障会让一次运行在模型本来能自我纠正的地方直接终止， 而轨迹上看不出它本可以继续。分界线是环境还能不能接着服务。 """ async def execute(self, action: Action) -> ActionOutcome: ... def parameters(self) -> Mapping[str, str]: """上报这个实现的可复现参数。 **这个接缝也必须上报，理由和别的一样硬。** 它可能是一个已经开好的会话，而会话是不是 有状态会改变跨步语义——换一个会话续跑而快照不比对，前几步的副作用留在旧会话里、 后几步在新会话上执行，全程零报错。 """ ... @runtime_checkable class RunStore(Protocol): """存储接缝。挂在定义上。 **端口不持有「当前运行」的隐式状态**：运行标识住在每一条记录里。一个有隐式当前运行的 实现会在并发下把 A 的意图写进 B 的日志，而那种错在单线程测试里永远不出现。 **写入粒度是契约的一部分**：两条意图各自单独落地；动作结果与步记录由 `write_step_completed` 一次原子落地，要么都可见、要么都不可见。实现还必须保证**前缀持久性**——第 k 次写入被确认 持久时，第 1 到 k-1 次也已经持久。 不设「这个 ID 的结果存在吗」这类存在性查询：它可以由 `read_log` 推出来，端口少一个方法 就是少一份永久合同。 """ async def write_run_started(self, record: RunStarted) -> None: ... async def write_intent(self, record: Intent) -> None: ... async def write_model_call_result(self, record: ModelCallResult) -> None: ... async def write_step_completed(self, record: StepCompleted) -> None: ... async def read_log(self, run_id: str) -> RunLog: """读回整份日志。 **读一个从没写过的运行标识时返回空日志，不抛异常。** 驱动入口开工前要判断「这个标识 是不是已经有日志了」，靠的就是这条。抛异常的话那个判断就得写成捕获异常，而用捕获 异常做流程控制会把真正的存储故障一起吞掉——于是「存储连不上」会被读成「这是一次全新 的运行」，然后覆盖式地重跑一遍。 它也是「每个方法只收一个记录对象」那条规则的例外：读的时候还没有记录对象可传。 """ ... async def write_run_finished(self, record: RunFinished) -> None: ... def parameters(self) -> Mapping[str, str]: ... @runtime_checkable class EventSink(Protocol): """事件出口。挂在定义上。 **允许抛异常。** 投递失败由库捕获、记日志、把失败计数加一，然后继续跑——事件是观察通道， 不是控制通道，一次运行不该因为进度回写的数据库连不上就终止。所以一个在后端不可用时抛 异常的实现是合规的。 库**不会**把投递失败转成一条事件从同一个出口再发一次：那会自我喂食，一个持续失败的出口 会让失败处理路径变成递归。 """ async def emit(self, event: Event) -> None: ... def parameters(self) -> Mapping[str, str]: ... __all__ = [ "Action", "ActionExecutor", "Decision", "DecisionParser", "Event", "EventSink", "FinalAnswer", "InvalidDecision", "ModelCall", "ModelClient", "ParsedReply", "RunLog", "RunStore", "ToolCall", ]