mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 16:41:59 +00:00
ThearchyHelios
fc57ed0e32
- 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，对应的测试各自变红
763 lines
36 KiB
Python
763 lines
36 KiB
Python
"""往停下的运行里追加事件（engine/run_append.py 的 append_events）。
|
||
|
||
- 停在审批时追加：事件接在最后一条后面、落进以后的封存范围。批准、运行结束封存之后 verify 照旧一致，
|
||
改动追加的事件 verify 就对不上——它们确实在封存范围里
|
||
- 追加途中运行被恢复（条件更新落空）：什么都不写，返回 None，last_seq 不动
|
||
- 质询指明问的是哪张审批卡（approval_id）时，那张卡仍待审批才写；返回的事件和落库的 id、序号一样
|
||
- 先落库再广播，没写入的一条都不广播
|
||
- 已封存的运行追加：带 post_seal、落在 manifest_seq 之后，verify 照旧一致；封存不是这次结束之后封的、
|
||
封存和调用方开始时看到的不同、没有终态事件的老封存，都不追加
|
||
- 连续追加、并发追加，序号严格递增，整条事件流没有重复的序号
|
||
- last_seq 落后于库里实际的最后一条（按需裁判的追加和封存后追加交错提交时会这样）：进程重启后恢复、
|
||
继续运行、放弃、按需裁判都接在实际的最后一条后面，不撞号；runner 记事件不把 last_seq 写小
|
||
- 恢复后的重放不受影响：前面的节点不重复执行，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 Approval, 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": "完成"}]))
|
||
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 }}"}]))
|
||
|
||
|
||
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
|
||
|
||
|
||
@pytest.fixture
|
||
def flaky(monkeypatch) -> dict:
|
||
"""FLAKY 里的「boom」在 fail 为真时失败。"""
|
||
state = {"fail": True}
|
||
original = compiler.RUNNERS[NodeType.TRANSFORM]
|
||
|
||
async def boom(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, boom)
|
||
return state
|
||
|
||
|
||
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 _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)
|
||
try:
|
||
db.execute(statement, params)
|
||
db.commit()
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 停在审批时追加
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
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_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 == []
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 封存之后追加
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
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 _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
|
||
|
||
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)
|
||
|
||
|
||
@pytest.mark.parametrize("action", ["continue", "resume", "abandon"])
|
||
async def test_the_runner_numbers_after_the_stored_events_when_last_seq_lags(flaky, action):
|
||
"""last_seq 比库里实际的最后一条小（怎么来的见下一条），本进程又没有这次运行的序号计数（重启过）：继续运行、
|
||
恢复、放弃都接在实际的最后一条后面。只按 last_seq 接，写出的 run.resumed / run.cancelled 撞上已有的序号，
|
||
重新封存时两条同号的事件一起进了清单。"""
|
||
if action == "continue":
|
||
run = await _finished(FLAKY)
|
||
assert run.status == "failed"
|
||
else:
|
||
run = await _paused()
|
||
stored = (await _events(run.id))[-1].seq
|
||
_sql("UPDATE runs SET last_seq = last_seq - 3 WHERE id = ?", run.id)
|
||
bus._seq.pop(run.id, None)
|
||
|
||
if action == "continue":
|
||
flaky["fail"] = False
|
||
await run_manager.continue_failed(run.id)
|
||
first_new = "run.resumed"
|
||
elif action == "resume":
|
||
await run_manager.resume(run.id, {"approved": True})
|
||
first_new = "run.resumed"
|
||
else:
|
||
assert await run_manager.cancel(run.id) == "cancelled"
|
||
first_new = "run.cancelled"
|
||
final = await _until(run.id, ("succeeded", "cancelled"), sealed=True)
|
||
assert final.status == ("cancelled" if action == "abandon" else "succeeded"), final.error
|
||
assert [e.seq for e in await _events(run.id, first_new)][-1] == stored + 1
|
||
await _no_repeated_seq(run.id)
|
||
assert (await verify_manifest(run.id))["ok"] is True
|
||
|
||
|
||
async def test_runner_event_writes_never_lower_last_seq():
|
||
"""runner 记事件时 last_seq 只往大里改。取号和提交的先后可以不一样：别处另开会话记的事件（run_manager.note）
|
||
和执行任务同时在写，号大的先提交、号小的后提交，以前后提交的那条把 last_seq 写回小的。"""
|
||
from app.engine.runner import _stage
|
||
|
||
paused = await _paused()
|
||
early = run_manager._event(paused.id, EventType.LOG, data={"level": "info", "message": "先取号"})
|
||
late = run_manager._event(paused.id, EventType.LOG, data={"level": "info", "message": "后取号"})
|
||
for event in (late, early):
|
||
async with SessionLocal() as session:
|
||
await _stage(session, event)
|
||
await session.commit()
|
||
assert (await _row(paused.id)).last_seq == late.seq
|
||
|
||
|
||
def _judged_args(run: Run) -> tuple:
|
||
"""直接调按需裁判记判定的那一步（api/evidence.py 的 _append_judged）要的参数：问过一次模型、什么都没判成。"""
|
||
from app.api import evidence
|
||
|
||
sealed = evidence._Sealed(run_id=run.id, graph={}, seal={"sealed": True, "ok": True}, events=[],
|
||
manifest_hash=run.manifest_hash, run_manifest_seq=run.manifest_seq)
|
||
report = evidence._Report(node_id="out", doc_artifact="doc-1", ok=None, repairs=None, stats=None, doc=None,
|
||
hash_ok=None, fields=[])
|
||
outcome = {key: {} for key in ("verdicts", "keys", "unjudged", "skipped", "budget")}
|
||
outcome.update(model="judge-model", priced=True, same_model=False, writer_model=None, cost_usd=0.0021,
|
||
calls=1, duration_ms=1200, limits_hit=[], gaps=[], notes=[])
|
||
return sealed, report, ["u1"], outcome
|
||
|
||
|
||
async def test_a_judge_interleaved_with_a_post_seal_append_survives_a_restart(flaky, monkeypatch):
|
||
"""按需裁判记判定（_append_judged）的条件更新不带 last_seq 条件：它取了号、还没写，封存后追加抢先提交了，
|
||
它再提交时把 last_seq 写回自己那个更小的号。同一进程里序号计数只增不减，不会撞号；进程重启后计数从 0 起，
|
||
各处又按 last_seq 接号——裁判自己也是。所以启动时按库里实际的最后一条改正 last_seq：之后再裁判、继续运行，
|
||
都接在实际的最后一条后面。"""
|
||
from app.api import evidence
|
||
from app.engine.run_append import append_events
|
||
|
||
first = await _finished(FLAKY)
|
||
assert first.status == "failed"
|
||
held, opened = asyncio.Event(), evidence.SessionLocal
|
||
|
||
class HeldBeforeUpdate:
|
||
"""裁判的会话：执行条件更新之前停住，等封存后追加先提交。"""
|
||
|
||
def __init__(self):
|
||
self._cm = opened()
|
||
|
||
async def __aenter__(self):
|
||
session = await self._cm.__aenter__()
|
||
execute = session.execute
|
||
|
||
async def held_execute(statement, *args, **kwargs):
|
||
if getattr(statement, "is_update", False):
|
||
await held.wait()
|
||
return await execute(statement, *args, **kwargs)
|
||
|
||
session.execute = held_execute
|
||
return session
|
||
|
||
async def __aexit__(self, *exc):
|
||
return await self._cm.__aexit__(*exc)
|
||
|
||
monkeypatch.setattr(evidence, "SessionLocal", HeldBeforeUpdate)
|
||
judging = asyncio.create_task(evidence._append_judged(*_judged_args(first)))
|
||
for _ in range(200): # 裁判取到号了
|
||
if bus._seq.get(first.id, 0) > first.last_seq:
|
||
break
|
||
await asyncio.sleep(0.01)
|
||
written = await append_events(first.id, [inquiry("这次为什么失败")], require="sealed")
|
||
held.set()
|
||
assert await judging == {"seq": first.last_seq + 1}
|
||
assert written is not None and written[0].seq == first.last_seq + 2
|
||
assert (await _row(first.id)).last_seq == first.last_seq + 1, "裁判后提交，last_seq 被写回了小的号"
|
||
monkeypatch.setattr(evidence, "SessionLocal", opened)
|
||
|
||
# 进程重启：内存里的序号计数没了，启动时 setup 照常执行一遍。只改落后的那条，别的运行原样不动
|
||
lagging = await _row(first.id)
|
||
others = [await _finished(), await _paused()]
|
||
await run_manager.shutdown()
|
||
bus._seq.pop(first.id, None)
|
||
await run_manager.setup()
|
||
repaired = await _row(first.id)
|
||
assert (repaired.last_seq, repaired.updated_at) == (written[-1].seq, lagging.updated_at)
|
||
for other in others:
|
||
now = await _row(other.id)
|
||
assert (now.last_seq, now.updated_at) == (other.last_seq, other.updated_at)
|
||
assert await evidence._append_judged(*_judged_args(first)) == {"seq": written[-1].seq + 1}
|
||
await _no_repeated_seq(first.id)
|
||
|
||
flaky["fail"] = False
|
||
await run_manager.continue_failed(first.id)
|
||
final = await _until(first.id, ("succeeded",), sealed=True)
|
||
assert final.manifest_seq > written[-1].seq + 1
|
||
await _no_repeated_seq(first.id)
|
||
assert (await verify_manifest(first.id))["ok"] is True
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 恢复后的重放：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)
|