test(contract): 落成五个接缝的契约骨架，用它逼出七个设计洞
第 ④ 阶段的核心交付物。这套测试不针对任何具体实现，写的是「不管你怎么实现 都必须满足这些行为」，下游写完自己的实现接到 fixture 上跑一遍即可， 是任何新适配器的准入标准（CLAUDE.md §0）。 现在全部跳过，因为公共类型与 Protocol 还没落地。价值不在跑，在写： 写一条契约要求把每次调用逐字写出来——方法叫什么、参数填什么、返回值怎么取， 而散文里读着通顺的地方，落到这一步就露出来了。 三轮文档评审没报出的七个洞，写这套测试时全部撞了出来： 没有动作的步 result_id 填什么；三个 ActionStatus 取值的触发条件； 动作被拒绝时观察的来源（执行器与 SyntheticObservations 两处都有）； 解释器能不能抛异常；Event 没有字段所以事件出口的契约只写得出一半； read_log 读不存在的运行必须返回空日志而不是抛异常； 以及原子性与前缀持久性这两条 0005 的承诺根本没有机器兜底—— 写这套测试之前我们默认它们会被契约测试接住。 洞标成 xfail(strict=True) 而不是常驻 fail：一个永远红的套件会训练所有人忽略红。 它们都不带 fixture，否则会被「实现还没有」那个跳过挡住， 于是「答不上来」就伪装成了「还没轮到」。补上之后 XPASS 会报错，逼人回来删标记。 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,73 @@
|
||||
"""契约套件的装配点。
|
||||
|
||||
这套测试**不针对任何具体实现**。它写的是「不管你怎么实现，都必须满足这些行为」，所以它
|
||||
自己不造实现，只声明「你得提供什么」。任何一个下游写完自己的存储或适配器，把它接到这里
|
||||
的 fixture 上跑一遍，全绿就算合格——这是 `CLAUDE.md` §0 说的「任何新适配器的准入标准」。
|
||||
|
||||
**接法**：下游在自己的 `conftest.py` 里覆盖同名 fixture，返回自己的实现。
|
||||
|
||||
**现在这套测试全部跳过**，因为 `polyloop` 下还没有任何公共类型与 Protocol
|
||||
（`research-wiki/design/0006-public-names-and-signatures.md` 还没过人类门）。跳过的理由写在
|
||||
每个 fixture 里，读跳过原因就能知道缺的是哪一块。
|
||||
|
||||
**为什么在实现之前就写它**：写一条契约测试要求把每一次调用逐字写出来——方法叫什么、参数
|
||||
填什么、返回值怎么取。散文里读着通顺的地方，落到这一步就会露出来。前三轮文档评审抓不到的
|
||||
洞，几乎全是这么冒出来的。
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
#: 公共类型与 Protocol 落地之前，套件里的每一条都缺同一样东西。
|
||||
_NOT_YET = (
|
||||
"polyloop 的公共类型与 Protocol 还没落地（design/0006 待确认）。"
|
||||
"这条契约要断言的行为已经写在测试的 docstring 里，落地之后去掉这个跳过即可。"
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def store():
|
||||
"""被测的存储接缝实现。
|
||||
|
||||
下游覆盖这个 fixture，返回自己的实例。每次调用应当返回一个**空的**存储——套件里的每条
|
||||
测试都假设自己面对一份干净的日志，共用状态会让测试之间的顺序变成隐式依赖。
|
||||
"""
|
||||
pytest.skip(_NOT_YET)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def records():
|
||||
"""构造各类记录的辅助工厂。
|
||||
|
||||
它不是被测对象，是让测试正文读得懂的一层薄封装：`records.model_call_intent(...)` 比
|
||||
直接写一长串构造参数更能看出这条测试在断言什么。工厂本身由库提供，因为记录类的字段
|
||||
是库的公共承诺，下游不该为了跑契约测试去手写构造。
|
||||
"""
|
||||
pytest.skip(_NOT_YET)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def action_executor():
|
||||
"""被测的动作执行接缝实现。"""
|
||||
pytest.skip(_NOT_YET)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def decision_parser():
|
||||
"""被测的决策解释接缝实现。"""
|
||||
pytest.skip(_NOT_YET)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def model_client():
|
||||
"""被测的模型调用接缝实现。
|
||||
|
||||
注意这一层的契约测试**不打真实网关**——那是 e2e 的事。这里断言的是返回结构体的形状
|
||||
与失败时的表达方式，用一个受控的替身就能验。
|
||||
"""
|
||||
pytest.skip(_NOT_YET)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def event_sink():
|
||||
"""被测的事件出口实现。"""
|
||||
pytest.skip(_NOT_YET)
|
||||
@@ -0,0 +1,101 @@
|
||||
"""动作执行接缝的行为契约。
|
||||
|
||||
两个已知形态差别很大：一个把一段代码交给已经开好的容器会话、状态恒为「已执行」，一个查
|
||||
工具注册表分发、工具不存在或参数不合法时返回「未执行」。下面每一条都要对两者同时成立。
|
||||
|
||||
## 写这份文件时撞出来的、`design/0006` 还答不上的问题
|
||||
|
||||
1. 三个状态取值分别在什么条件下被赋上，从来没有正面写过。
|
||||
2. 返回「未执行」时，那段观察是执行器给的还是库合成的——两处都有来源，没说以谁为准。
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
pytestmark = pytest.mark.contract
|
||||
|
||||
|
||||
async def test_returns_all_five_fields(action_executor, records):
|
||||
"""返回值必须带齐五个字段，一个都不能省。
|
||||
|
||||
库拿这五个字段填步记录里对应的五列。少一个，那一列就只能填默认值，而默认值与真实值在
|
||||
轨迹里长得一模一样——事后没有任何办法把「执行器没给」和「值确实是这个」分开。
|
||||
"""
|
||||
outcome = await action_executor.execute(records.action(text="noop"))
|
||||
|
||||
assert outcome.status is not None
|
||||
assert isinstance(outcome.observation, str)
|
||||
assert isinstance(outcome.observation_is_synthetic, bool)
|
||||
assert isinstance(outcome.env_reported_completion, bool)
|
||||
assert isinstance(outcome.observation_truncated_chars, int)
|
||||
|
||||
|
||||
async def test_completion_signal_is_a_plain_boolean(action_executor, records):
|
||||
"""完成信号是布尔，没有第三个取值，恒为「未完成」不是故障。
|
||||
|
||||
没有环境完成信号的环境就是这么返回的——GovDoc 全部、dissect 的两个非 AppWorld
|
||||
benchmark 都是。初稿把它定成「可为空表示取不到」并把空值判为环境故障，照那个写法
|
||||
GovDoc 的每一次运行都会在第一步撞环境故障终止。
|
||||
"""
|
||||
outcome = await action_executor.execute(records.action(text="noop"))
|
||||
|
||||
assert outcome.env_reported_completion in (True, False)
|
||||
|
||||
|
||||
async def test_action_error_is_a_normal_observation_not_an_env_error(action_executor, records):
|
||||
"""动作本身报错是正常观察，要原样回喂让模型自己纠正，不是环境故障。
|
||||
|
||||
代码抛异常、命令返回非零，都属于这一类。判成环境故障会让一次运行在模型本来能自我纠正
|
||||
的地方直接终止，而轨迹上看不出它本可以继续。只有环境自己坏了（连不上、协议不对）才
|
||||
另算。
|
||||
"""
|
||||
outcome = await action_executor.execute(records.action(text="raise RuntimeError()"))
|
||||
|
||||
assert outcome.status == records.action_status.EXECUTED
|
||||
assert outcome.observation != ""
|
||||
|
||||
|
||||
async def test_cancellation_propagates_and_is_not_swallowed(action_executor, records):
|
||||
"""取消要能穿过动作执行，`CancelledError` 不许被捕获吞没。
|
||||
|
||||
吞掉它的后果不是「取消失败」这么直白——是容器租约、连接和临时目录持续泄漏，而且一声
|
||||
不吭。这条是 `CLAUDE.md` §1.6，对每一个执行器实现都成立。
|
||||
"""
|
||||
import asyncio
|
||||
|
||||
task = asyncio.ensure_future(action_executor.execute(records.action(text="sleep")))
|
||||
await asyncio.sleep(0)
|
||||
task.cancel()
|
||||
|
||||
with pytest.raises(asyncio.CancelledError):
|
||||
await task
|
||||
|
||||
|
||||
@pytest.mark.xfail(reason="design/0006 答不上，见 docstring", strict=True)
|
||||
def test_status_values_have_defined_trigger_conditions():
|
||||
"""三个状态取值各自在什么条件下被赋上。
|
||||
|
||||
`design/0006` 只列了 `EXECUTED` / `NOT_EXECUTED` / `ENV_ERROR` 三个取值，没有正面写过
|
||||
触发条件。现在只能从 `SyntheticObservations` 那两个字段名反推——工具不存在或参数不合法
|
||||
大概是 `NOT_EXECUTED`，环境故障大概是 `ENV_ERROR`——而「大概」不能写成断言。
|
||||
|
||||
这条不定下来，两个下游会各自理解一套，而两套都不报错：dissect 的执行器状态恒为
|
||||
`EXECUTED`，它撞不到这个分歧；GovDoc 撞得到，但表现是停止原因的分布变了，不是异常。
|
||||
|
||||
还有一处连带的：`ActionStatus.ENV_ERROR` 与 `StopReason.ENV_ERROR` 同名不同类型，前者
|
||||
出现是不是必然导致后者，也没写。
|
||||
"""
|
||||
pytest.fail("三个 ActionStatus 取值的触发条件没有定义")
|
||||
|
||||
|
||||
@pytest.mark.xfail(reason="design/0006 答不上，见 docstring", strict=True)
|
||||
def test_who_supplies_the_observation_when_the_action_is_rejected():
|
||||
"""动作被拒绝时，那段观察是执行器给的还是库合成的。
|
||||
|
||||
两处都有来源：执行器的返回值里有 `observation` 字段，而定义上又挂着
|
||||
`SyntheticObservations.action_rejected`。`design/0006` 没说以谁为准。
|
||||
|
||||
这不是风格问题。如果以库为准，执行器填的那段就被丢掉，而它可能带着「哪个参数不合法」
|
||||
这种只有执行器知道的信息；如果以执行器为准，那 `SyntheticObservations` 那个字段永远
|
||||
用不上，它就是个死字段。而 `observation_is_synthetic` 该填什么，取决于这个答案。
|
||||
"""
|
||||
pytest.fail("动作被拒绝时观察的来源没有定义")
|
||||
@@ -0,0 +1,77 @@
|
||||
"""决策解释接缝的行为契约。
|
||||
|
||||
两个已知形态：一个从代码围栏里抽 Python 源码，一个从 JSON 里抽工具名与参数。库不带任何
|
||||
默认实现——带了就等于替某一家定了动作语言。
|
||||
|
||||
## 写这份文件时撞出来的、`design/0006` 还答不上的问题
|
||||
|
||||
模型输出完全无法解释时，解释器是返回「无效决策」还是抛异常。
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
pytestmark = pytest.mark.contract
|
||||
|
||||
|
||||
def test_parse_is_synchronous(decision_parser, records):
|
||||
"""`parse` 是同步的，不是协程。
|
||||
|
||||
解释一次模型回复是纯计算，没有等待点。写成协程会让每个只想写测试替身的下游多套一层
|
||||
`async def`，也会诱导实现方在里面做 I/O——而这个接缝一旦做起 I/O，「恢复时重新解释
|
||||
被打断的那一步」就不再是安全操作了。
|
||||
"""
|
||||
parsed = decision_parser.parse(records.reply(content="anything"))
|
||||
|
||||
assert not hasattr(parsed, "__await__")
|
||||
|
||||
|
||||
def test_history_text_is_what_goes_back_into_the_conversation(decision_parser, records):
|
||||
"""`history_text` 是这一步回填进历史的那段文本，可以与模型原文不同。
|
||||
|
||||
解释器有权改写它：dissect 的解析器把第一个代码围栏之后的内容整段丢掉，因为模型常在
|
||||
代码块后面编造「执行结果」。库这边只有模型原文，照它回填，模型下一轮会看见自己编的
|
||||
那段，而迁移前它看不见。
|
||||
"""
|
||||
reply = records.reply(content="```python\nprint(1)\n```\nThe command succeeded.")
|
||||
parsed = decision_parser.parse(reply)
|
||||
|
||||
assert isinstance(parsed.history_text, str)
|
||||
assert len(parsed.history_text) <= len(reply.content)
|
||||
|
||||
|
||||
def test_invalid_decision_explanation_is_what_is_fed_back(decision_parser, records):
|
||||
"""无效决策的说明文本**就是**回喂给模型的那段观察，不是从一个固定串里取。
|
||||
|
||||
dissect 的解析器对五种解析失败各有一条对症说明（没有代码块、空的未闭合块、闭合围栏后
|
||||
跟了别的内容、多块策略下第一块为空、拼接策略下全空）。压成一句会改掉它的实验条件——
|
||||
模型收到的纠错信息变了，它的纠错行为也就变了。
|
||||
"""
|
||||
parsed = decision_parser.parse(records.reply(content="没有任何代码块"))
|
||||
|
||||
assert isinstance(parsed.decision.explanation, str)
|
||||
assert parsed.decision.explanation != ""
|
||||
|
||||
|
||||
def test_action_carries_its_trace_form(decision_parser, records):
|
||||
"""动作分支要带「这一步的动作在轨迹里长什么样」，由实现方决定内容，库原样填进步记录。
|
||||
|
||||
dissect 传那段 Python 源码，GovDoc 传序列化后的参数。没有这个字段，dissect 轨迹里那一列
|
||||
会被库改写，而那个文件是它的反思模型的唯一输入界面。
|
||||
"""
|
||||
parsed = decision_parser.parse(records.reply(content="```python\nprint(1)\n```"))
|
||||
|
||||
assert isinstance(parsed.decision.text, str)
|
||||
|
||||
|
||||
@pytest.mark.xfail(reason="design/0006 答不上，见 docstring", strict=True)
|
||||
def test_unparseable_output_returns_invalid_decision_rather_than_raising():
|
||||
"""模型输出完全无法解释时，解释器返回「无效决策」还是抛异常。
|
||||
|
||||
`design/0006` 定了三个分支——动作、最终回答、无效决策——但没说「解释器可以抛异常吗」。
|
||||
两条路后果完全不同：返回无效决策，那一步照常留痕、说明文本回喂给模型、循环继续；抛
|
||||
异常，库要么把它翻译成某个停止原因终止整次运行，要么让它穿出去炸掉调用方。
|
||||
|
||||
dissect 的解析器不抛异常，所以它撞不到这个分歧。但契约测试是**任何新适配器的准入
|
||||
标准**，一个会抛异常的实现照现在的契约既不算违规也不算合规。
|
||||
"""
|
||||
pytest.fail("解释器能不能抛异常、抛了怎么办，没有定义")
|
||||
@@ -0,0 +1,55 @@
|
||||
"""事件出口的行为契约。
|
||||
|
||||
两个已知形态差别在可靠性要求上：一个把进度逐步回写业务数据库供前端轮询（要求低延迟、
|
||||
可以丢），一个把审计事件送进日志管道（要求不丢、可以慢）。
|
||||
|
||||
## 写这份文件时撞出来的、`design/0006` 还答不上的问题
|
||||
|
||||
`Event` 只有一个名字，没有字段，所以这个接缝的契约现在只能验「投递失败不打断循环」这一半，
|
||||
验不了「发出去的事件里有什么」。
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
pytestmark = pytest.mark.contract
|
||||
|
||||
|
||||
async def test_emit_accepts_an_event(event_sink, records):
|
||||
"""能收下一个事件，正常路径不抛异常。"""
|
||||
await event_sink.emit(records.event())
|
||||
|
||||
|
||||
async def test_delivery_failure_does_not_break_the_caller(event_sink, records):
|
||||
"""投递失败由库捕获、记日志、把失败计数加一，然后继续跑。
|
||||
|
||||
一次运行不该因为进度回写的数据库连不上就终止——事件是观察通道，不是控制通道。
|
||||
这条测试面对的是一个必然投递失败的出口，断言它不会把异常泄漏成循环的终止条件。
|
||||
"""
|
||||
await event_sink.emit(records.event(kind="always-fails"))
|
||||
|
||||
|
||||
@pytest.mark.xfail(reason="事件集还没定，见 docstring", strict=True)
|
||||
def test_failure_is_not_re_emitted_through_the_same_sink():
|
||||
"""投递失败不再转成一条事件从同一个出口发出去。
|
||||
|
||||
那会自我喂食：一个持续失败的出口会让失败处理路径变成递归，而递归的表现是进程卡住或
|
||||
栈溢出，不是一条错误日志。
|
||||
|
||||
**这条现在验不了**，因为验它要求能识别「这是一条失败事件」，而 `Event` 还没有字段——
|
||||
`design/0006` 里它只有一个名字。方向已经定了（观察走事件流、干预走具名回调），但事件
|
||||
集与回调清单要独立成一份 design doc，这条要等到那时候。
|
||||
"""
|
||||
pytest.fail("Event 还没有字段，识别不了「失败事件」")
|
||||
|
||||
|
||||
@pytest.mark.xfail(reason="事件集还没定，见 docstring", strict=True)
|
||||
def test_audit_events_carry_both_raw_and_repaired_model_output():
|
||||
"""审计事件要同时带模型原文与修复之后的结果。
|
||||
|
||||
GovDoc 有一条硬纪律：agent 的原始输出、修复后的输出、恢复来源全程留痕，禁止静默修复。
|
||||
它现有的审计出口是一个「发一条带类型和载荷的事件」的接口，迁移之后这条纪律要由事件流
|
||||
承载——能不能承载，取决于事件里带不带这两样。
|
||||
|
||||
这是 `../research-wiki/migrations/govdoc-saas.md` 缺口登记里那一条，同样等事件集定下来。
|
||||
"""
|
||||
pytest.fail("Event 还没有字段，承载不了审计纪律")
|
||||
@@ -0,0 +1,74 @@
|
||||
"""模型调用接缝的行为契约。
|
||||
|
||||
**这一层不打真实网关**——那是 e2e 的事。这里断言的是返回结构体的形状与失败的表达方式，
|
||||
用一个受控替身就能验。
|
||||
|
||||
两个已知形态：一个按三本账各记一条并自己按价格表算成本，一个在调用外面套退避并累加本次
|
||||
运行的 token。
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
pytestmark = pytest.mark.contract
|
||||
|
||||
|
||||
async def test_returns_three_fields(model_client, records):
|
||||
"""返回三个字段：调用标识、可见回复、推理段。
|
||||
|
||||
可见回复与推理段的长度由库自己数字符，不从任何用量对象取——实测中转网关会用本地分词器
|
||||
补算并整体替换用量对象，把明细一起吃掉，某次标定里 24 次调用的推理 token 全部没上报。
|
||||
"""
|
||||
reply = await model_client.call(records.model_call(call_index=0, result_id="m0"))
|
||||
|
||||
assert isinstance(reply.content, str)
|
||||
assert isinstance(reply.thinking, str)
|
||||
assert reply.call_id is None or isinstance(reply.call_id, str)
|
||||
|
||||
|
||||
async def test_call_id_is_never_an_empty_string(model_client, records):
|
||||
"""调用标识可以是「没有」，但绝不能是空串。
|
||||
|
||||
它是轨迹与账目之间唯一的连接键。空串是个「看起来合法」的键，连表时静默匹配不上；显式
|
||||
的「没有」至少能被筛出来。它为空的合法含义只有一个：调用在记账之前就失败了。
|
||||
"""
|
||||
reply = await model_client.call(records.model_call(call_index=0, result_id="m0"))
|
||||
|
||||
assert reply.call_id != ""
|
||||
|
||||
|
||||
async def test_failure_is_expressed_as_an_exception(model_client, records):
|
||||
"""调用失败以异常表达，不以「返回一个空回复」表达。
|
||||
|
||||
库接住它、翻译成模型故障、记一条调用标识为空的步。如果失败被表达成一个内容为空串的
|
||||
正常返回，库没有任何办法把它和「模型真的回了空字符串」分开——而后者是模型行为，前者
|
||||
是基础设施故障，两者在分析里属于完全不同的类别。
|
||||
"""
|
||||
with pytest.raises(Exception): # noqa: B017 具体异常类型归实现，契约只要求「抛」
|
||||
await model_client.call(records.model_call(call_index=0, result_id="fail"))
|
||||
|
||||
|
||||
async def test_cancellation_propagates_and_is_not_swallowed(model_client, records):
|
||||
"""取消要能穿过模型调用，`CancelledError` 不许被捕获吞没。"""
|
||||
import asyncio
|
||||
|
||||
task = asyncio.ensure_future(
|
||||
model_client.call(records.model_call(call_index=0, result_id="m0"))
|
||||
)
|
||||
await asyncio.sleep(0)
|
||||
task.cancel()
|
||||
|
||||
with pytest.raises(asyncio.CancelledError):
|
||||
await task
|
||||
|
||||
|
||||
def test_signature_carries_no_retry_or_rate_limit_parameters(model_client):
|
||||
"""签名里不出现重试次数、退避时长、限流配额。
|
||||
|
||||
出现即意味着库在治理一次模型调用，而那归 PolyGateway（`CLAUDE.md` §1.5）。这条断言的是
|
||||
名字，不是行为——按 §1.8，公共 Protocol 的签名本身就是对下游的承诺，断言它是应该的。
|
||||
"""
|
||||
import inspect
|
||||
|
||||
names = set(inspect.signature(model_client.call).parameters)
|
||||
|
||||
assert not (names & {"retries", "max_retries", "backoff", "timeout", "rate_limit"})
|
||||
@@ -0,0 +1,234 @@
|
||||
"""存储接缝的行为契约。
|
||||
|
||||
这份文件是「一次运行的日志到底保证什么」的权威（`CLAUDE.md` §0）。两个已知实现形态差别
|
||||
很大——一个逐行追加本地文件，一个写关系数据库——所以下面每一条都只说行为，不碰形态。
|
||||
|
||||
**行为的理由不在这里。** 崩溃恢复为什么这么设计见 `design/0002`，写入粒度与前缀持久性见
|
||||
`design/0005`。这里只断言结果。
|
||||
|
||||
## 写这份文件时撞出来的、`design/0006` 还答不上的问题
|
||||
|
||||
每一条都在下面对应的测试里标成 `xfail`，摘要里每次都看得见，但不把套件拖红——**一个永远
|
||||
红的套件会训练所有人忽略红**。它们也都不带 fixture，否则会被「实现还没有」那个跳过挡住，
|
||||
于是「答不上来」就伪装成了「还没轮到」。
|
||||
|
||||
`strict=True` 是配套的：哪天这个洞被补上、测试真的能过了，它会以 XPASS 报错，逼人回来把
|
||||
这个标记连同这段说明一起删掉。
|
||||
|
||||
1. 没有动作的那些步，`StepCompleted.result_id` 填什么。
|
||||
2. 「重放」是把动作再执行一次，还是把上次的结果填回去。
|
||||
3. 原子性与前缀持久性能不能写成契约测试。
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
pytestmark = pytest.mark.contract
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 一、写进去的读得回来
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def test_written_intent_is_readable(store, records):
|
||||
"""写一条意图，读回整份日志时它必须在里面。
|
||||
|
||||
这是全套最基本的一条：意图日志的全部意义是「比进程活得久」，写了读不回来，后面每一条
|
||||
恢复语义都建立在空气上。
|
||||
"""
|
||||
intent = records.model_call_intent(run_id="r1", call_index=0, result_id="m0")
|
||||
await store.write_intent(intent)
|
||||
|
||||
log = await store.read_log("r1")
|
||||
|
||||
assert intent in log.intents
|
||||
|
||||
|
||||
async def test_log_of_unknown_run_is_empty_not_an_error(store):
|
||||
"""读一个从没写过的运行标识，得到一份空日志，而不是异常。
|
||||
|
||||
`run` 在开工前要判断「这个标识是不是已经有日志了」，靠的就是这一条。如果读不存在的
|
||||
运行会抛异常，那个判断就得写成捕获异常——而捕获异常来做流程控制，会把真正的存储故障
|
||||
一起吞掉。
|
||||
"""
|
||||
log = await store.read_log("never-written")
|
||||
|
||||
assert log.started is None
|
||||
assert log.intents == ()
|
||||
assert log.finished is None
|
||||
|
||||
|
||||
async def test_two_runs_do_not_leak_into_each_other(store, records):
|
||||
"""两个运行标识各写各的，互相看不见对方的记录。
|
||||
|
||||
端口不持有「当前运行」的隐式状态，这条测试是那个要求的外部可观测形式。一个有隐式当前
|
||||
运行的实现会在并发下把 A 的意图写进 B 的日志，而那种错在单线程测试里永远不出现。
|
||||
"""
|
||||
a = records.model_call_intent(run_id="run-a", call_index=0, result_id="m0")
|
||||
b = records.model_call_intent(run_id="run-b", call_index=0, result_id="m0")
|
||||
await store.write_intent(a)
|
||||
await store.write_intent(b)
|
||||
|
||||
assert (await store.read_log("run-a")).intents == (a,)
|
||||
assert (await store.read_log("run-b")).intents == (b,)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 二、四态：恢复靠「意图有没有 / 结果有没有」判定
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def test_intent_without_result_is_readable_as_such(store, records):
|
||||
"""写了意图、没写结果，读回来必须能看出「这个 ID 没有结果」。
|
||||
|
||||
这是四态表里「状态未知」那一档的输入。存储不负责判定，但它必须让判定问得出口——恢复
|
||||
要按预分配的 ID 精确地问，而不是模糊匹配去猜哪条结果对应哪次执行。
|
||||
"""
|
||||
intent = records.action_intent(run_id="r1", call_index=0, result_id="a0")
|
||||
await store.write_intent(intent)
|
||||
|
||||
log = await store.read_log("r1")
|
||||
|
||||
assert intent in log.intents
|
||||
assert all(step.result_id != "a0" for step in log.steps)
|
||||
|
||||
|
||||
async def test_result_without_intent_is_visible_to_the_reader(store, records):
|
||||
"""只写结果不写意图，读回来必须原样可见，存储自己不许修复也不许拒收。
|
||||
|
||||
「有结果没意图」是日志损坏，处置是拒绝续跑——但那个判断归恢复逻辑，不归存储。存储在
|
||||
这里悄悄补一条意图或者拒绝这次写入，都会让损坏变得不可见，而不可见的损坏会被当成
|
||||
正常数据继续用下去。
|
||||
"""
|
||||
result = records.model_call_result(run_id="r1", result_id="orphan", reply=records.reply())
|
||||
await store.write_model_call_result(result)
|
||||
|
||||
log = await store.read_log("r1")
|
||||
|
||||
assert log.model_results == (result,)
|
||||
assert log.intents == ()
|
||||
|
||||
|
||||
async def test_failed_model_call_is_recorded_as_a_result_not_as_nothing(store, records):
|
||||
"""模型调用失败也要落一条结果记录，否则恢复会把它读成「状态未知」。
|
||||
|
||||
失败这件事是确定的：调用发出去了、失败了、库记了一条步。如果这时不写结果条目，恢复
|
||||
只看见「意图有、结果无」，走重放策略——而这次调用的状态一点都不未知。下游按停止原因
|
||||
做的统计会照单收下这个错误。
|
||||
"""
|
||||
result = records.model_call_result(run_id="r1", result_id="m0", reply=None, failure="连接超时")
|
||||
await store.write_model_call_result(result)
|
||||
|
||||
(readback,) = (await store.read_log("r1")).model_results
|
||||
|
||||
assert readback.reply is None
|
||||
assert readback.failure == "连接超时"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 三、原子写
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def test_action_result_and_step_land_together(store, records):
|
||||
"""动作结果与步记录一次原子落地：读回来要么两者都在，要么都不在。
|
||||
|
||||
不原子的话，崩在两者之间会让那一步的历史文本永远丢失，而恢复判定会把它读成「执行完了，
|
||||
跳过」——恢复出来的消息序列比不中断跑完时少一轮，后面每一步都跟着偏。
|
||||
|
||||
**这条测试只能验「一起可见」，验不了「一起不可见」。** 见本文件末尾那条。
|
||||
"""
|
||||
step = records.step_completed(
|
||||
run_id="r1", result_id="a0", action_outcome=records.outcome(), step=records.step(step_idx=0)
|
||||
)
|
||||
await store.write_step(step)
|
||||
|
||||
log = await store.read_log("r1")
|
||||
|
||||
assert log.steps == (step,)
|
||||
assert log.steps[0].action_outcome is not None
|
||||
|
||||
|
||||
@pytest.mark.xfail(reason="design/0006 答不上，见 docstring", strict=True)
|
||||
def test_step_without_an_action_is_still_recorded():
|
||||
"""没有动作的步照样留痕：解析失败、模型调用失败、环境故障三种都算一步。
|
||||
|
||||
dissect 的预算对等要求它们计入步数——它们确实消耗了一次模型调用。丢掉那一步还会丢掉
|
||||
模型在出故障时说了什么，而那正是排查「环境坏了还是模型写了危险代码」最需要的。
|
||||
|
||||
**这条现在写不出来。** 这一步没有写过动作意图，所以没有预分配的结果 ID，而
|
||||
`StepCompleted.result_id` 在 `design/0006` 里是必填的字符串。照那个形状写，恢复会读到
|
||||
一条对不上任何意图的记录，按 `design/0002` 四态表最后一行判为「日志损坏，拒绝续跑」
|
||||
——而这本该是一次恢复成 `llm_error` 正常终止的运行。
|
||||
|
||||
**它不带 fixture，所以不会被「实现还没有」那个跳过挡住。** 挡住了它就看起来像「还没
|
||||
轮到」，而它是「答不上来」，两者要分得开。
|
||||
"""
|
||||
pytest.fail("StepCompleted.result_id 对没有动作意图的步没有定义")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 四、运行的开始与结束
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def test_run_finished_is_visible_before_the_result_is_returned(store, records):
|
||||
"""「这次运行结束了」这个标记由库写下，而且写在把结果交给调用方之前。
|
||||
|
||||
另一条路有个具体的失败场景：结果由项目落盘的话，「跑完了、库返回了、项目存的时候崩了」
|
||||
这种情况下，重启后日志显示最后一步有结果、没有结束标记，而项目那边什么都没有。续跑会
|
||||
重复执行最后一步的副作用，不续跑就丢掉一次已经花完钱的运行。歧义来自结果跨了两个存储。
|
||||
"""
|
||||
finished = records.run_finished(run_id="r1", result=records.result(run_id="r1"))
|
||||
await store.write_run_finished(finished)
|
||||
|
||||
assert (await store.read_log("r1")).finished == finished
|
||||
|
||||
|
||||
async def test_run_started_carries_the_parameter_snapshot(store, records):
|
||||
"""运行开始记录带着这次的参数快照，续跑时拿它与当前装配比对。
|
||||
|
||||
没有它，用同一个运行标识换一份定义续跑，前几步与后几步会来自两个不同的配置而全程零
|
||||
报错——那正是要到统计阶段才分不清哪些行是真的那类损坏。
|
||||
"""
|
||||
started = records.run_started(run_id="r1", parameter_snapshot={"model": "m-1"})
|
||||
await store.write_run_started(started)
|
||||
|
||||
assert (await store.read_log("r1")).started.parameter_snapshot == {"model": "m-1"}
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 五、这套测试**验不了**的两条承诺
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.xfail(reason="已知缺口：这条承诺没有机器兜底", strict=True)
|
||||
def test_atomicity_under_crash_is_not_checkable_here():
|
||||
"""原子性的另一半——「崩在中间时两者都不可见」——这一层验不了。
|
||||
|
||||
要验它得在写入过程中把进程杀掉，而契约测试跑在一个进程里、面对的是一个已经装配好的
|
||||
实现，没有位置插入那次崩溃。给端口加一个「故意在这里失败」的钩子能验，但那个钩子会
|
||||
变成公共 API 的一部分，而它只为测试存在。
|
||||
|
||||
结论是这条承诺**没有机器兜底**，只能靠 `CLAUDE.md` §3 那轮对抗审查看实现。把这件事
|
||||
写成一条会失败的测试而不是一句注释，是为了让它在每次跑套件时都被看见。
|
||||
"""
|
||||
pytest.fail(
|
||||
"已知缺口：原子写的「一起不可见」这一半没有机器检查。"
|
||||
"落地时要在 stores 的 unit 测试里用可注入的故障点覆盖，"
|
||||
"并在 design/0005 决策二登记这条契约测试覆盖不到。"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.xfail(reason="已知缺口：这条承诺没有机器兜底", strict=True)
|
||||
def test_prefix_durability_is_not_checkable_here():
|
||||
"""前缀持久性同样验不了，理由更硬一层。
|
||||
|
||||
它说的是「第 k 次写入被确认持久时，前 k-1 次也已经持久」，而「已经持久」是掉电之后
|
||||
才看得出来的性质。在一个进程里读得回来，不等于它落了盘。
|
||||
"""
|
||||
pytest.fail(
|
||||
"已知缺口：前缀持久性没有机器检查。两个已知形态天然满足它"
|
||||
"（同一文件的追加写、同一连接上顺序提交的事务），"
|
||||
"所以它实际是对实现形态的约束，落地时靠评审看，不靠这套测试。"
|
||||
)
|
||||
Reference in New Issue
Block a user