feat(session): 落成事件出口，五个接缝全部有调用点

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) <noreply@anthropic.com>

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；在那之前这个类型只有名字，
+    事件集定在 `research-wiki/design/0013-event-set-and-callbacks.md`。**它带的每一条事实在
-    `EventSink.emit` 的签名不会因为它定下来而改变。
+    存储里都另有一份**——这条不变量是「投递失败不打断循环」那条契约成立的前提，不然一个连不上
+    的后端会让一次运行的部分事实静默消失，而运行本身照常返回成功。
+
+    **没有 `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",

src/polyloop/session/__init__.py

@@ -8,9 +8,9 @@
 一起」，与一次运行怎么称呼是两件事——`session` 在业界普遍指一个长期存在、可以来回对话的
 东西，而这里的治理单位是有界的、一次性的（`0006` 决策一）。
 
-**事件出口现在不发任何事件。** `Event` 还没有字段，事件集与具名回调清单要独立成一份 design
+**事件只在一处发出去**：一步走完、`StepCompleted` 原子落地之后。顺序不能倒过来——先发后写的话，
-doc；在那之前发一条内容为空的事件既没用又会变成一份要兼容的形状。所以 `event_sink` 这个字段
+进程崩在两者之间会让观察者看见一步而存储里没有，而事件流的全部安全性建立在「它带的事实在存储
-收下了但没有调用点，`RunResult.event_delivery_failures` 恒为 0。
+里另有一份」上（`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(

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:
+    def event(self, *, run_id: str = "run-1", step_idx: int = 0) -> Event:
-        return Event()
+        return Event(
+            kind=EventKind.STEP_FINISHED,
+            run_id=run_id,
+            model_binding={"item": "a"},
+            step=self.step(step_idx=step_idx),
+        )
 
 
 @pytest.fixture

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` 里它只有一个名字。方向已经定了（观察走事件流、干预走具名回调），但事件
+    `tests/unit/test_session.py` 里——那边用一个恒抛异常的出口跑完一次运行，验出口收到的
-    集与回调清单要独立成一份 design doc，这条要等到那时候。
+    条数恰好等于步数。这个接缝自己看不到「库发了几次」。
     """
-    pytest.fail("Event 还没有字段，识别不了「失败事件」")
 
 
-@pytest.mark.xfail(reason="事件集还没定，见 docstring", strict=True)
+def test_audit_events_do_not_carry_the_model_output_the_store_does():
-def test_audit_events_carry_both_raw_and_repaired_model_output():
+    """审计纪律由存储承担，不由事件流承担（`design/0013` 决策二）。
-    """审计事件要同时带模型原文与修复之后的结果。
 
     GovDoc 有一条硬纪律：agent 的原始输出、修复后的输出、恢复来源全程留痕，禁止静默修复。
-    它现有的审计出口是一个「发一条带类型和载荷的事件」的接口，迁移之后这条纪律要由事件流
+    这条测试原来断言「事件要同时带原文与修复后的文本」，而那个前提是错的——事件流可丢，
-    承载——能不能承载，取决于事件里带不带这两样。
+    一件只存在于可丢通道里的事实撑不起「禁止静默修复」。
 
-    这是 `../research-wiki/migrations/govdoc-saas.md` 缺口登记里那一条，同样等事件集定下来。
+    两份文本在意图日志里各有位置：原文在模型调用结果记录的回复里，修复后的那份是步记录的
+    `raw_output`。断言落在 `tests/unit/test_session.py`，因为要跑完一次完整运行再把日志读
+    回来，而这个接缝的契约只看得见一个出口实现。
     """
-    pytest.fail("Event 还没有字段，承载不了审计纪律")

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 == "真动作"