diff --git a/backend/app/core/events.py b/backend/app/core/events.py index 95c5211a..b23b4936 100644 --- a/backend/app/core/events.py +++ b/backend/app/core/events.py @@ -65,6 +65,10 @@ class EventType(StrEnum): # 人工介入 HUMAN_REQUESTED = "human.requested" HUMAN_RESOLVED = "human.resolved" + # 审批质询:运行停在审批时,审批人就待审的报告问的一个问题和得到的回答(engine/run_append.py 追加, + # node_id 为空)。这次问答的用量写在事件数据的 usage 里,计进运行用量(runner._usage_of_events)。 + # 停在审批时追加的随运行结束一起封存;封存之后追加的带 post_seal: true,是批注不是证据 + APPROVAL_INQUIRY = "approval.inquiry" # 出具与口径治理 ISSUANCE = "issuance" # 出具契约的判定结果(档位 / 缺数据 / 未回指数字) diff --git a/backend/app/engine/run_append.py b/backend/app/engine/run_append.py new file mode 100644 index 00000000..f4f20be5 --- /dev/null +++ b/backend/app/engine/run_append.py @@ -0,0 +1,147 @@ +"""往一次停下的运行里追加事件:运行停在审批时的审批质询、封存之后的批注。 + +事件流是封存的依据:清单按 seq 排序算哈希,而 run_events 的 (run_id, seq) 没有唯一约束——两条事件撞了号, +排序就说不准,封存核对时对时不对。所以追加和 runner 发事件守同一套序号:接在库里最后一条后面,从 bus 取号 +(同一进程里和执行任务共用一个计数器),写事件和条件更新 last_seq 在同一个事务里,谁先提交算谁的。做法同 +api/evidence.py 的 _append_judged(按需裁判的判定)。 + +require 说明往哪种运行里追加: +- "interrupted":运行停在审批、还没封存。事件落在以后的封存范围里,随运行结束一起封存。 +- "sealed":已经封存的终态运行。事件带 post_seal: true,落在 manifest_seq 之后,不进封存范围,verify 照旧一致。 + +没抢到(读完运行之后它被恢复、被放弃、继续运行后重新封存了,或者别处刚往里写过事件)就一条都不写、返回 None, +调用方据此说明「未写入运行记录」。 +""" +from __future__ import annotations + +import asyncio +import contextlib +from collections.abc import AsyncIterator, Sequence +from typing import Any, Literal + +from sqlalchemy import 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.engine.runner import _stage, run_manager + +#: 运行的终态。封存过的运行只会停在这几种上 +_TERMINAL = ("succeeded", "failed", "cancelled") +#: 宣布运行结束的事件。封存盖住了最后一条这种事件,才是这次结束之后封的(见 _current_seal) +_ENDINGS = (EventType.RUN_FINISHED, EventType.RUN_FAILED, EventType.RUN_CANCELLED) + +#: 一条要追加的事件:(事件类型, node_id, 数据) +Item = tuple[EventType | str, str | None, dict[str, Any] | None] + + +class _RunLocks: + """每次运行一把锁,没人用了就扔:同一次运行的追加在本进程里排队,读 last_seq 到条件更新之间不会被 + 另一次追加插进来——插进来的话后提交的那次条件更新落空,明明没有冲突也成了「未写入」。""" + + def __init__(self) -> None: + self._locks: dict[str, asyncio.Lock] = {} + self._users: dict[str, int] = {} + + @contextlib.asynccontextmanager + async def hold(self, key: str) -> AsyncIterator[None]: + lock = self._locks.setdefault(key, asyncio.Lock()) + self._users[key] = self._users.get(key, 0) + 1 + try: + async with lock: + yield + finally: + self._users[key] -= 1 + if not self._users[key]: + del self._users[key], self._locks[key] + + +_APPENDING = _RunLocks() + + +async def append_events( + run_id: str, + items: Sequence[Item], + *, + require: Literal["interrupted", "sealed"], + seal: tuple[int | None, str | None] | None = None, +) -> list[RunEventModel] | None: + """把 items 依次接在运行记录的最后一条后面,返回写入的事件(seq 严格递增);没写入返回 None。 + + seal 只用于 "sealed":调用方开始做事时看到的 (manifest_seq, manifest_hash)。做的事要花几秒(问模型), + 这期间继续运行、重新封存了的话,结果依据的已经不是现在的封存,不追加。不传就按这次读到的封存。 + items 为空时什么都不做,返回空列表。 + """ + if require not in ("interrupted", "sealed"): + raise ValueError(f"require 只能是 interrupted 或 sealed,收到的是 {require!r}") + # 事件类型先验过:写错了在取号之前就报出来 + entries = [(EventType(str(kind)), node_id, dict(data or {})) for kind, node_id, data in items] + if not entries: + return [] + sealed = require == "sealed" + for _, _, data in entries: + if sealed: + data["post_seal"] = True + else: + # 停在审批时追加的事件要进封存范围,不能自称在封存之后 + data.pop("post_seal", None) + + async with _APPENDING.hold(run_id): + async with SessionLocal() as session: + run = await session.get(Run, run_id) + if run is None: + return None + if sealed: + if run.status not in _TERMINAL or not run.manifest_hash \ + or (seal is not None and (run.manifest_seq, run.manifest_hash) != tuple(seal)) \ + or not await _current_seal(session, run): + return None + guard = (Run.status.in_(_TERMINAL), _same(Run.manifest_seq, run.manifest_seq), + _same(Run.manifest_hash, run.manifest_hash)) + else: + if run.status != "interrupted": + return None + guard = (Run.status == "interrupted",) + last = run.last_seq + # 进程重启过的话内存里的序号从 0 起:接在库里最后一条后面。last_seq 按说就是最后一条,另取一次 + # 实际的最大序号兜底——别处的追加把 last_seq 写小了,也不至于在这里撞号 + stored = await session.scalar(select(func.max(RunEvent.seq)).where(RunEvent.run_id == run_id)) + bus.set_seq(run_id, max(last or 0, stored or 0)) + events = [run_manager._event(run_id, kind, node_id=node_id, data=data) + for kind, node_id, data in entries] + # 核对和写入是同一行上的条件更新,和恢复、放弃、继续运行(runner._claim / _abandon)抢的是同一行 + claimed = await session.execute( + update(Run).where(Run.id == run_id, _same(Run.last_seq, last), *guard) + .values(last_seq=events[-1].seq)) + if claimed.rowcount != 1: + # 读完运行之后它变了。烧掉几个序号不要紧:不落库的事件(流式增量)本来就占序号, + # 序号只要求递增、不撞号 + await session.rollback() + return None + for event in events: + await _stage(session, event, run) + await session.commit() + # 先落库再广播,理由同 RunManager._emit + for event in events: + await bus.publish(event) + return events + + +def _same(column: Any, value: Any) -> Any: + return column.is_(None) if value is None else column == value + + +async def _current_seal(session: Any, run: Run) -> bool: + """运行记录上的封存是不是这次结束之后封的:manifest_seq 盖住了最后一条宣布结束的事件。 + + 继续运行之后再次结束时,终态和终态事件先提交、封存后算。这中间运行记录上还是上一次结束时的封存, + 当成现在的封存去追加,事件会被接下来的封存收进去,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]))) + return ended is None or run.manifest_seq >= ended diff --git a/backend/tests/test_run_append.py b/backend/tests/test_run_append.py new file mode 100644 index 00000000..2a7707b3 --- /dev/null +++ b/backend/tests/test_run_append.py @@ -0,0 +1,469 @@ +"""往停下的运行里追加事件(engine/run_append.py 的 append_events)。 + +- 停在审批时追加:事件接在最后一条后面、落进以后的封存范围。批准、运行结束封存之后 verify 照旧一致, + 改动追加的事件 verify 就对不上——它们确实在封存范围里 +- 追加途中运行被恢复(条件更新落空):什么都不写,返回 None,last_seq 不动 +- 已封存的运行追加:带 post_seal、落在 manifest_seq 之后,verify 照旧一致;封存不是这次结束之后封的、 + 封存和调用方开始时看到的不同,都不追加 +- 连续追加、并发追加,序号严格递增,整条事件流没有重复的序号 +- 恢复后的重放不受影响:前面的节点不重复执行,Agent 里批过的工具不重复执行(含升级前写下的断点) +""" +from __future__ import annotations + +import asyncio +import json +import sqlite3 +from collections import Counter + +import pytest +from langchain_core.messages import AIMessage, ToolMessage +from pydantic import BaseModel +from sqlalchemy import select + +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.engine import compiler +from app.engine.context import NodeError +from app.engine.runner import RunManager, run_manager, verify_manifest +from app.engine.schema import NodeType +from app.tools.registry import register + +#: Agent 场景里工具真正执行的次数。注册表是进程级的,工具只注册一次,计数按测试清零 +CALLS: list[str] = [] + + +class _TallyArgs(BaseModel): + label: str + + +@register("inquiry_tally", "测试用:每执行一次记一笔", "test", _TallyArgs, dangerous=True, needs_context=False) +async def _tally(label: str) -> str: + CALLS.append(label) + return json.dumps({"label": label, "n": len(CALLS)}, ensure_ascii=False) + + +@pytest.fixture(autouse=True) +async def engine_up(): + CALLS.clear() + await run_manager.setup() + yield + await run_manager.shutdown() + + +def node(nid, ntype, **config): + return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}} + + +def chain(*nodes): + return {"nodes": list(nodes), "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]} + + +GATED = chain(node("start", "input"), + node("work", "transform", mode="template", template="本周销售额 4321.5 元"), + 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": "完成"}])) + + +def inquiry(question: str) -> tuple: + return (EventType.APPROVAL_INQUIRY, None, {"question": question, "answer": f"关于「{question}」:按订单表核对一致"}) + + +@pytest.fixture +def executed(monkeypatch) -> Counter: + """每个「数据整形」节点真正执行了几次。""" + seen: Counter = Counter() + original = compiler.RUNNERS[NodeType.TRANSFORM] + + async def counting(state, ctx): + seen[ctx.node.id] += 1 + return await original(state, ctx) + + monkeypatch.setitem(compiler.RUNNERS, NodeType.TRANSFORM, counting) + return seen + + +async def _row(run_id: str) -> Run: + async with SessionLocal() as session: + return await session.get(Run, run_id) + + +async def _until(run_id: str, statuses: tuple[str, ...], sealed: bool = False) -> Run: + for _ in range(500): + row = await _row(run_id) + if row.status in statuses and (not sealed or row.manifest_seq is not None): + await run_manager.wait_idle(run_id) + return await _row(run_id) + await asyncio.sleep(0.03) + raise AssertionError(f"运行 {run_id} 停在 {row.status}") + + +async def _paused(graph=GATED) -> Run: + run = await run_manager.start(graph=graph, input_payload={}) + return await _until(run.id, ("interrupted",)) + + +async def _finished(graph=SIMPLE) -> Run: + run = await run_manager.start(graph=graph, input_payload={}) + return await _until(run.id, ("succeeded", "failed"), sealed=True) + + +async def _events(run_id: str, etype: str | None = None) -> list[RunEvent]: + async with SessionLocal() as session: + query = select(RunEvent).where(RunEvent.run_id == run_id) + if etype: + query = query.where(RunEvent.type == etype) + return list((await session.execute(query.order_by(RunEvent.seq))).scalars()) + + +def _strictly_increasing(seqs: list[int]) -> bool: + return all(a < b for a, b in zip(seqs, seqs[1:])) + + +async def _no_repeated_seq(run_id: str) -> None: + seqs = [e.seq for e in await _events(run_id)] + assert len(seqs) == len(set(seqs)), f"事件流里有重复的序号:{seqs}" + + +# -------------------------------------------------------------------------- +# 停在审批时追加 +# -------------------------------------------------------------------------- + + +async def test_appended_while_waiting_are_sealed_with_the_run(executed): + from app.engine.run_append import append_events + + paused = await _paused() + subscription = bus.attach(paused.id) + try: + written = await append_events(paused.id, [inquiry("销售额为什么涨了"), inquiry("退款算进去了吗")], + require="interrupted") + assert written is not None and len(written) == 2 + assert written[0].seq > paused.last_seq and _strictly_increasing([e.seq for e in written]) + # 提交之后才广播:在线的客户端收到的就是落了库的这两条 + stream = subscription.stream() + live = [await asyncio.wait_for(stream.__anext__(), 2) for _ in range(2)] + assert [e.seq for e in live] == [e.seq for e in written] + finally: + subscription.close() + row = await _row(paused.id) + assert row.status == "interrupted" and row.last_seq == written[-1].seq + stored = await _events(paused.id, "approval.inquiry") + assert [(e.seq, e.node_id, e.data["question"]) for e in stored] == [ + (written[0].seq, None, "销售额为什么涨了"), (written[1].seq, None, "退款算进去了吗")] + assert all("post_seal" not in e.data for e in stored), "进封存范围的事件不该标成封存之后" + + await run_manager.resume(paused.id, {"approved": True}) + final = await _until(paused.id, ("succeeded", "failed"), sealed=True) + assert final.status == "succeeded", final.error + assert final.manifest_seq > written[-1].seq + check = await verify_manifest(paused.id) + assert check["ok"] is True and check["sealed_at"] == final.manifest_seq + # 恢复之后前面的节点没有再执行一遍 + assert executed["work"] == 1 + assert len([e for e in await _events(paused.id, "node.started") if e.node_id == "work"]) == 1 + await _no_repeated_seq(paused.id) + + # 追加的事件确实在封存范围里:改它一个字,封存核对就对不上 + async with SessionLocal() as session: + event = await session.get(RunEvent, stored[0].id) + event.data = {**event.data, "answer": "改过的回答"} + await session.commit() + assert (await verify_manifest(paused.id))["ok"] is False + + +async def test_a_resume_that_wins_the_row_leaves_nothing_behind(monkeypatch): + """读完运行、写入之前,恢复先把运行占成了 running:条件更新落空,一条都不写、last_seq 不动。""" + from app.engine.run_append import append_events + + paused = await _paused() + set_seq = bus.set_seq + + def resumed_in_between(run_id, value): + if run_id == paused.id: + db = sqlite3.connect(settings.db_path, timeout=5) + try: + db.execute("UPDATE runs SET status = 'running' WHERE id = ?", (paused.id,)) + db.commit() + finally: + db.close() + set_seq(run_id, value) + + monkeypatch.setattr(bus, "set_seq", resumed_in_between) + assert await append_events(paused.id, [inquiry("毛利率怎么算的")], require="interrupted") is None + row = await _row(paused.id) + assert row.status == "running" and row.last_seq == paused.last_seq + assert await _events(paused.id, "approval.inquiry") == [] + + +async def test_events_written_elsewhere_in_between_win(monkeypatch): + """读完运行、写入之前,别处(另一个进程,或者没走这把锁的追加)已经接着写了一条:状态没变,last_seq 变了。 + 条件更新落空、不写——否则两边各拿着同一个序号,事件流里撞号。""" + from app.engine.run_append import append_events + + paused = await _paused() + taken = paused.last_seq + 1 + set_seq = bus.set_seq + + def written_in_between(run_id, value): + if run_id == paused.id: + db = sqlite3.connect(settings.db_path, timeout=5) + try: + db.execute("INSERT INTO run_events (id, run_id, seq, type, node_id, ts, data) " + "VALUES ('elsewhere-1', ?, ?, 'log', NULL, 0, '{}')", (paused.id, taken)) + db.execute("UPDATE runs SET last_seq = ? WHERE id = ?", (taken, paused.id)) + db.commit() + finally: + db.close() + set_seq(run_id, value) + + monkeypatch.setattr(bus, "set_seq", written_in_between) + bus._seq.pop(paused.id, None) # 另一个进程的序号本进程不知道 + assert await append_events(paused.id, [inquiry("同时写进来的")], require="interrupted") is None + assert (await _row(paused.id)).last_seq == taken + assert await _events(paused.id, "approval.inquiry") == [] + await _no_repeated_seq(paused.id) + + +async def test_a_last_seq_left_behind_does_not_cause_a_clash(): + """last_seq 比库里实际的最后一条小(别处的追加交错着把它写小了):照实际的最后一条接号,顺手把 last_seq 写对。""" + from app.engine.run_append import append_events + + paused = await _paused() + async with SessionLocal() as session: + run = await session.get(Run, paused.id) + run.last_seq = paused.last_seq - 3 + await session.commit() + bus._seq.pop(paused.id, None) + written = await append_events(paused.id, [inquiry("接在哪里")], require="interrupted") + assert written[0].seq == paused.last_seq + 1 + assert (await _row(paused.id)).last_seq == written[0].seq + await _no_repeated_seq(paused.id) + + +async def test_a_run_that_already_moved_on_gets_nothing(executed): + from app.engine.run_append import append_events + + paused = await _paused() + await run_manager.resume(paused.id, {"approved": True}) + final = await _until(paused.id, ("succeeded",), sealed=True) + assert await append_events(paused.id, [inquiry("晚到的问题")], require="interrupted") is None + assert (await _row(paused.id)).last_seq == final.last_seq + assert await _events(paused.id, "approval.inquiry") == [] + + +# -------------------------------------------------------------------------- +# 封存之后追加 +# -------------------------------------------------------------------------- + + +async def test_post_seal_appends_stay_outside_the_seal(): + from app.engine.run_append import append_events + + final = await _finished() + assert final.status == "succeeded", final.error + before = await verify_manifest(final.id) + written = await append_events(final.id, [inquiry("这个数是含税的吗"), inquiry("和上周口径一样吗")], + require="sealed") + assert written is not None and all(e.data["post_seal"] is True for e in written) + assert written[0].seq > final.manifest_seq and _strictly_increasing([e.seq for e in written]) + row = await _row(final.id) + assert (row.status, row.manifest_seq, row.manifest_hash) == ("succeeded", final.manifest_seq, final.manifest_hash) + assert row.last_seq == written[-1].seq + after = await verify_manifest(final.id) + assert after["ok"] is True and (after["sealed_at"], after["events"]) == (before["sealed_at"], before["events"]) + assert [e.data["post_seal"] for e in await _events(final.id, "approval.inquiry")] == [True, True] + await _no_repeated_seq(final.id) + + +async def test_a_seal_pinned_by_the_caller_must_still_be_the_seal(): + from app.engine.run_append import append_events + + final = await _finished() + seal = (final.manifest_seq, final.manifest_hash) + assert await append_events(final.id, [inquiry("甲")], require="sealed", seal=(seal[0] - 1, seal[1])) is None + assert await append_events(final.id, [inquiry("乙")], require="sealed", seal=(seal[0], "0" * 64)) is None + assert await _events(final.id, "approval.inquiry") == [] + written = await append_events(final.id, [inquiry("丙")], require="sealed", seal=seal) + assert written is not None and written[0].data["question"] == "丙" + + +@pytest.mark.parametrize("change", [ + "UPDATE runs SET status = 'running' WHERE id = ?", # 继续运行占住了它 + "UPDATE runs SET manifest_hash = 'resealed' WHERE id = ?", # 继续运行之后重新封存了 + "UPDATE runs SET manifest_seq = manifest_seq + 1 WHERE id = ?", +], ids=["continued", "resealed-hash", "resealed-seq"]) +async def test_a_sealed_run_that_changes_in_between_gets_nothing(monkeypatch, change): + """封存之后追加也是核对和写入同一行上的条件更新:读完运行、写入之前它被继续运行占住、或者重新封存了,就不写。""" + from app.engine.run_append import append_events + + final = await _finished() + set_seq = bus.set_seq + + def changed_in_between(run_id, value): + if run_id == final.id: + db = sqlite3.connect(settings.db_path, timeout=5) + try: + db.execute(change, (final.id,)) + db.commit() + finally: + db.close() + set_seq(run_id, value) + + monkeypatch.setattr(bus, "set_seq", changed_in_between) + assert await append_events(final.id, [inquiry("这期为什么少了")], require="sealed") is None + assert (await _row(final.id)).last_seq == final.last_seq + assert await _events(final.id, "approval.inquiry") == [] + + +async def test_a_seal_left_from_an_earlier_ending_is_not_the_seal(monkeypatch): + """继续运行后又失败:终态和 run.failed 先提交、封存后算。这中间运行记录上还是上一次结束时的封存, + 这时追加的事件会被接下来的封存收进去,post_seal 就成了假话——不追加。""" + from app.engine.run_append import append_events + + original = compiler.RUNNERS[NodeType.TRANSFORM] + + async def boom(state, ctx): + if ctx.node.id == "boom": + raise NodeError(ctx.node.id, "数据源暂时不可用") + return await original(state, ctx) + + monkeypatch.setitem(compiler.RUNNERS, NodeType.TRANSFORM, boom) + first = await _finished(chain(node("start", "input"), node("boom", "transform", mode="template", template="x"), + node("out", "output", fields=[{"name": "r", "value": "x"}]))) + assert first.status == "failed" + + entered, release = asyncio.Event(), asyncio.Event() + real_seal = RunManager._seal + + async def held_seal(self, run_id): + if run_id == first.id and not entered.is_set(): + entered.set() + await release.wait() + return await real_seal(self, run_id) + + monkeypatch.setattr(RunManager, "_seal", held_seal) + await run_manager.continue_failed(first.id) + await asyncio.wait_for(entered.wait(), 10) + try: + row = await _row(first.id) + failures = [e.seq for e in await _events(first.id, "run.failed")] + assert row.status == "failed" and row.manifest_seq == first.manifest_seq < failures[-1] + assert await append_events(first.id, [inquiry("为什么又失败了")], require="sealed") is None + finally: + release.set() + await run_manager.wait_idle(first.id) + resealed = await _row(first.id) + assert resealed.manifest_seq >= failures[-1] + written = await append_events(first.id, [inquiry("为什么又失败了")], require="sealed") + assert written is not None and written[0].seq > resealed.manifest_seq + assert (await verify_manifest(first.id))["ok"] is True + + +async def test_the_wrong_kind_of_run_gets_nothing(): + from app.engine.run_append import append_events + + paused, final = await _paused(), await _finished() + assert await append_events(paused.id, [inquiry("甲")], require="sealed") is None, "停在审批的运行还没封存" + assert await append_events(final.id, [inquiry("乙")], require="interrupted") is None, "已结束的运行不在等审批" + assert await append_events("no-such-run", [inquiry("丙")], require="interrupted") is None + assert await append_events(paused.id, [], require="interrupted") == [] + with pytest.raises(ValueError): + await append_events(paused.id, [inquiry("丁")], require="running") + with pytest.raises(ValueError): + await append_events(paused.id, [("approval.unknown", None, {})], require="interrupted") + for run in (paused, final): + assert (await _row(run.id)).last_seq == run.last_seq + assert await _events(run.id, "approval.inquiry") == [] + + +# -------------------------------------------------------------------------- +# 序号 +# -------------------------------------------------------------------------- + + +async def test_seqs_strictly_increase_and_never_repeat(): + from app.engine.run_append import append_events + + paused = await _paused() + seqs: list[int] = [] + for batch in ([inquiry("一")], [inquiry("二"), inquiry("三")], [inquiry("四")]): + written = await append_events(paused.id, batch, require="interrupted") + seqs += [e.seq for e in written] + assert seqs[0] > paused.last_seq and _strictly_increasing(seqs) + + # 同时来几个:同一运行在本进程里排队,读 last_seq 到写入之间不会被另一次插进来,所以一个都不落空 + results = await asyncio.gather(*(append_events(paused.id, [inquiry(f"并发{i}")], require="interrupted") + for i in range(6))) + assert all(r is not None for r in results), "并发的追加有落空的" + together = sorted(r[0].seq for r in results) + assert together[0] > seqs[-1] and _strictly_increasing(together) + assert (await _row(paused.id)).last_seq == together[-1] + await _no_repeated_seq(paused.id) + + await run_manager.resume(paused.id, {"approved": True}) + final = await _until(paused.id, ("succeeded",), sealed=True) + assert (await verify_manifest(final.id))["ok"] is True + await _no_repeated_seq(final.id) + + +async def test_a_restarted_process_continues_after_the_stored_events(): + """进程重启过,内存里的序号从 0 起:追加接在库里最后一条后面。""" + from app.engine.run_append import append_events + + paused = await _paused() + bus._seq.pop(paused.id, None) + written = await append_events(paused.id, [inquiry("重启之后的问题")], require="interrupted") + assert written[0].seq == paused.last_seq + 1 + await _no_repeated_seq(paused.id) + + +# -------------------------------------------------------------------------- +# 恢复后的重放:Agent 里批过的工具不重复执行 +# -------------------------------------------------------------------------- + + +def _script_two_calls(monkeypatch) -> None: + from app.providers import mock_model + + def _decide(self, messages): + done = sum(isinstance(m, ToolMessage) for m in messages) + if self.tools and done < 2: + return AIMessage(content="", tool_calls=[ + {"name": "inquiry_tally", "args": {"label": f"第{done + 1}次"}, "id": f"call_{done + 1}"}]) + return AIMessage(content="两笔都记好了。") + + monkeypatch.setattr(mock_model.MockChatModel, "_decide", _decide) + + +@pytest.mark.parametrize("older_pause", [False, True], ids=["current", "older-protocol"]) +async def test_inquiries_do_not_disturb_the_replay(monkeypatch, older_pause): + """Agent 停在工具审批上时追加问答,批准后工具各执行一次。升级前写下的断点要按事件流判断能不能 + 安全恢复(runner._stale_replay 只看 tool.start / human.resolved),追加的问答不影响判断。""" + from app.engine import replay, runner + from app.engine.run_append import append_events + + _script_two_calls(monkeypatch) + if older_pause: + monkeypatch.setattr(runner, "PROTOCOL", runner.PROTOCOL - 1) + graph = chain(node("start", "input"), + node("bot", "agent", prompt="记两笔", tools=["inquiry_tally"], approval="always", max_steps=5), + node("out", "output", fields=[{"name": "r", "value": "{{ nodes.bot.text }}"}])) + run = await run_manager.start(graph=graph, input_payload={}) + for round_ in range(2): + paused = await _until(run.id, ("interrupted", "failed", "succeeded")) + assert paused.status == "interrupted", (paused.status, paused.error) + assert await append_events(run.id, [inquiry(f"第{round_ + 1}笔记的是什么")], require="interrupted") + if older_pause: + monkeypatch.setattr(runner, "PROTOCOL", replay.PROTOCOL) + await run_manager.resume(run.id, {"approved": True}) + await asyncio.sleep(0.05) + final = await _until(run.id, ("succeeded", "failed"), sealed=True) + assert final.status == "succeeded", final.error + assert CALLS == ["第1次", "第2次"], "批过的工具又执行了一次" + assert len(await _events(run.id, "approval.inquiry")) == 2 + assert (await verify_manifest(run.id))["ok"] is True + await _no_repeated_seq(run.id)