From fc57ed0e3234f1a1ae0cab066f1bfd8e26b43e8d Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Mon, 5 Oct 2026 01:18:03 +0800 Subject: [PATCH] =?UTF-8?q?=E8=BF=BD=E5=8A=A0=E4=BA=8B=E4=BB=B6=E6=94=B6?= =?UTF-8?q?=E7=B4=A7=E5=87=A0=E5=A4=84=EF=BC=8C=E5=B0=81=E5=AD=98=E5=90=8E?= =?UTF-8?q?=E8=BF=BD=E5=8A=A0=E7=9A=84=E8=B4=A8=E8=AF=A2=E4=B8=8D=E8=AE=A1?= =?UTF-8?q?=E8=BF=9B=E8=BF=90=E8=A1=8C=E7=94=A8=E9=87=8F=E3=80=82=E5=A4=8D?= =?UTF-8?q?=E6=A0=B8=E6=8F=90=E5=87=BA=EF=BC=9A=E5=81=9C=E5=9C=A8=E5=AE=A1?= =?UTF-8?q?=E6=89=B9=E6=97=B6=E8=BF=BD=E5=8A=A0=E5=8F=AA=E7=9C=8B=E8=BF=90?= =?UTF-8?q?=E8=A1=8C=E8=BF=98=E5=81=9C=E7=9D=80=EF=BC=8C=E5=85=B3=E4=BA=8E?= =?UTF-8?q?=E7=AC=AC=E4=B8=80=E5=BC=A0=E5=AE=A1=E6=89=B9=E5=8D=A1=E7=9A=84?= =?UTF-8?q?=E8=B4=A8=E8=AF=A2=E9=97=AE=E5=AE=8C=E6=97=B6=E8=BF=90=E8=A1=8C?= =?UTF-8?q?=E5=8F=AF=E8=83=BD=E5=B7=B2=E7=BB=8F=E6=89=B9=E8=BF=87=E5=AE=83?= =?UTF-8?q?=E3=80=81=E5=81=9C=E5=88=B0=E4=BA=86=E7=AC=AC=E4=BA=8C=E5=BC=A0?= =?UTF-8?q?=E4=B8=8A=EF=BC=8C=E9=97=AE=E7=AD=94=E7=85=A7=E6=A0=B7=E5=86=99?= =?UTF-8?q?=E8=BF=9B=E5=8E=BB=EF=BC=9B=E6=B2=A1=E6=9C=89=E7=BB=88=E6=80=81?= =?UTF-8?q?=E4=BA=8B=E4=BB=B6=E7=9A=84=E8=80=81=E5=B0=81=E5=AD=98=E8=BF=90?= =?UTF-8?q?=E8=A1=8C=E8=BF=BD=E5=8A=A0=E4=B9=8B=E5=90=8E=20verify=20?= =?UTF-8?q?=E7=94=B1=E4=B8=80=E8=87=B4=E5=8F=98=E6=88=90=E4=B8=8D=E4=B8=80?= =?UTF-8?q?=E8=87=B4=EF=BC=9B=E8=BF=94=E5=9B=9E=E7=BB=99=E8=B0=83=E7=94=A8?= =?UTF-8?q?=E6=96=B9=E7=9A=84=E4=BA=8B=E4=BB=B6=20id=20=E5=92=8C=E5=BA=93?= =?UTF-8?q?=E9=87=8C=E7=9A=84=E4=B8=8D=E4=B8=80=E6=A0=B7=EF=BC=9B=E6=B5=8B?= =?UTF-8?q?=E8=AF=95=E8=AF=81=E6=98=8E=E4=B8=8D=E4=BA=86=E5=85=88=E8=90=BD?= =?UTF-8?q?=E5=BA=93=E5=86=8D=E5=B9=BF=E6=92=AD=EF=BC=9B=E5=B0=81=E5=AD=98?= =?UTF-8?q?=E5=90=8E=E8=BF=BD=E5=8A=A0=E7=9A=84=E8=B4=A8=E8=AF=A2=E5=B9=B3?= =?UTF-8?q?=E6=97=B6=E4=B8=8D=E8=AE=A1=E7=94=A8=E9=87=8F=EF=BC=8C=E7=BB=A7?= =?UTF-8?q?=E7=BB=AD=E8=BF=90=E8=A1=8C=E4=B9=8B=E5=90=8E=E5=8F=88=E8=A2=AB?= =?UTF-8?q?=E8=AE=A1=E8=BF=9B=E5=8E=BB=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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，对应的测试各自变红 --- backend/app/engine/run_append.py | 40 +++++--- backend/app/engine/runner.py | 13 ++- backend/tests/test_inquiry_usage.py | 31 +++++++ backend/tests/test_run_append.py | 139 +++++++++++++++++++++++++++- 4 files changed, 205 insertions(+), 18 deletions(-) diff --git a/backend/app/engine/run_append.py b/backend/app/engine/run_append.py index f7a515ad..70de2784 100644 --- a/backend/app/engine/run_append.py +++ b/backend/app/engine/run_append.py @@ -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 diff --git a/backend/app/engine/runner.py b/backend/app/engine/runner.py index 15b1c0a6..a1dd51f8 100644 --- a/backend/app/engine/runner.py +++ b/backend/app/engine/runner.py @@ -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): diff --git a/backend/tests/test_inquiry_usage.py b/backend/tests/test_inquiry_usage.py index c29dfd43..d3f28f8e 100644 --- a/backend/tests/test_inquiry_usage.py +++ b/backend/tests/test_inquiry_usage.py @@ -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): """停在审批时放弃：停下时记的用量里还没有这次停下期间的质询，放弃时补上；上一次停下时已经计过的不再计。""" diff --git a/backend/tests/test_run_append.py b/backend/tests/test_run_append.py index 1052fef4..e73d5f8d 100644 --- a/backend/tests/test_run_append.py +++ b/backend/tests/test_run_append.py @@ -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