From e71dca7c35e5cf39cd03e24b43be4c2d22e6c062 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Mon, 10 Aug 2026 22:51:19 -0400 Subject: [PATCH] =?UTF-8?q?feat(session):=20=E8=90=BD=E6=88=90=E4=BA=8B?= =?UTF-8?q?=E4=BB=B6=E5=87=BA=E5=8F=A3=EF=BC=8C=E4=BA=94=E4=B8=AA=E6=8E=A5?= =?UTF-8?q?=E7=BC=9D=E5=85=A8=E9=83=A8=E6=9C=89=E8=B0=83=E7=94=A8=E7=82=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Event 从零字段变成 kind + run_id + model_binding + step，新增 EventKind（只有一种取值， 但第一天就带 kind，逼每个出口分发）。事件在「一步走完」原子落地之后发，只有这次进程里 真的执行过的步才发；投递失败接住、计数进 RunResult、继续跑，CancelledError 原样穿过。 契约套件那两条 xfail 关掉：一条要断言的是库发了几次、接缝自己看不到；另一条的前提是错的 ——审计纪律由意图日志承担不由事件流承担，改成在 unit 层验日志里原文与改写后的文本各有 位置。_project_observation 那段说「将来靠事件流送出去」的注释一并改对。 283 passed / 15 skipped / 2 xfailed，剩下两条 xfail 是原子写与前缀持久性，没有机器兜底。 Co-Authored-By: Claude Opus 5 (1M context) --- src/polyloop/ports/__init__.py | 36 ++++- src/polyloop/session/__init__.py | 49 ++++++- tests/contract/conftest.py | 11 +- tests/contract/test_event_sink.py | 31 ++-- tests/unit/test_session.py | 234 +++++++++++++++++++++++++++++- 5 files changed, 331 insertions(+), 30 deletions(-) diff --git a/src/polyloop/ports/__init__.py b/src/polyloop/ports/__init__.py index dba278b..a6379a5 100644 --- a/src/polyloop/ports/__init__.py +++ b/src/polyloop/ports/__init__.py @@ -19,6 +19,7 @@ from collections.abc import Mapping from dataclasses import dataclass +from enum import StrEnum from typing import Protocol, runtime_checkable from polyloop.types import ( @@ -30,6 +31,7 @@ from polyloop.types import ( RunFinished, RunStarted, StepCompleted, + StepRecord, ) # --------------------------------------------------------------------------- @@ -119,14 +121,43 @@ class RunLog: finished: RunFinished | None = None +class EventKind(StrEnum): + """事件的种类。 + + **只有一种取值，但 `Event` 从第一天起就带 `kind`。** 加一个取值在类型上是兼容变更，可是 + 对一个「假定每条事件都是步事件」写出来的出口来说，新取值会被静默地当成步事件读。带上 + `kind` 逼每个出口从第一天起就分发，新取值那天它至少是显式地没被处理。 + + 加取值要走 `CLAUDE.md` §2 那道人类门，和改停止原因的取值同一档。 + """ + + #: 一步走完，而且是**在这次进程里真的走完的**。从日志里读回来直接跳过的步不发这条。 + STEP_FINISHED = "step_finished" + + @dataclass(frozen=True, slots=True, kw_only=True) class Event: """从事件出口发出去的一条事件。 - **字段还没定。** 事件集与具名回调清单要独立成一份 design doc；在那之前这个类型只有名字， - `EventSink.emit` 的签名不会因为它定下来而改变。 + 事件集定在 `research-wiki/design/0013-event-set-and-callbacks.md`。**它带的每一条事实在 + 存储里都另有一份**——这条不变量是「投递失败不打断循环」那条契约成立的前提，不然一个连不上 + 的后端会让一次运行的部分事实静默消失，而运行本身照常返回成功。 + + **没有 `schema_version`。** 事件不是持久化结构（`CLAUDE.md` §1.4 管的是会被下游存进数据库 + 或实验数据集的那些）；下游把它存下来时，存的是它自己那张表的 schema。真正需要版本的那部分 + 是步记录，它带着自己的 `schema_version` 一起进来。 """ + kind: EventKind + #: 一个出口可以被并发的多次运行共用，没有这个标识那些事件在出口那边混成一串。 + run_id: str + #: 项目自己那套标识，原样来自请求上的同名字段。**它不能从运行标识倒推**——一次业务会话 + #: 可能包含多次运行，两者不是一回事。 + model_binding: Mapping[str, str] + #: 整条步记录，不是挑几个字段拼的摘要。摘要是一次投影，而投影会漂移：步记录加一个字段， + #: 事件里自动就有，两边不可能对不上。 + step: StepRecord + # --------------------------------------------------------------------------- # 五个接缝 @@ -257,6 +288,7 @@ __all__ = [ "Decision", "DecisionParser", "Event", + "EventKind", "EventSink", "FinalAnswer", "InvalidDecision", diff --git a/src/polyloop/session/__init__.py b/src/polyloop/session/__init__.py index 29ffd77..c5fdeb3 100644 --- a/src/polyloop/session/__init__.py +++ b/src/polyloop/session/__init__.py @@ -8,9 +8,9 @@ 一起」，与一次运行怎么称呼是两件事——`session` 在业界普遍指一个长期存在、可以来回对话的 东西，而这里的治理单位是有界的、一次性的（`0006` 决策一）。 -**事件出口现在不发任何事件。** `Event` 还没有字段，事件集与具名回调清单要独立成一份 design -doc；在那之前发一条内容为空的事件既没用又会变成一份要兼容的形状。所以 `event_sink` 这个字段 -收下了但没有调用点，`RunResult.event_delivery_failures` 恒为 0。 +**事件只在一处发出去**：一步走完、`StepCompleted` 原子落地之后。顺序不能倒过来——先发后写的话， +进程崩在两者之间会让观察者看见一步而存储里没有，而事件流的全部安全性建立在「它带的事实在存储 +里另有一份」上（`0013` 决策一与决策五）。 """ import asyncio @@ -38,6 +38,8 @@ from polyloop.ports import ( Action, ActionExecutor, DecisionParser, + Event, + EventKind, EventSink, FinalAnswer, InvalidDecision, @@ -234,9 +236,10 @@ def _project_observation( 得见的东西」，而那种东西必须能进参数快照。执行器每次现造一段文本的话，两次运行之间它可以 变而不会有任何地方报错，于是「同一份配置跑出来的两次运行」在模型看来其实不同。 - 代价是执行器知道的细节丢了（「哪个参数不合法」只有它知道）。接受它，因为另一头的代价更 - 大；要补的话将来靠事件流把执行器原文送出去做审计——**进历史的东西必须可复现，进审计的 - 不必**。 + **执行器那段不进历史，但它没有丢**：「一步走完」那条记录落盘的是动作执行接缝的原样返回 + 值，替换只发生在步记录的这一列上。被拒绝那一档下，它是日志里唯一的拒绝说明（「哪个参数 + 不合法」只有执行器知道），所以不必再靠事件流把它送出去——事件可丢，而这段文本是审计要的 + （`0013` 决策六）。 """ if outcome.status is ActionStatus.NOT_EXECUTED: return synthetic.action_rejected, True, 0 @@ -256,13 +259,16 @@ class _Driver: frozen 的，共享它们没有问题。 """ - __slots__ = ("_counters", "_definition", "_request", "_steps") + __slots__ = ("_counters", "_definition", "_event_failures", "_request", "_steps") def __init__(self, definition: AgentDefinition, request: RunRequest) -> None: self._definition = definition self._request = request self._counters = RunCounters() self._steps: list[StepRecord] = [] + #: 投递失败的次数。**住在这里而不是出口上**——出口挂在定义上、可以被并发的多次运行 + #: 共用，而这个数属于一次运行的结果。 + self._event_failures = 0 # -- 写入 --------------------------------------------------------------- @@ -286,6 +292,33 @@ class _Driver: ) self._steps.append(step) self._counters = self._counters.with_step_appended() + await self._emit_step_finished(step) + + async def _emit_step_finished(self, step: StepRecord) -> None: + """把这一步发给事件出口。**只有走到这里的步才发**，从日志里读回来直接跳过的不发。 + + 补发已经完成的步等于宣称一件早就发生过的事刚刚发生，而接进度表的那一侧会多出一批 + 重复行。观察者要补全前半段，从存储里读。 + + **投递失败接住、计数、继续跑**：事件是观察通道不是控制通道，一次运行不该因为进度回写 + 的数据库连不上就终止。接的是 `Exception` 不是 `BaseException`——`CancelledError` 必须 + 原样穿过（`CLAUDE.md` §1.6），在这一下把取消吞掉，取消就会晚一整步才生效。 + + **失败不转成一条事件从同一个出口再发一次**：那会自我喂食，一个持续失败的出口会让失败 + 处理路径变成递归，而递归的表现是进程卡住或栈溢出，不是一条错误日志。 + """ + try: + await self._definition.event_sink.emit( + Event( + kind=EventKind.STEP_FINISHED, + run_id=self._request.run_id, + model_binding=self._request.model_binding, + step=step, + ) + ) + except Exception: + self._event_failures += 1 + logger.exception("运行 %s 第 %d 步的事件投递失败", self._request.run_id, step.step_idx) async def _finish(self, stop_reason: StopReason, final_answer: str | None = None) -> RunResult: """写结束标记，然后返回结果。 @@ -300,6 +333,7 @@ class _Driver: stop_reason=stop_reason, final_answer=final_answer, steps=tuple(self._steps), + event_delivery_failures=self._event_failures, ) await self._definition.store.write_run_finished( RunFinished(run_id=self._request.run_id, result=result) @@ -418,6 +452,7 @@ class _Driver: stop_reason=StopReason.CANCELLED, final_answer=None, steps=tuple(self._steps), + event_delivery_failures=self._event_failures, ) await self._drain_within_grace( asyncio.ensure_future( diff --git a/tests/contract/conftest.py b/tests/contract/conftest.py index 7357235..35d02e1 100644 --- a/tests/contract/conftest.py +++ b/tests/contract/conftest.py @@ -15,7 +15,7 @@ from collections.abc import Mapping import pytest -from polyloop.ports import Action, Event, ToolCall +from polyloop.ports import Action, Event, EventKind, ToolCall from polyloop.stores import JsonlRunStore from polyloop.types import ( ActionOutcome, @@ -155,8 +155,13 @@ class _Records: tool_call=None if tool_name is None else ToolCall(name=tool_name, arguments={}), ) - def event(self) -> Event: - return Event() + def event(self, *, run_id: str = "run-1", step_idx: int = 0) -> Event: + return Event( + kind=EventKind.STEP_FINISHED, + run_id=run_id, + model_binding={"item": "a"}, + step=self.step(step_idx=step_idx), + ) @pytest.fixture diff --git a/tests/contract/test_event_sink.py b/tests/contract/test_event_sink.py index f70cb2b..4e713b1 100644 --- a/tests/contract/test_event_sink.py +++ b/tests/contract/test_event_sink.py @@ -3,10 +3,11 @@ 两个已知形态差别在可靠性要求上：一个把进度逐步回写业务数据库供前端轮询（要求低延迟、 可以丢），一个把审计事件送进日志管道（要求不丢、可以慢）。 -## 写这份文件时撞出来的、`design/0006` 还答不上的问题 +## 写这份文件时撞出来的问题，`design/0013` 答了 -`Event` 只有一个名字，没有字段，所以这个接缝的契约现在只能验「投递失败不打断循环」这一半， -验不了「发出去的事件里有什么」。 +「发出去的事件里有什么」当时验不了，因为 `Event` 只有一个名字没有字段。现在事件集定下来了， +而答案把这份文件里的两条测试都挪走了——它们要断言的行为都在库那一侧，不在出口这一侧，见文末 +那两条说明。 """ import pytest @@ -32,28 +33,26 @@ def test_a_sink_is_allowed_to_raise_on_delivery_failure(): """ -@pytest.mark.xfail(reason="事件集还没定，见 docstring", strict=True) def test_failure_is_not_re_emitted_through_the_same_sink(): - """投递失败不再转成一条事件从同一个出口发出去。 + """投递失败不再转成一条事件从同一个出口发出去（`design/0013` 决策七）。 那会自我喂食：一个持续失败的出口会让失败处理路径变成递归，而递归的表现是进程卡住或 栈溢出，不是一条错误日志。 - **这条现在验不了**，因为验它要求能识别「这是一条失败事件」，而 `Event` 还没有字段—— - `design/0006` 里它只有一个名字。方向已经定了（观察走事件流、干预走具名回调），但事件 - 集与回调清单要独立成一份 design doc，这条要等到那时候。 + **要断言的是库有没有再发一次，那是整次运行的行为**，所以断言在 + `tests/unit/test_session.py` 里——那边用一个恒抛异常的出口跑完一次运行，验出口收到的 + 条数恰好等于步数。这个接缝自己看不到「库发了几次」。 """ - pytest.fail("Event 还没有字段，识别不了「失败事件」") -@pytest.mark.xfail(reason="事件集还没定，见 docstring", strict=True) -def test_audit_events_carry_both_raw_and_repaired_model_output(): - """审计事件要同时带模型原文与修复之后的结果。 +def test_audit_events_do_not_carry_the_model_output_the_store_does(): + """审计纪律由存储承担，不由事件流承担（`design/0013` 决策二）。 GovDoc 有一条硬纪律：agent 的原始输出、修复后的输出、恢复来源全程留痕，禁止静默修复。 - 它现有的审计出口是一个「发一条带类型和载荷的事件」的接口，迁移之后这条纪律要由事件流 - 承载——能不能承载，取决于事件里带不带这两样。 + 这条测试原来断言「事件要同时带原文与修复后的文本」，而那个前提是错的——事件流可丢， + 一件只存在于可丢通道里的事实撑不起「禁止静默修复」。 - 这是 `../research-wiki/migrations/govdoc-saas.md` 缺口登记里那一条，同样等事件集定下来。 + 两份文本在意图日志里各有位置：原文在模型调用结果记录的回复里，修复后的那份是步记录的 + `raw_output`。断言落在 `tests/unit/test_session.py`，因为要跑完一次完整运行再把日志读 + 回来，而这个接缝的契约只看得见一个出口实现。 """ - pytest.fail("Event 还没有字段，承载不了审计纪律") diff --git a/tests/unit/test_session.py b/tests/unit/test_session.py index 66962cf..48078ea 100644 --- a/tests/unit/test_session.py +++ b/tests/unit/test_session.py @@ -14,6 +14,7 @@ import pytest from polyloop._recovery import CorruptLogError from polyloop.ports import ( Action, + EventKind, FinalAnswer, InvalidDecision, ModelCall, @@ -46,6 +47,7 @@ from polyloop.types import ( RunFinished, RunStarted, StepCompleted, + StepRecord, StopReason, SyntheticObservations, TextBlock, @@ -191,12 +193,14 @@ def _outcome( ) -def _definition(store: FakeStore, model: FakeModel, parser: FakeParser) -> AgentDefinition: +def _definition( + store: FakeStore, model: FakeModel, parser: FakeParser, sink: object | None = None +) -> AgentDefinition: return AgentDefinition( model_client=model, decision_parser=parser, store=store, - event_sink=FakeSink(), + event_sink=sink or FakeSink(), # type: ignore[arg-type] synthetic_observations=SYNTHETIC, ) @@ -1030,3 +1034,229 @@ def test_an_empty_call_id_is_refused_at_construction() -> None: """空串是个看起来合法的键，连表时静默匹配不上，而 None 至少能被显式筛出来。""" with pytest.raises(ValueError, match="call_id"): ModelReply(call_id="", content="hi", thinking="") + + +# --------------------------------------------------------------------------- +# 事件出口 +# --------------------------------------------------------------------------- + + +class _RaisingSink(FakeSink): + """按脚本抛异常的出口。抛完仍然把这条记下来，好断言「库有没有再发一次」。""" + + def __init__(self, error: BaseException) -> None: + super().__init__() + self._error = error + + async def emit(self, event: object) -> None: + self.events.append(event) + raise self._error + + +async def test_every_step_emits_one_event_carrying_the_whole_record() -> None: + """一步一条，带的是整条步记录而不是挑几个字段拼的摘要。 + + 摘要是一次投影，而投影会漂移——步记录加一个字段，带整条的话事件里自动就有。 + """ + store = FakeStore() + sink = FakeSink() + model = FakeModel([_reply("go"), _reply("go")]) + definition = _definition(store, model, FakeParser({"go": ACT}), sink) + + result = await run(definition, _request(FakeExecutor([_outcome(), _outcome(completed=True)]))) + + assert [event.kind for event in sink.events] == [EventKind.STEP_FINISHED] * 2 + assert [event.step for event in sink.events] == list(result.steps) + + +async def test_the_event_carries_the_run_id_and_the_project_binding() -> None: + """运行标识让共用同一个出口的并发运行分得开；绑定没法从运行标识倒推。""" + store = FakeStore() + sink = FakeSink() + definition = _definition(store, FakeModel([_reply("go")]), FakeParser({"go": ACT}), sink) + + await run( + definition, + _request(FakeExecutor([_outcome(completed=True)]), binding={"item": "a", "task": "t7"}), + ) + + (event,) = sink.events + assert event.run_id == "run-1" + assert event.model_binding == {"item": "a", "task": "t7"} + + +async def test_the_event_goes_out_after_the_step_landed_not_before() -> None: + """先发后写的话，进程崩在两者之间会让观察者看见一步而存储里没有。 + + 事件流的全部安全性建立在「它带的事实在存储里另有一份」上，而这个顺序是那条不变量在 + 崩溃点上的兑现方式。 + """ + store = FakeStore() + seen_at_emit: list[int] = [] + + class _OrderSink(FakeSink): + async def emit(self, event: object) -> None: + seen_at_emit.append(len(store.of_type(StepCompleted))) + await super().emit(event) + + definition = _definition( + store, FakeModel([_reply("go")]), FakeParser({"go": ACT}), _OrderSink() + ) + + await run(definition, _request(FakeExecutor([_outcome(completed=True)]))) + + assert seen_at_emit == [1] + + +async def test_a_failing_sink_does_not_stop_the_run_and_is_counted() -> None: + """事件是观察通道不是控制通道：进度回写的数据库连不上，运行照跑完。 + + 计数放在返回值上而不是只记日志，因为日志没人看。 + """ + store = FakeStore() + sink = _RaisingSink(ConnectionError("进度库连不上")) + model = FakeModel([_reply("go"), _reply("go")]) + definition = _definition(store, model, FakeParser({"go": ACT}), sink) + + result = await run(definition, _request(FakeExecutor([_outcome(), _outcome(completed=True)]))) + + assert result.stop_reason is StopReason.TASK_COMPLETED + assert len(result.steps) == 2 + assert result.event_delivery_failures == 2 + + +async def test_a_delivery_failure_is_not_re_emitted_through_the_same_sink() -> None: + """失败不转成一条事件从同一个出口再发一次——那会自我喂食。 + + 一个持续失败的出口会让失败处理路径变成递归，而递归的表现是进程卡住或栈溢出，不是一条 + 错误日志。所以出口收到的条数必须恰好等于步数。 + """ + store = FakeStore() + sink = _RaisingSink(ConnectionError("一直连不上")) + definition = _definition(store, FakeModel([_reply("go")]), FakeParser({"go": ACT}), sink) + + result = await run(definition, _request(FakeExecutor([_outcome(completed=True)]))) + + assert len(sink.events) == len(result.steps) == 1 + + +async def test_cancellation_during_delivery_is_not_swallowed() -> None: + """接的是 `Exception` 不是 `BaseException`：在这一下吞掉取消，取消会晚一整步才生效。""" + store = FakeStore() + sink = _RaisingSink(asyncio.CancelledError()) + definition = _definition(store, FakeModel([_reply("go")]), FakeParser({"go": ACT}), sink) + + with pytest.raises(asyncio.CancelledError): + await run(definition, _request(FakeExecutor([_outcome(completed=True)]))) + + finished = store.of_type(RunFinished) + assert finished[0].result.stop_reason is StopReason.CANCELLED # type: ignore[attr-defined] + assert finished[0].result.event_delivery_failures == 0 # type: ignore[attr-defined] + + +async def test_steps_read_back_from_the_log_are_not_re_emitted() -> None: + """续跑不给已经完成的步补发事件。 + + 补发等于宣称一件早就发生过的事刚刚发生，而接进度表的那一侧会多出一批重复行。判据是这次 + 进程里有没有真的执行过，观察者要补全前半段就从存储里读。 + """ + sink = FakeSink() + store = FakeStore() + definition = _definition(store, FakeModel([_reply("go")]), FakeParser({"go": ACT}), sink) + request = _request(FakeExecutor([_outcome(completed=True)])) + store._log = _log_with_one_finished_step( # noqa: SLF001 + {**definition.parameter_snapshot(), **request.parameter_snapshot()} + ) + + result = await resume(definition, request) + + # 第 0 步是从日志里读回来的，第 1 步是这次进程里真的走的。只有后者发了事件。 + assert [step.step_idx for step in result.steps] == [0, 1] + assert [event.step.step_idx for event in sink.events] == [1] + + +def _log_with_one_finished_step(snapshot: Mapping[str, str]) -> RunLog: + """一份「第 0 步完整走完、还没写结束记录」的日志。 + + 手工搭而不是先跑一次再续跑：跑出来的那一步要么带完成信号（续跑会当场收尾，走不到第二步）， + 要么撞预算上限（续跑在预算准入那一档就停了），两种都验不到「读回来的不发、真跑的发」这条 + 边界。 + """ + outcome = _outcome() + return RunLog( + started=RunStarted(run_id="run-1", parameter_snapshot=snapshot), + intents=( + Intent( + run_id="run-1", + kind=IntentKind.MODEL_CALL, + call_index=0, + result_id="run-1#model#0", + replay_policy=ReplayPolicy.NEVER, + ), + Intent( + run_id="run-1", + kind=IntentKind.ACTION, + call_index=0, + result_id="run-1#action#0", + replay_policy=ReplayPolicy.NEVER, + ), + ), + model_results=( + ModelCallResult( + run_id="run-1", result_id="run-1#model#0", reply=_reply("go"), failure=None + ), + ), + steps=( + StepCompleted( + run_id="run-1", + result_id="run-1#action#0", + action_outcome=outcome, + step=StepRecord( + step_idx=0, + raw_output="go", + content_chars=2, + thinking_chars=0, + action="做点事", + parse_ok=True, + parse_error=None, + observation=outcome.observation, + observation_is_synthetic=False, + observation_truncated_chars=0, + prompt_chars=10, + call_id="c1", + step_wall_ms=1, + action_status=ActionStatus.EXECUTED, + env_reported_completion=False, + ), + ), + ), + ) + + +async def test_the_log_keeps_both_the_raw_and_the_repaired_model_output() -> None: + """模型原文与解释器改写之后的文本各有位置，两份都不可丢（`design/0013` 决策二）。 + + GovDoc 有一条硬纪律：agent 的原始输出、修复后的输出、恢复来源全程留痕，禁止静默修复。 + 承载它的是意图日志而不是事件流——事件可丢，一件只存在于可丢通道里的事实撑不起「禁止 + 静默修复」。 + + 两份文本天然分开存，是写入序列决定的：模型调用结算时写结果记录，那时还没解释；解释完、 + 动作走完之后才写步记录，那里面的文本是解释器交回来的。 + """ + + class _RewritingParser(FakeParser): + """把第一个代码围栏之后的内容整段丢掉——模型常在代码块后面编造执行结果。""" + + def parse(self, reply: ModelReply) -> ParsedReply: + return ParsedReply(history_text=reply.content.split("|", 1)[0], decision=ACT) + + store = FakeStore() + definition = _definition( + store, FakeModel([_reply("真动作|模型编的执行结果")]), _RewritingParser({}) + ) + + result = await run(definition, _request(FakeExecutor([_outcome(completed=True)]))) + + (call_result,) = store.of_type(ModelCallResult) + assert call_result.reply.content == "真动作|模型编的执行结果" # type: ignore[attr-defined,union-attr] + assert result.steps[0].raw_output == "真动作"