追加事件收紧几处，封存后追加的质询不计进运行用量。复核提出：停在审批时追加只看运行还停着，关于第一张审批卡的质询问完时运行可能已经批过它、停到了第二张上，问答照样写进去；没有终态事件的老封存运行追加之后 verify 由一致变成不一致；返回给调用方的事件 id 和库里的不一样；测试证明不了先落库再广播；封存后追加的质询平时不计用量，继续运行之后又被计进去。

- append_events 加 approval_id（只用于 interrupted）：问的是哪张审批卡，那张卡仍待审批才写。核对放在和其余条件同一次条件更新里（EXISTS 待审批的那条），批过了、不是这次运行的卡都不写、返回 None。条件更新写明不同步会话里的运行：条件里带了 EXISTS 时 SQLAlchemy 会悄悄换一种同步办法、给语句加上 RETURNING - manifest_seq 为空的老封存运行，封到哪一条是 verify_manifest 按事件类型推出来的：有终态事件的封到最后一条终态事件，追加在后面不碍事，照旧追加；没有终态事件的封到最后一行，追加一条范围就挪到它身上，我们自己写的批注被当成篡改，现在不追加 - _stage 落库时用事件自己的 id：广播出去的、交回给调用方的和库里的是同一个 id - 运行用量不计数据带 post_seal 的质询：运行用量是这次运行自己花的，结束时就定下了、写进了终态事件；封存后的批注另有自己的账，同按需裁判的判定。继续运行之后它落进新的封存范围，照样不计测试：两道审批的运行批过第一道、停在第二道，钉住第一道的质询不写、钉住第二道的写，读完运行之后那张卡被批了也不写，返回的 id 和序号与落库的一致；广播时查库，每条都已落库，落空的一条都不广播（把广播挪到条件更新之前这条就变红，以前的测试照样通过）；老封存三种（无终态、run.finished、run.failed）追加前后 verify 都一致，无终态的不写；失败封存的运行追加 post_seal 质询后继续运行，运行用量里没有它。去掉审批卡的核对、改回老封存一律追加、改回计 post_seal、去掉落库 id，对应的测试各自变红
2026-11-15 21:12:01 +00:00 · 2026-10-05 01:18:03 +08:00
parent f8b2857630
commit fc57ed0e32
4 changed files with 205 additions and 18 deletions
@@ -9,8 +9,8 @@ require 说明往哪种运行里追加：
 - "interrupted"：运行停在审批、还没封存。事件落在以后的封存范围里，随运行结束一起封存。
 - "sealed"：已经封存的终态运行。事件带 post_seal: true，落在 manifest_seq 之后，不进封存范围，verify 照旧一致。

-没抢到（读完运行之后它被恢复、被放弃、继续运行后重新封存了，或者别处刚往里写过事件）就一条都不写、返回 None，
-调用方据此说明「未写入运行记录」。
+没抢到（读完运行之后它被恢复、被放弃、继续运行后重新封存了，问的那张审批卡已经批过了，或者别处刚往里写过
+事件）就一条都不写、返回 None，调用方据此说明「未写入运行记录」。
 """
 from __future__ import annotations

@@ -19,12 +19,12 @@ import contextlib
 from collections.abc import AsyncIterator, Sequence
 from typing import Any, Literal

-from sqlalchemy import func, select, update
+from sqlalchemy import exists, func, select, update

 from app.core.bus import bus
 from app.core.events import EventType, RunEventModel
 from app.db.base import SessionLocal
-from app.db.models import Run, RunEvent
+from app.db.models import Approval, Run, RunEvent
 from app.engine.runner import _seq_floor, _stage, run_manager

 #: 运行的终态。封存过的运行只会停在这几种上
@@ -66,15 +66,21 @@ async def append_events(
    *,
    require: Literal["interrupted", "sealed"],
    seal: tuple[int | None, str | None] | None = None,
+    approval_id: str | None = None,
 ) -> list[RunEventModel] | None:
-    """把 items 依次接在运行记录的最后一条后面，返回写入的事件（seq 严格递增）；没写入返回 None。
+    """把 items 依次接在运行记录的最后一条后面，返回写入的事件（seq 严格递增，id 和落库的相同）；没写入返回 None。

-    seal 只用于 "sealed"：调用方开始做事时看到的 (manifest_seq, manifest_hash)。做的事要花几秒（问模型），
-    这期间继续运行、重新封存了的话，结果依据的已经不是现在的封存，不追加。不传就按这次读到的封存。
+    调用方开始做事时看到的状态可以钉住，做的事要花几秒（问模型），这期间变了就不追加：
+    - seal 只用于 "sealed"：当时的 (manifest_seq, manifest_hash)。继续运行、重新封存了的话，结果依据的已经
+      不是现在的封存。不传就按这次读到的封存。
+    - approval_id 只用于 "interrupted"：问的是哪张审批卡。那张卡仍待审批才写——批过了，运行可能已经停到了
+      下一张卡上，仍是「停在审批」，关于上一张的问答却会写进这一次停下里。
    items 为空时什么都不做，返回空列表。
    """
    if require not in ("interrupted", "sealed"):
        raise ValueError(f"require 只能是 interrupted 或 sealed，收到的是 {require!r}")
+    if approval_id is not None and require != "interrupted":
+        raise ValueError("approval_id 只用于 interrupted：封存之后没有待审批的卡")
    # 事件类型先验过：写错了在取号之前就报出来
    entries = [(EventType(str(kind)), node_id, dict(data or {})) for kind, node_id, data in items]
    if not entries:
@@ -103,16 +109,21 @@ async def append_events(
                if run.status != "interrupted":
                    return None
                guard = (Run.status == "interrupted",)
+                if approval_id is not None:
+                    guard += (exists().where(Approval.id == approval_id, Approval.run_id == run_id,
+                                             Approval.status == "pending"),)
            last = run.last_seq
            # 进程重启过的话内存里的序号从 0 起：接在库里最后一条后面。last_seq 按说就是最后一条，_seq_floor
            # 另取一次实际的最大序号兜底——别处的追加把 last_seq 写小了，也不至于在这里撞号
            bus.set_seq(run_id, await _seq_floor(session, run_id, last))
            events = [run_manager._event(run_id, kind, node_id=node_id, data=data)
                      for kind, node_id, data in entries]
-            # 核对和写入是同一行上的条件更新，和恢复、放弃、继续运行（runner._claim / _abandon）抢的是同一行
+            # 核对和写入是同一行上的条件更新，和恢复、放弃、继续运行（runner._claim / _abandon）抢的是同一行。
+            # 不同步会话里的 run（下面 _stage 会把它的 last_seq 改成同一个值）：不写明的话，条件里带了 EXISTS 时
+            # SQLAlchemy 会悄悄换一种同步办法、给这条语句加上 RETURNING
            claimed = await session.execute(
                update(Run).where(Run.id == run_id, _same(Run.last_seq, last), *guard)
-                .values(last_seq=events[-1].seq))
+                .values(last_seq=events[-1].seq).execution_options(synchronize_session=False))
            if claimed.rowcount != 1:
                # 读完运行之后它变了。烧掉几个序号不要紧：不落库的事件（流式增量）本来就占序号，
                # 序号只要求递增、不撞号
@@ -135,12 +146,15 @@ async def _current_seal(session: Any, run: Run) -> bool:
    """运行记录上的封存是不是这次结束之后封的：manifest_seq 盖住了最后一条宣布结束的事件。

    继续运行之后再次结束时，终态和终态事件先提交、封存后算。这中间运行记录上还是上一次结束时的封存，
-    当成现在的封存去追加，事件会被接下来的封存收进去，post_seal 就成了假话。manifest_seq 为空的是加这一列
-    之前封存的老运行，当时的封存范围按 verify_manifest 的老口径算，这里不拦。
+    当成现在的封存去追加，事件会被接下来的封存收进去，post_seal 就成了假话。
+
+    manifest_seq 为空的是加这一列之前封存的老运行，封到哪一条是 verify_manifest 按老口径推出来的：有终态事件的
+    封到最后一条终态事件，追加在后面不碍事；没有的封到最后一行，追加一条，范围就挪到它身上，我们自己写的
+    批注被当成篡改。所以老运行要有终态事件才追加。
    """
-    if run.manifest_seq is None:
-        return True
    ended = await session.scalar(
        select(func.max(RunEvent.seq)).where(RunEvent.run_id == run.id,
                                             RunEvent.type.in_([str(t) for t in _ENDINGS])))
+    if run.manifest_seq is None:
+        return ended is not None
    return ended is None or run.manifest_seq >= ended
@@ -1263,8 +1263,11 @@ async def _claim(


 async def _stage(session: Any, event: RunEventModel, run: Run | None = None) -> None:
-    """把一条事件写进这个 session，随它一起提交。run 已经在 session 里时直接改它的 last_seq。"""
-    session.add(RunEvent(run_id=event.run_id, seq=event.seq, type=str(event.type),
+    """把一条事件写进这个 session，随它一起提交。run 已经在 session 里时直接改它的 last_seq。
+
+    落库的 id 用事件自己的：广播出去的、交回给调用方的和库里的是同一个 id，回头按 id 找得到。
+    """
+    session.add(RunEvent(id=event.id, run_id=event.run_id, seq=event.seq, type=str(event.type),
                         node_id=event.node_id, ts=event.ts, data=event.data))
    if run is not None:
        run.last_seq = event.seq
@@ -1314,6 +1317,10 @@ async def _usage_of_events(session: Any, run_id: str) -> dict[str, Any]:
    llm.end 逐次加总：每次模型调用都有一条，不管节点最后成没成。审批质询不在图里执行、不发 llm.end，
    它的用量写在 approval.inquiry 事件数据的 usage 里，单独加总放在 inquiry 下（没有质询就没有这个键）：
    llm.end 这本账要和状态里的用量比大小（_spent），质询花的两本里都没有，混进去比会被状态那本盖掉。
+
+    封存之后追加的质询（数据带 post_seal）不算：运行用量是这次运行自己花的，结束时就定下了、写进了终态事件；
+    事后的批注另有自己的账，同按需裁判的判定。继续运行之后它落进新的封存范围，照样不算——不然同一条质询
+    计不计，全看之后有没有继续运行。
    """
    rows = await session.execute(
        select(RunEvent.type, RunEvent.data).where(
@@ -1327,7 +1334,7 @@ async def _usage_of_events(session: Any, run_id: str) -> dict[str, Any]:
    for kind, data in rows:
        if kind == EventType.APPROVAL_INQUIRY:
            spent = (data or {}).get("usage")
-            if isinstance(spent, dict):
+            if isinstance(spent, dict) and not (data or {}).get("post_seal"):
                for key in _INQUIRY_KEYS:
                    value = spent.get(key)
                    if isinstance(value, (int, float)) and not isinstance(value, bool):
@@ -7,6 +7,7 @@
 - 运行结束（成功、失败）时：图的用量照旧（成功取状态、其余两本账取大），再加上质询的；明细另放在 inquiry 下
 - 停在审批时被放弃：停下时记的用量里还没有这次停下期间的质询，放弃时补上
 - 停过两次、两次都问过：每次质询只计一遍
+- 封存之后追加的质询（post_seal）不计进运行用量，继续运行、重新封存之后也不计
 - 没有质询的运行，用量和以前逐字相同
 """
 from __future__ import annotations
@@ -213,6 +214,36 @@ async def test_each_inquiry_counts_once_across_two_pauses():
                                    "calls": 3, "unpriced_calls": 1}


+async def test_post_seal_inquiries_stay_out_of_the_run_usage(monkeypatch):
+    """运行用量是这次运行自己花的，结束时就定下了、写进了封存范围里的终态事件。封存之后追加的质询是事后的批注，
+    和按需裁判的判定一样另有自己的账，不计进来。以前平时不计，继续运行、重新算用量时又被计了进去：同一条质询
+    计不计，全看之后有没有继续运行。"""
+    from app.engine.run_append import append_events
+
+    original = compiler.RUNNERS[NodeType.TRANSFORM]
+    state = {"fail": True}
+
+    async def flaky(st, ctx):
+        if ctx.node.id == "boom" and state["fail"]:
+            raise NodeError(ctx.node.id, "数据源暂时不可用")
+        return await original(st, ctx)
+
+    monkeypatch.setitem(compiler.RUNNERS, NodeType.TRANSFORM, flaky)
+    graph = chain(node("start", "input"), node("boom", "transform", mode="template", template="x"),
+                  node("out", "output", fields=[{"name": "r", "value": "x"}]))
+    run = await run_manager.start(graph=graph, input_payload={})
+    failed = await _until(run.id, ("failed",))
+    written = await append_events(run.id, [ask("这次为什么失败", A)], require="sealed")
+    assert written and written[0].data["post_seal"] is True
+    assert set(failed.usage) == {"duration_ms", "wall_ms", "active_ms", "wait_ms"}
+
+    state["fail"] = False
+    await run_manager.continue_failed(run.id)
+    row = await _until(run.id, ("succeeded",))
+    assert row.manifest_seq > written[0].seq, "继续运行之后重新封存，这条质询落进了新的封存范围"
+    assert set(row.usage) == {"duration_ms", "wall_ms", "active_ms", "wait_ms"}
+
+
@pytest.mark.parametrize("asked_before", [False, True], ids=["first-pause", "second-pause"])
 async def test_abandoning_a_paused_run_counts_what_was_asked(asked_before):
    """停在审批时放弃：停下时记的用量里还没有这次停下期间的质询，放弃时补上；上一次停下时已经计过的不再计。"""
@@ -3,8 +3,10 @@
 - 停在审批时追加：事件接在最后一条后面、落进以后的封存范围。批准、运行结束封存之后 verify 照旧一致，
  改动追加的事件 verify 就对不上——它们确实在封存范围里
 - 追加途中运行被恢复（条件更新落空）：什么都不写，返回 None，last_seq 不动
+- 质询指明问的是哪张审批卡（approval_id）时，那张卡仍待审批才写；返回的事件和落库的 id、序号一样
+- 先落库再广播，没写入的一条都不广播
 - 已封存的运行追加：带 post_seal、落在 manifest_seq 之后，verify 照旧一致；封存不是这次结束之后封的、
-  封存和调用方开始时看到的不同，都不追加
+  封存和调用方开始时看到的不同、没有终态事件的老封存，都不追加
 - 连续追加、并发追加，序号严格递增，整条事件流没有重复的序号
 - last_seq 落后于库里实际的最后一条（按需裁判的追加和封存后追加交错提交时会这样）：进程重启后恢复、
  继续运行、放弃、按需裁判都接在实际的最后一条后面，不撞号；runner 记事件不把 last_seq 写小
@@ -26,7 +28,7 @@ from app.core.bus import bus
 from app.core.config import settings
 from app.core.events import EventType
 from app.db.base import SessionLocal
-from app.db.models import Run, RunEvent
+from app.db.models import Approval, Run, RunEvent
 from app.engine import compiler
 from app.engine.context import NodeError
 from app.engine.runner import RunManager, run_manager, verify_manifest
@@ -68,6 +70,10 @@ GATED = chain(node("start", "input"),
              node("gate", "human", mode="approve", title="这份周报可以出具吗"),
              node("out", "output", fields=[{"name": "r", "value": "{{ nodes.work }}"}]))
 SIMPLE = chain(node("start", "input"), node("out", "output", fields=[{"name": "r", "value": "完成"}]))
+TWO_GATES = chain(node("start", "input"),
+                  node("g1", "human", mode="approve", title="第一道：口径对吗"),
+                  node("g2", "human", mode="approve", title="第二道：可以出具吗"),
+                  node("out", "output", fields=[{"name": "r", "value": "完成"}]))
 #: 「boom」节点按 flaky 的开关失败：先让运行失败、封存，再关掉开关继续运行
 FLAKY = chain(node("start", "input"), node("boom", "transform", mode="template", template="门店 17 家"),
              node("out", "output", fields=[{"name": "r", "value": "{{ nodes.boom }}"}]))
@@ -148,6 +154,12 @@ async def _no_repeated_seq(run_id: str) -> None:
    assert len(seqs) == len(set(seqs)), f"事件流里有重复的序号：{seqs}"


+async def _pending(run_id: str) -> list[Approval]:
+    async with SessionLocal() as session:
+        return list((await session.execute(
+            select(Approval).where(Approval.run_id == run_id, Approval.status == "pending"))).scalars())
+
+
 def _sql(statement: str, *params) -> None:
    """另开一条连接改库：模拟别处（恢复、继续运行、另一个写入者）抢在中间提交。"""
    db = sqlite3.connect(settings.db_path, timeout=5)
@@ -285,6 +297,89 @@ async def test_a_run_that_already_moved_on_gets_nothing(executed):
    assert await _events(paused.id, "approval.inquiry") == []


+async def test_an_inquiry_about_an_approval_needs_that_approval_still_pending(monkeypatch):
+    """质询问的是某一张审批卡：开始问模型时那张卡还在等，问完时运行可能已经批过它、停到了下一张上。只看
+    「运行停在审批」的话，关于第一张的问答会写进第二次停下里。传 approval_id 时那张卡仍待审批才写，
+    和其余条件在同一次条件更新里核对。"""
+    from app.engine.run_append import append_events
+
+    paused = await _paused(TWO_GATES)
+    [first] = await _pending(paused.id)
+    await run_manager.resume(paused.id, {"approved": True}, approval_id=first.id)
+    await asyncio.sleep(0.05)
+    later = await _until(paused.id, ("interrupted",))
+    [second] = await _pending(paused.id)
+    assert (first.node_id, second.node_id) == ("g1", "g2")
+
+    assert await append_events(paused.id, [inquiry("第一道的口径")], require="interrupted",
+                               approval_id=first.id) is None
+    assert await append_events(paused.id, [inquiry("不存在的卡")], require="interrupted",
+                               approval_id="no-such-approval") is None
+    assert (await _row(paused.id)).last_seq == later.last_seq
+    assert await _events(paused.id, "approval.inquiry") == []
+
+    written = await append_events(paused.id, [inquiry("出具前再确认一次")], require="interrupted",
+                                  approval_id=second.id)
+    assert written is not None
+    # 返回的就是落了库的那几条：调用方按 id 或序号回头找，两边对得上
+    stored = await _events(paused.id, "approval.inquiry")
+    assert [(e.id, e.seq) for e in stored] == [(e.id, e.seq) for e in written]
+
+    # 读完运行、写入之前那张卡被人批了（并行的审批还没凑齐，运行仍停着）：不写
+    set_seq = bus.set_seq
+
+    def answered_in_between(run_id, value):
+        if run_id == paused.id:
+            _sql("UPDATE approvals SET status = 'answered' WHERE id = ?", second.id)
+        set_seq(run_id, value)
+
+    monkeypatch.setattr(bus, "set_seq", answered_in_between)
+    assert await append_events(paused.id, [inquiry("批完才问到的")], require="interrupted",
+                               approval_id=second.id) is None
+    assert (await _row(paused.id)).last_seq == written[-1].seq
+    assert len(await _events(paused.id, "approval.inquiry")) == 1
+    with pytest.raises(ValueError):
+        await append_events(paused.id, [inquiry("封存后的卡")], require="sealed", approval_id=second.id)
+
+
+async def test_published_only_once_stored_and_never_when_not_written(monkeypatch):
+    """先落库再广播：在线的客户端收到的每一条库里都已经有了（订阅者靠「先读历史、再接实时」拼完整，见
+    RunManager._emit）。没写进运行记录的一条都不广播，否则界面上会多出一条运行记录里没有的问答。"""
+    from app.engine.run_append import append_events
+
+    published: list[tuple[int, int]] = []
+    publish = bus.publish
+
+    async def stored_when_published(event):
+        if event.type == EventType.APPROVAL_INQUIRY:
+            db = sqlite3.connect(settings.db_path, timeout=5)
+            try:
+                (count,) = db.execute("SELECT count(*) FROM run_events WHERE run_id = ? AND seq = ?",
+                                      (event.run_id, event.seq)).fetchone()
+            finally:
+                db.close()
+            published.append((event.seq, count))
+        await publish(event)
+
+    monkeypatch.setattr(bus, "publish", stored_when_published)
+    paused = await _paused()
+    written = await append_events(paused.id, [inquiry("客单价怎么算的"), inquiry("含不含退款")],
+                                  require="interrupted")
+    assert published == [(e.seq, 1) for e in written]
+
+    published.clear()
+    set_seq = bus.set_seq
+
+    def resumed_in_between(run_id, value):
+        if run_id == paused.id:
+            _sql("UPDATE runs SET status = 'running' WHERE id = ?", paused.id)
+        set_seq(run_id, value)
+
+    monkeypatch.setattr(bus, "set_seq", resumed_in_between)
+    assert await append_events(paused.id, [inquiry("恢复之后才到的")], require="interrupted") is None
+    assert published == []
+
+
 # --------------------------------------------------------------------------
 # 封存之后追加
 # --------------------------------------------------------------------------
@@ -393,6 +488,46 @@ async def test_a_seal_left_from_an_earlier_ending_is_not_the_seal(monkeypatch):
    assert (await verify_manifest(first.id))["ok"] is True


+async def _legacy_sealed(ending: str | None) -> Run:
+    """加 manifest_seq 这一列之前封存的老运行：只记了清单哈希，封到哪一条要由 verify_manifest 按事件类型推。"""
+    from app.core.artifact_store import manifest_hash
+
+    rows = [(1, "run.started", None, {"nodes": 2}), (2, "node.started", "out", {}),
+            (3, "node.finished", "out", {"output": {"r": "完成"}})]
+    if ending:
+        rows.append((4, ending, None, {}))
+    # 老口径：run.finished 在算哈希之后才落库，不在清单里；run.failed / run.cancelled 在
+    covered = rows[:-1] if ending == "run.finished" else rows
+    async with SessionLocal() as session:
+        run = Run(workflow_name="老运行·月报", status="failed" if ending == "run.failed" else "succeeded",
+                  graph={}, input={}, last_seq=rows[-1][0], manifest_hash=manifest_hash(covered), manifest_seq=None)
+        session.add(run)
+        await session.flush()
+        for seq, kind, node_id, data in rows:
+            session.add(RunEvent(run_id=run.id, seq=seq, type=kind, node_id=node_id, ts=float(seq), data=data))
+        await session.commit()
+        return run
+
+
+@pytest.mark.parametrize("ending", [None, "run.finished", "run.failed"], ids=["no-ending", "finished", "failed"])
+async def test_a_legacy_seal_takes_appends_only_when_its_range_does_not_move(ending):
+    """老封存的范围是推出来的：有终态事件的，封到最后一条终态事件为止，追加在后面不进范围；没有终态事件的，
+    封到最后一行——追加一条，范围就跟着挪到它身上，我们自己写的批注被当成篡改。这种不追加。"""
+    from app.engine.run_append import append_events
+
+    legacy = await _legacy_sealed(ending)
+    before = await verify_manifest(legacy.id)
+    assert before["ok"] is True and before["legacy"] is True
+    written = await append_events(legacy.id, [inquiry("这期为什么少了一家店")], require="sealed")
+    if ending is None:
+        assert written is None
+        assert (await _row(legacy.id)).last_seq == legacy.last_seq
+    else:
+        assert written is not None and written[0].data["post_seal"] is True
+    after = await verify_manifest(legacy.id)
+    assert after["ok"] is True and (after["sealed_at"], after["events"]) == (before["sealed_at"], before["events"])
+
+
 async def test_the_wrong_kind_of_run_gets_nothing():
    from app.engine.run_append import append_events