"""运行固定上传表格的数据版本（期 1 WP-D）。 上传源每次重传是一个新快照。运行发起时把这张图用到的上传源的当前快照记进 runs.data_versions， 之后这次运行里的一切——工具描述、db_schema、冻结的表结构、查询结果——都出自那一版： - 运行中途重传：审批恢复、失败后继续运行，查到的仍是发起时那一版； - 固定的快照被回收、文件被删：工具明确报错，绝不改用当前版本； - 只固定图里（含子工作流、图级 defaults 绑的）实际用到的上传源；工具名带模板时全部固定；手工源不进来； - 子工作流沿用父运行固定的版本；构造工具时没拿到版本（漏传）的上传源报错，不改用当前版本； - 发起时没算到的源，第一次用到时补固定、记进运行，并在事件流里留一条日志； - 调用工具节点遇到数据本身用不了时判失败，报错照交原因，不让人改 SQL； - 查询快照记下 data_version（快照 id），手工源不记；工具库试用（不在运行里）用当前版本。 夹具：表格在测试里用 openpyxl 现造，标签是假名，数字手写。 """ from __future__ import annotations import asyncio import io import json import os import sqlite3 import stat import uuid import pytest from langchain_core.messages import AIMessage, ToolMessage from sqlalchemy import select from app.api.runs import missing_tools from app.core import artifact_store from app.core.config import settings from app.core.events import EventType from app.data import table_versions from app.data.engine import engines from app.db.base import SessionLocal from app.db.models import DataSource, Run, RunEvent, SourceSnapshot, Workflow from app.engine.context import RunContext from app.engine.runner import _data_versions, run_manager from app.engine.schema import GraphSpec from app.tools.datasource import ( DATA_UNAVAILABLE, LATE_PIN_CODE, RUN_VERSIONS_KEY, build_datasource_tools, run_versions, ) from app.tools.registry import ToolContext, Versions SUM = 'SELECT SUM("数量") AS total FROM "明细"' @pytest.fixture(autouse=True) def store(monkeypatch, tmp_path): data = tmp_path / "data" (data / "uploads").mkdir(parents=True) monkeypatch.setattr(settings, "data_dir", data) yield data @pytest.fixture async def engine_up(store): # 依赖 store：checkpoint 库跟着 data_dir 走，要先换好目录再建 await run_manager.setup() yield await run_manager.shutdown() # -------------------------------------------------------------------------- # 夹具：表格、数据源 # -------------------------------------------------------------------------- def book(sheets: dict[str, list[list]]) -> bytes: from openpyxl import Workbook wb = Workbook() wb.remove(wb.active) for title, rows in sheets.items(): ws = wb.create_sheet(title) for row in rows: ws.append(row) buf = io.BytesIO() wb.save(buf) return buf.getvalue() #: 第一版：一张「明细」，数量合计 3 V1 = {"明细": [["区域", "数量"], ["分区甲", 1], ["分区乙", 2]]} #: 第二版：「明细」合计 100，另多一张「新增」。表名、数字都和第一版不同，串了版本一眼看得出 V2 = {"明细": [["区域", "数量"], ["分区甲", 60], ["分区乙", 40]], "新增": [["时段", "客流"], ["上午", 7]]} def uniq(prefix: str = "dv") -> str: return f"{prefix}_{uuid.uuid4().hex[:10]}" async def publish(name: str, sheets: dict[str, list[list]]) -> tuple[str, str]: """发布一版上传表格，返回 (源 id, 快照 id)。源不存在就新建。""" async with SessionLocal() as session: row = (await session.execute(select(DataSource).where(DataSource.name == name))).scalar_one_or_none() if row is None: row = DataSource(id=uuid.uuid4().hex, name=name, kind="sqlite", readonly=True, origin="upload") result = await table_versions.publish_upload(session, row, book(sheets), "客流.xlsx", header_row=1, mixed="reject", raw_mode=False) return row.id, result.snapshot_id async def manual_source(tmp_path, total: int) -> tuple[str, str, str]: """一个手工登记的 SQLite 源（文件在上传目录外）。返回 (源 id, 源名, 文件路径)。""" path = tmp_path / f"{uuid.uuid4().hex[:8]}.db" db = sqlite3.connect(path) db.executescript('CREATE TABLE "明细" ("区域" TEXT, "数量" INTEGER);') db.execute('INSERT INTO "明细" VALUES (?, ?)', ("分区甲", total)) db.commit() db.close() name = uniq("manual") async with SessionLocal() as session: row = DataSource(name=name, kind="sqlite", database=str(path), readonly=True, enabled=True) session.add(row) await session.commit() return row.id, name, str(path) async def keep(source_id: str, name: str, snapshot: str) -> str: """登记一条引用这个快照的中断运行，让之后重传触发的回收不删它。返回运行 id。""" async with SessionLocal() as session: run = Run(workflow_name="保留旧版本", status="interrupted", data_versions={source_id: {"snapshot": snapshot, "name": name}}) session.add(run) await session.commit() return run.id async def tools_for(names: list[str], *, data_versions=Versions.CURRENT, run_id: str = "", emit=None) -> dict: """建数据源工具。缺省按不在运行里（工具库试用）建，用当前版本。""" async with SessionLocal() as session: built = await build_datasource_tools( names, ToolContext(run_id=run_id, node_id="n", data_versions=data_versions, emit=emit), session) return {t.name: t for t in built} def total_of(text: str) -> object: return json.loads(text)["rows"][0][0] def pinned_of(pins: dict, sid: str) -> str | None: return (pins.get(sid) or {}).get("snapshot") # -------------------------------------------------------------------------- # 跑图的工具 # -------------------------------------------------------------------------- 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:])]} async def wait(run_id: str, statuses=("interrupted", "failed", "succeeded"), timeout=30.0) -> Run: deadline = asyncio.get_event_loop().time() + timeout while asyncio.get_event_loop().time() < deadline: async with SessionLocal() as session: row = await session.get(Run, run_id) if row and row.status in statuses and run_manager._tasks.get(run_id) is None: return row await asyncio.sleep(0.05) raise AssertionError(f"等超时了：{run_id}") async def events_of(run_id: str, kind: str, node_id: str | None = None) -> list[RunEvent]: async with SessionLocal() as session: query = select(RunEvent).where(RunEvent.run_id == run_id, RunEvent.type == kind) if node_id is not None: query = query.where(RunEvent.node_id == node_id) return list((await session.execute(query.order_by(RunEvent.seq))).scalars()) async def started_events(run_id: str) -> list[RunEvent]: return await events_of(run_id, "run.started") async def late_pin_events(run_id: str) -> list[RunEvent]: return [e for e in await events_of(run_id, "log") if (e.data or {}).get("code") == LATE_PIN_CODE] def loop_graph(first: str, second: str, condition: str) -> dict: """t1 查 first → 循环（条件执行时才报错的话停在这里）→ t2 查 second → 输出两次的合计。""" return { "nodes": [ node("start", "input"), node("t1", "tool", tool=f"db_query__{first}", args={"sql": SUM}), node("lp", "loop", mode="while", condition=condition, max_iterations=1), node("body", "transform", mode="template", template="转一圈"), node("t2", "tool", tool=f"db_query__{second}", args={"sql": SUM}), node("out", "output", fields=[{"name": "t1", "value": "{{ nodes.t1 }}"}, {"name": "t2", "value": "{{ nodes.t2 }}"}]), ], "edges": [ {"source": "start", "target": "t1"}, {"source": "t1", "target": "lp"}, {"source": "lp", "target": "body", "sourceHandle": "body"}, {"source": "body", "target": "lp"}, {"source": "lp", "target": "t2", "sourceHandle": "done"}, {"source": "t2", "target": "out"}, ], } #: 过得了静态检查、执行时才报错的循环条件（指数超限）：运行停在循环节点上，失败 FAILS_AT_RUNTIME = "2 ** 100000 > 1" def output_total(run: Run, field: str) -> object: value = (run.output or {})[field] return total_of(value if isinstance(value, str) else json.dumps(value)) def scripted_agent(monkeypatch, source: str) -> dict: """Agent 先列表结构、再查合计、然后收尾；协作成员（只绑了查询工具）直接查合计。 记下各自看到的工具描述（按 (谁, 工具名)）和工具结果（按调用 id）。 """ from app.providers import mock_model seen: dict = {"descriptions": {}, "results": {}} routes = {"n": 0} def _decide(self, messages): if self.response_format: if "assignments" not in json.dumps(self.response_format, ensure_ascii=False): return AIMessage(content="{}") # 调度者：第一轮派给成员，第二轮收工 routes["n"] += 1 plan = ({"assignments": [{"agent": "analyst", "instruction": "查合计"}]} if routes["n"] == 1 else {"assignments": [], "done": True}) return AIMessage(content=json.dumps(plan, ensure_ascii=False)) if not self.tools: return AIMessage(content="收尾") names = {t["name"] for t in self.tools} who = "agent" if f"db_schema__{source}" in names else "member" for t in self.tools: seen["descriptions"][(who, t["name"])] = t["description"] done = set() for m in messages: if isinstance(m, ToolMessage): seen["results"][m.tool_call_id] = str(m.content) done.add(m.tool_call_id) plan = [(f"{who}_schema", f"db_schema__{source}", {}), (f"{who}_sum", f"db_query__{source}", {"sql": SUM})] for call_id, name, args in plan: if name in names and call_id not in done: return AIMessage(content="查一下。", tool_calls=[{"name": name, "args": args, "id": call_id}]) return AIMessage(content="查完了。") monkeypatch.setattr(mock_model.MockChatModel, "_decide", _decide) return seen def json_in(text: str) -> dict: """工具结果交给模型时前面可能标了证据编号：从第一个「{」起读。""" return json.loads(text[text.index("{"):]) # -------------------------------------------------------------------------- # 发起时固定哪些源 # -------------------------------------------------------------------------- async def test_start_pins_only_the_upload_sources_the_graph_uses(engine_up, monkeypatch, tmp_path): used, unused = uniq(), uniq() used_id, used_snap = await publish(used, V1) unused_id, _ = await publish(unused, V1) manual_id, manual, _ = await manual_source(tmp_path, 5) scripted_agent(monkeypatch, used) graph = chain( node("start", "input"), node("t1", "tool", tool=f"db_query__{used}", args={"sql": SUM}), node("bot", "agent", prompt="查一下", tools=[f"db_schema__{used}", f"db_query__{manual}"], max_steps=3), node("out", "output", fields=[{"name": "t1", "value": "{{ nodes.t1 }}"}]), ) run = await run_manager.start(graph=graph, input_payload={}) pinned = {used_id: {"snapshot": used_snap, "name": used}} assert run.data_versions == pinned done = await wait(run.id, ("succeeded", "failed")) assert done.status == "succeeded", done.error assert done.data_versions == pinned assert unused_id not in done.data_versions and manual_id not in done.data_versions # run.started 记下固定的版本：封存清单里查得到这次运行用的是哪一版 assert (await started_events(run.id))[0].data["data_versions"] == pinned async def test_subgraph_sources_are_pinned_and_templated_names_pin_every_upload_source(tmp_path): top, inner, other, off = uniq(), uniq(), uniq(), uniq() top_id, top_snap = await publish(top, V1) inner_id, inner_snap = await publish(inner, V1) other_id, _ = await publish(other, V1) off_id, _ = await publish(off, V1) _, manual, _ = await manual_source(tmp_path, 5) async with SessionLocal() as session: (await session.get(DataSource, off_id)).enabled = False sub = Workflow(name="子流程", graph=chain( node("s", "input"), node("q", "tool", tool=f"db_query__{inner}", args={"sql": SUM}), node("o", "output"))) session.add(sub) await session.commit() sub_id = sub.id graph = chain( node("start", "input"), node("t", "tool", tool=f"db_query__{top}", args={"sql": SUM}), node("sub", "subgraph", workflow_id=sub_id), node("m", "tool", tool=f"db_query__{manual}", args={"sql": SUM}), node("out", "output"), ) async with SessionLocal() as session: pinned = await _data_versions(session, GraphSpec.model_validate(graph)) # 子工作流里绑的源也固定；没用到的、手工的不固定 assert pinned == {top_id: {"snapshot": top_snap, "name": top}, inner_id: {"snapshot": inner_snap, "name": inner}} # 工具名带模板：发起时说不准会用哪个源，所有启用着的上传源都固定上，停用的、手工的不算 graph["nodes"][1]["data"]["config"]["tool"] = "db_query__{{ vars.which }}" async with SessionLocal() as session: everything = await _data_versions(session, GraphSpec.model_validate(graph)) uploads = {r.id: r.current_snapshot_id for r in (await session.execute(select(DataSource).where( DataSource.origin == "upload", DataSource.enabled.is_(True), DataSource.current_snapshot_id.is_not(None)))).scalars()} assert {sid: v["snapshot"] for sid, v in everything.items()} == uploads assert {top_id, inner_id, other_id} <= set(everything) and off_id not in everything # 一个上传源都没用到：空 dict（不是 None，None 留给升级前发起的运行） async with SessionLocal() as session: assert await _data_versions(session, GraphSpec.model_validate( chain(node("start", "input"), node("out", "output")))) == {} # -------------------------------------------------------------------------- # 运行中途重传：恢复、继续运行都还是发起时那一版 # -------------------------------------------------------------------------- async def test_reupload_while_waiting_for_approval_does_not_change_the_run(engine_up, monkeypatch): src = uniq() sid, s1 = await publish(src, V1) seen = scripted_agent(monkeypatch, src) graph = chain( node("start", "input"), node("t1", "tool", tool=f"db_query__{src}", args={"sql": SUM}), node("gate", "human", mode="approve", title="看一眼"), node("bot", "agent", prompt="查合计", tools=[f"db_schema__{src}", f"db_query__{src}"], max_steps=4), node("team", "supervisor", goal="查合计", max_rounds=2, agents=[{"name": "analyst", "system": "你是分析员", "tools": [f"db_query__{src}"], "max_steps": 3}]), node("t2", "tool", tool=f"db_query__{src}", args={"sql": SUM}), node("out", "output", fields=[{"name": "t1", "value": "{{ nodes.t1 }}"}, {"name": "t2", "value": "{{ nodes.t2 }}"}]), ) run = await run_manager.start(graph=graph, input_payload={}) assert (await wait(run.id)).status == "interrupted" # 停在审批上时重传：当前版本变成 S2（合计 100，多一张「新增」） _, s2 = await publish(src, V2) assert s2 != s1 await run_manager.resume(run.id, {"approved": True}) done = await wait(run.id, ("succeeded", "failed")) assert done.status == "succeeded", done.error assert output_total(done, "t1") == 3 assert output_total(done, "t2") == 3, "恢复之后查到了重传的新版本" # 恢复后才建的 Agent 和协作成员的工具：描述、表结构、查询都还是 S1 for who in ("agent", "member"): desc = seen["descriptions"][(who, f"db_query__{src}")] assert "明细" in desc and "新增" not in desc, who assert "共 1 张表" in seen["results"]["agent_schema"] and "新增" not in seen["results"]["agent_schema"] assert json_in(seen["results"]["agent_sum"])["rows"] == [[3]] assert json_in(seen["results"]["member_sum"])["rows"] == [[3]] # 运行记录和两段 run.started 记的都是 S1 assert done.data_versions == {sid: {"snapshot": s1, "name": src}} assert [e.data["data_versions"] for e in await started_events(run.id)] == [ {sid: {"snapshot": s1, "name": src}}] * 2 async def test_continue_after_failure_keeps_the_version(engine_up): src = uniq() sid, s1 = await publish(src, V1) run = await run_manager.start(graph=loop_graph(src, src, FAILS_AT_RUNTIME), input_payload={}) assert (await wait(run.id, ("failed", "succeeded"))).status == "failed" await publish(src, V2) await run_manager.continue_failed(run.id, graph=loop_graph(src, src, "false")) done = await wait(run.id, ("succeeded", "failed")) assert done.status == "succeeded", done.error assert output_total(done, "t1") == 3 assert output_total(done, "t2") == 3, "继续运行时改用了重传的新版本" assert [e.data["data_versions"] for e in await started_events(run.id)] == [ {sid: {"snapshot": s1, "name": src}}] * 2 # -------------------------------------------------------------------------- # 工具层：按固定的版本生成、固定的版本没了就报错 # -------------------------------------------------------------------------- async def test_tools_follow_the_pinned_snapshot_and_record_data_version(): src = uniq() sid, s1 = await publish(src, V1) await keep(sid, src, s1) _, s2 = await publish(src, V2) pins = {sid: {"snapshot": s1, "name": src}} try: tools = await tools_for([f"db_query__{src}", f"db_schema__{src}"], data_versions=pins) query, schema = tools[f"db_query__{src}"], tools[f"db_schema__{src}"] assert "明细" in query.description and "新增" not in query.description listed = await schema.coroutine() assert "共 1 张表" in listed and "新增" not in listed assert "数量" in await schema.coroutine(table="明细") text = await query.coroutine(sql=SUM) payload = json.loads(text) assert total_of(text) == 3 # data_version 只进查询快照，不进交给模型的结果 assert "data_version" not in payload assert artifact_store.load(payload["artifact"])["data_version"] == s1 # 冻结的表结构是 S1 的那份，不是数据源上现在的 S2 assert set(artifact_store.load(payload["schema_artifact"])["tables"]) == {"明细"} # 不在运行里（工具库试用、data_versions 为 None）用当前版本 current = (await tools_for([f"db_query__{src}"]))[f"db_query__{src}"] assert "新增" in current.description text = await current.coroutine(sql=SUM) assert total_of(text) == 100 assert artifact_store.load(json.loads(text)["artifact"])["data_version"] == s2 finally: await engines.invalidate(sid) async def test_pinned_snapshot_collected_by_gc_gives_a_clear_error_and_never_reads_current(monkeypatch): # 期 3 起回收另外保留最近 SNAPSHOT_KEEP 个快照；调成 0（期 1 的规则），没人引用的 S1 才会在重传后被回收 monkeypatch.setattr(table_versions, "SNAPSHOT_KEEP", 0) src = uniq() sid, s1 = await publish(src, V1) # 没有运行引用 S1：重传后的回收把它标成已回收、删掉文件 await publish(src, V2) async with SessionLocal() as session: snap = await session.get(SourceSnapshot, s1) assert snap.retired_at is not None and not os.path.exists(snap.db_path) try: tools = await tools_for([f"db_query__{src}", f"db_schema__{src}"], data_versions={sid: {"snapshot": s1, "name": src}}) query, schema = tools[f"db_query__{src}"], tools[f"db_schema__{src}"] # 描述不列表名：表名是当前版本的 assert "明细" not in query.description and "新增" not in query.description assert "本次运行固定的数据版本已不存在" in query.description text = await query.coroutine(sql=SUM) assert text.startswith(f"{DATA_UNAVAILABLE}本次运行固定的数据版本已不存在") assert "不会改用当前版本" in text and "100" not in text # Agent 里这句话作为工具结果交回（tool.end），运行面板按「查询失败」开头把那一步标成失败 assert text.startswith("查询失败") listed = await schema.coroutine() assert "本次运行固定的数据版本已不存在" in listed and "新增" not in listed # 版本根本不属于这个源（记录被改坏、别的源的快照）也一样 other = uniq() _, foreign = await publish(other, V2) tools = await tools_for([f"db_query__{src}"], data_versions={sid: {"snapshot": foreign, "name": src}}) text = await tools[f"db_query__{src}"].coroutine(sql=SUM) assert text.startswith(f"{DATA_UNAVAILABLE}本次运行固定的数据版本已不存在") and "100" not in text finally: await engines.invalidate(sid) async def test_snapshot_file_deleted_after_tools_were_built(): """工具建好之后快照文件被删：缓存的连接还开着已删的文件，也不能拿它接着查。""" src = uniq() sid, s1 = await publish(src, V1) await keep(sid, src, s1) await publish(src, V2) try: tools = await tools_for([f"db_query__{src}", f"db_schema__{src}"], data_versions={sid: {"snapshot": s1, "name": src}}) query = tools[f"db_query__{src}"] assert total_of(await query.coroutine(sql=SUM)) == 3 # 连接已建好、进了缓存 async with SessionLocal() as session: path = (await session.get(SourceSnapshot, s1)).db_path os.chmod(path, stat.S_IRUSR | stat.S_IWUSR) os.unlink(path) text = await query.coroutine(sql=SUM) assert text.startswith(f"{DATA_UNAVAILABLE}本次运行固定的数据版本已不存在") and "100" not in text assert (await tools[f"db_schema__{src}"].coroutine()).startswith( f"{DATA_UNAVAILABLE}本次运行固定的数据版本已不存在") finally: await engines.invalidate(sid) async def test_source_first_used_mid_run_is_pinned_and_recorded(): """发起时没算到的源（继续运行时改绑、子工作流中途改过）：第一次用到时固定，记进运行。""" src = uniq() sid, s1 = await publish(src, V1) async with SessionLocal() as session: run = Run(workflow_name="中途补固定", status="running", data_versions={}) session.add(run) await session.commit() run_id = run.id pins: dict = {} seen: list = [] emit = lambda kind, **data: seen.append((kind, data)) # noqa: E731 - 和 NodeContext.emit 一样是同步的 try: tools = await tools_for([f"db_query__{src}"], data_versions=pins, run_id=run_id, emit=emit) assert pins == {sid: {"snapshot": s1, "name": src, "late": True}} async with SessionLocal() as session: assert (await session.get(Run, run_id)).data_versions == pins # 补固定这件事本身进事件流：当次这一段的 run.started 里没有这个源 [(kind, data)] = seen assert kind == EventType.LOG and data["code"] == LATE_PIN_CODE assert (data["source_id"], data["snapshot"], data["late"]) == (sid, s1, True) assert src in data["message"] # 之后重传：同一段执行里（同一个 dict）、恢复后（从运行记录读回）都还是 S1 await publish(src, V2) assert total_of(await tools[f"db_query__{src}"].coroutine(sql=SUM)) == 3 again = await tools_for([f"db_query__{src}"], data_versions=pins, run_id=run_id, emit=emit) assert len(seen) == 1, "已经固定过的源不该再发一遍" assert total_of(await again[f"db_query__{src}"].coroutine(sql=SUM)) == 3 async with SessionLocal() as session: stored = dict((await session.get(Run, run_id)).data_versions) resumed = await tools_for([f"db_query__{src}"], data_versions=stored, run_id=run_id) assert total_of(await resumed[f"db_query__{src}"].coroutine(sql=SUM)) == 3 finally: await engines.invalidate(sid) async def test_late_pin_takes_what_the_run_record_already_has(): """运行记录里已经有这个源（上一段执行补固定的），内存里还没有：认运行记录的，不取当前版本。""" src = uniq() sid, s1 = await publish(src, V1) async with SessionLocal() as session: run = Run(workflow_name="已记过", status="running", data_versions={sid: {"snapshot": s1, "name": src, "late": True}}) session.add(run) await session.commit() run_id = run.id await publish(src, V2) pins: dict = {} seen: list = [] async def emit(kind, **data): # 协程函数也认 seen.append((kind, data)) try: tools = await tools_for([f"db_query__{src}"], data_versions=pins, run_id=run_id, emit=emit) assert pinned_of(pins, sid) == s1 assert seen == [], "上一段执行补固定时已经发过事件" assert total_of(await tools[f"db_query__{src}"].coroutine(sql=SUM)) == 3 finally: await engines.invalidate(sid) async def test_manual_sources_are_not_versioned(tmp_path): sid, name, path = await manual_source(tmp_path, 5) pins: dict = {} try: tools = await tools_for([f"db_query__{name}"], data_versions=pins) text = await tools[f"db_query__{name}"].coroutine(sql=SUM) assert total_of(text) == 5 assert "data_version" not in artifact_store.load(json.loads(text)["artifact"]) assert pins == {}, "手工源不该进 data_versions" # 手工源查的就是那个文件此刻的内容 db = sqlite3.connect(path) db.execute('UPDATE "明细" SET "数量" = 9') db.commit() db.close() assert total_of(await tools[f"db_query__{name}"].coroutine(sql=SUM)) == 9 finally: await engines.invalidate(sid) # -------------------------------------------------------------------------- # 子工作流、漏传版本、调用工具节点的报错、图级 defaults、补固定的事件 # -------------------------------------------------------------------------- async def test_subgraph_resumed_after_approval_reads_the_pinned_version(engine_up): """子工作流的工具经它自己的 RunContext 拿版本：父运行停在审批时重传，恢复后子工作流查的仍是 S1。""" src = uniq() sid, s1 = await publish(src, V1) async with SessionLocal() as session: sub = Workflow(name="子流程取数", graph=chain( node("s", "input"), node("q", "tool", tool=f"db_query__{src}", args={"sql": SUM}), node("o", "output", fields=[{"name": "t", "value": "{{ nodes.q }}"}]))) session.add(sub) await session.commit() sub_id = sub.id graph = chain( node("start", "input"), node("gate", "human", mode="approve", title="看一眼"), node("sub", "subgraph", workflow_id=sub_id), node("out", "output", fields=[{"name": "sub", "value": "{{ nodes.sub.output.t }}"}]), ) try: run = await run_manager.start(graph=graph, input_payload={}) assert run.data_versions == {sid: {"snapshot": s1, "name": src}} assert (await wait(run.id)).status == "interrupted" _, s2 = await publish(src, V2) assert s2 != s1 await run_manager.resume(run.id, {"approved": True}) done = await wait(run.id, ("succeeded", "failed")) assert done.status == "succeeded", done.error assert output_total(done, "sub") == 3, "子工作流恢复后查到了重传的新版本" payload = json.loads((done.output or {})["sub"]) assert artifact_store.load(payload["artifact"])["data_version"] == s1 # 拿到的是发起时固定的那份，不是第一次用到时才补的 assert done.data_versions == {sid: {"snapshot": s1, "name": src}} assert await late_pin_events(run.id) == [] finally: await engines.invalidate(sid) async def test_tools_built_without_versions_refuse_upload_sources(tmp_path): """构造 ToolContext 时没给版本（在运行里漏传了）：上传源报错，不当成「不在运行里」改用当前版本。""" src = uniq() sid, _ = await publish(src, V1) spec = GraphSpec.model_validate(chain(node("start", "input"), node("out", "output"))) # 运行上下文里没有这个键（不是 runner 建的，或往下传时丢了）：UNSET，不是 None assert run_versions(RunContext(run_id="r", thread_id="r", spec=spec)) is Versions.UNSET assert run_versions(RunContext(run_id="r", thread_id="r", spec=spec, extra={RUN_VERSIONS_KEY: None})) is None pins: dict = {} assert run_versions(RunContext(run_id="r", thread_id="r", spec=spec, extra={RUN_VERSIONS_KEY: pins})) is pins assert ToolContext().data_versions is Versions.UNSET manual_id, manual, _ = await manual_source(tmp_path, 5) try: async with SessionLocal() as session: built = {t.name: t for t in await build_datasource_tools( [f"db_query__{src}", f"db_schema__{src}", f"db_query__{manual}"], ToolContext(run_id="r", node_id="n"), session)} query, schema = built[f"db_query__{src}"], built[f"db_schema__{src}"] assert "没有拿到本次运行固定的数据版本" in query.description and "明细" not in query.description text = await query.coroutine(sql=SUM) assert text.startswith(f"{DATA_UNAVAILABLE}没有拿到本次运行固定的数据版本") and "rows" not in text assert (await schema.coroutine()).startswith(f"{DATA_UNAVAILABLE}没有拿到本次运行固定的数据版本") # 手工源不分版本，照常查 assert total_of(await built[f"db_query__{manual}"].coroutine(sql=SUM)) == 5 finally: await engines.invalidate(sid) await engines.invalidate(manual_id) @pytest.mark.parametrize("prefix", ["db_query__", "db_schema__"]) async def test_tool_node_fails_with_the_reason_when_its_pinned_snapshot_is_gone(engine_up, prefix): """停在审批时重传、再删掉 S1 的文件：恢复后调用工具节点判失败，报错照交原因，不让人改 SQL。""" src = uniq() sid, s1 = await publish(src, V1) tool = f"{prefix}{src}" graph = chain( node("start", "input"), node("gate", "human", mode="approve", title="看一眼"), node("t", "tool", tool=tool, args={"sql": SUM} if prefix == "db_query__" else {}), node("out", "output", fields=[{"name": "t", "value": "{{ nodes.t }}"}]), ) try: run = await run_manager.start(graph=graph, input_payload={}) assert (await wait(run.id)).status == "interrupted" await publish(src, V2) # 这次运行引用着 S1，回收不删它 async with SessionLocal() as session: path = (await session.get(SourceSnapshot, s1)).db_path os.chmod(path, stat.S_IRUSR | stat.S_IWUSR) os.unlink(path) await run_manager.resume(run.id, {"approved": True}) done = await wait(run.id, ("succeeded", "failed")) assert done.status == "failed" error = done.error or "" assert f"工具 {tool} 无法使用：本次运行固定的数据版本已不存在（数据源「{src}」）" in error, error assert "请重新发起运行" in error # 改 SQL 没用，不能这么指；原话的开头也不能和外面那层叠成两遍 assert "SQL" not in error and "查询失败" not in error and error.count(DATA_UNAVAILABLE) == 0, error [err] = await events_of(run.id, "tool.error", "t") assert err.data["error"].startswith(f"{DATA_UNAVAILABLE}本次运行固定的数据版本已不存在") assert not await events_of(run.id, "tool.end", "t") finally: await engines.invalidate(sid) async def test_tools_bound_through_graph_defaults_are_pinned_at_start(): """Agent 节点没写 tools、协作节点没写成员时运行时取图级 defaults，发起时固定版本也按它算。""" agent_src, member_src = uniq(), uniq() agent_id, agent_snap = await publish(agent_src, V1) member_id, member_snap = await publish(member_src, V1) graph = chain(node("start", "input"), node("bot", "agent", prompt="查一下"), node("team", "supervisor", goal="查合计"), node("out", "output")) graph["defaults"] = {"tools": [f"db_query__{agent_src}"], "agents": [{"name": "analyst", "tools": [f"db_schema__{member_src}"]}]} async with SessionLocal() as session: assert await _data_versions(session, GraphSpec.model_validate(graph)) == { agent_id: {"snapshot": agent_snap, "name": agent_src}, member_id: {"snapshot": member_snap, "name": member_src}} # 发起前检查是同一个取法：defaults 里绑的工具不存在，发起前就拦下 ghost = uniq("ghost") broken = {**graph, "defaults": {"tools": [f"db_query__{ghost}"]}} missing = await missing_tools(session, GraphSpec.model_validate(broken)) assert [(name, where) for _, name, where in missing] == [(f"db_query__{ghost}", "data")] # 节点上写了（哪怕是空列表）就不看 defaults：和 NodeContext.cfg 一样 graph["nodes"][1]["data"]["config"]["tools"] = [] assert set(await _data_versions(session, GraphSpec.model_validate(graph))) == {member_id} async def test_source_first_used_after_continue_is_pinned_and_leaves_an_event(engine_up): """继续运行时改绑了发起时没用到的源：第一次用到时补固定，记进运行，事件流里留一条日志。""" first, later = uniq(), uniq() first_id, first_snap = await publish(first, V1) later_id, later_snap = await publish(later, V1) try: run = await run_manager.start(graph=loop_graph(first, first, FAILS_AT_RUNTIME), input_payload={}) failed = await wait(run.id, ("failed", "succeeded")) assert failed.status == "failed" assert failed.data_versions == {first_id: {"snapshot": first_snap, "name": first}} await run_manager.continue_failed(run.id, graph=loop_graph(first, later, "false")) done = await wait(run.id, ("succeeded", "failed")) assert done.status == "succeeded", done.error assert done.data_versions[later_id] == {"snapshot": later_snap, "name": later, "late": True} # 这一段的 run.started 里没有它（发起这一段时还没用到）：补固定的那条日志是事件流里唯一的记录 assert later_id not in (await started_events(run.id))[-1].data["data_versions"] [event] = await late_pin_events(run.id) assert event.node_id == "t2" assert (event.data["source_id"], event.data["snapshot"], event.data["late"]) == (later_id, later_snap, True) finally: await engines.invalidate(first_id) await engines.invalidate(later_id) async def test_continue_after_the_pinned_source_is_deleted_and_recreated_under_the_same_name(engine_up): """固定的源被删、又用同一个名字新建：续跑报错，不改查同名的新源，也不把新源当成中途新建的补固定。""" from httpx import ASGITransport, AsyncClient from app.main import app src = uniq() a_id, s1 = await publish(src, V1) b_id = None try: run = await run_manager.start(graph=loop_graph(src, src, FAILS_AT_RUNTIME), input_payload={}) failed = await wait(run.id, ("failed", "succeeded")) assert failed.status == "failed" assert failed.data_versions == {a_id: {"snapshot": s1, "name": src}} async with AsyncClient(transport=ASGITransport(app=app), base_url="http://t") as c: assert (await c.delete(f"/api/datasources/{a_id}")).status_code == 204 b_id, _ = await publish(src, V2) assert b_id != a_id await run_manager.continue_failed(run.id, graph=loop_graph(src, src, "false")) done = await wait(run.id, ("succeeded", "failed")) assert done.status == "failed", f"续跑改查了同名重建的新源：{done.output}" error = done.error or "" assert f"本次运行固定的数据源「{src}」已被删除或重建，不会改用同名的新数据源" in error, error assert "SQL" not in error and "100" not in error # 新源没有被补固定进运行记录，也没有补固定的日志 assert set(done.data_versions) == {a_id} assert not await late_pin_events(run.id) finally: await engines.invalidate(a_id) if b_id: await engines.invalidate(b_id) async def test_same_name_manual_source_does_not_stand_in_for_a_pinned_upload_source(tmp_path): """同名重建成手工源也拦住：手工源不分版本，但运行固定的是另一个源，不能拿它顶上。""" _, manual, _ = await manual_source(tmp_path, 5) pins = {"gone-source-id": {"snapshot": "0" * 64, "name": manual}} tools = await tools_for([f"db_query__{manual}", f"db_schema__{manual}"], data_versions=pins, run_id="r") text = await tools[f"db_query__{manual}"].coroutine(sql=SUM) assert text.startswith(f"{DATA_UNAVAILABLE}本次运行固定的数据源「{manual}」已被删除或重建"), text assert "5" not in text.replace(manual, "") assert (await tools[f"db_schema__{manual}"].coroutine()).startswith(DATA_UNAVAILABLE) # 运行里没固定过这个名字的手工源照常查 tools = await tools_for([f"db_query__{manual}"], data_versions={}, run_id="r") assert total_of(await tools[f"db_query__{manual}"].coroutine(sql=SUM)) == 5 def _tamper(path: str) -> None: """改掉库文件末尾一个字节（发布后是 0444，先放开写）。""" os.chmod(path, stat.S_IRUSR | stat.S_IWUSR) with open(path, "r+b") as f: f.seek(-1, os.SEEK_END) last = f.read(1) f.seek(-1, os.SEEK_END) f.write(bytes([last[0] ^ 0xFF])) async def test_tampered_snapshot_is_reported_as_data_unavailable_not_as_a_sql_error(engine_up): """快照文件被改过（哈希对不上）：工具按「数据不可用」交回，调用工具节点不让人去改 SQL。""" src = uniq() sid, s1 = await publish(src, V1) graph = chain(node("start", "input"), node("t", "tool", tool=f"db_query__{src}", args={"sql": SUM}), node("out", "output", fields=[{"name": "t", "value": "{{ nodes.t }}"}])) try: async with SessionLocal() as session: path = (await session.get(SourceSnapshot, s1)).db_path await engines.invalidate(sid) # 下次查询新建引擎，建之前核对哈希 _tamper(path) for pins in (Versions.CURRENT, {sid: {"snapshot": s1, "name": src}}): tools = await tools_for([f"db_query__{src}"], data_versions=pins, run_id="r") text = await tools[f"db_query__{src}"].coroutine(sql=SUM) assert text.startswith(f"{DATA_UNAVAILABLE}数据源「{src}」的数据文件与登记的版本不一致"), text assert "SQL" not in text and "请联系管理员" in text run = await run_manager.start(graph=graph, input_payload={}) done = await wait(run.id, ("succeeded", "failed")) assert done.status == "failed" error = done.error or "" assert error.startswith(f"工具 db_query__{src} 无法使用：数据源「{src}」的数据文件与登记的版本不一致"), error assert "SQL" not in error and "查询失败" not in error, error finally: await engines.invalidate(sid) async def test_reupload_while_a_run_is_being_started_does_not_collect_its_snapshot(engine_up, monkeypatch): """发起运行读完指针、运行记录还没提交时有人重传：发布后的回收看不到这条运行，以前会把刚固定的快照删掉， 这次运行只能报「固定的数据版本已不存在」。现在读指针到提交持着版本存储的锁，回收等它提交之后才开始。""" from app.engine import runner src = uniq() sid, s1 = await publish(src, V1) real = runner._data_versions pending: list[asyncio.Task] = [] async def racing(session, spec): pins = await real(session, spec) # 读完指针的这一刻有人重传。给它足够的时间做完（没有锁的话它会做完，连同回收） task = asyncio.create_task(publish(src, V2)) pending.append(task) await asyncio.wait({task}, timeout=0.5) return pins monkeypatch.setattr(runner, "_data_versions", racing) graph = chain(node("start", "input"), node("t", "tool", tool=f"db_query__{src}", args={"sql": SUM}), node("out", "output", fields=[{"name": "t", "value": "{{ nodes.t }}"}])) try: run = await run_manager.start(graph=graph, input_payload={}) _, s2 = await pending[0] assert run.data_versions == {sid: {"snapshot": s1, "name": src}} and s2 != s1 async with SessionLocal() as session: snap = await session.get(SourceSnapshot, s1) assert snap.retired_at is None and os.path.exists(snap.db_path), "回收删掉了刚固定的快照" done = await wait(run.id, ("succeeded", "failed")) assert done.status == "succeeded", done.error assert output_total(done, "t") == 3 finally: await engines.invalidate(sid) async def test_late_pin_reads_the_pointer_again_under_the_lock(monkeypatch): """补固定时数据源记录是建工具前读的：那之后重传过、旧快照已回收，不能拿旧指针去固定。 在版本存储的锁里重新读指针，固定的是此刻的当前版本，而且回收不会在写进运行记录之前删掉它。""" from app.tools import datasource src = uniq() sid, s1 = await publish(src, V1) async with SessionLocal() as session: run = Run(workflow_name="补固定撞上重传", status="running", data_versions={}) session.add(run) await session.commit() run_id = run.id real = datasource._late_lock reuploaded: list[str] = [] class Racing: """进补固定的锁之前，先让一次重传整个做完（发布、切指针、回收掉没人引用的 S1）。""" async def __aenter__(self): reuploaded.append((await publish(src, V2))[1]) self.lock = real() await self.lock.acquire() async def __aexit__(self, *exc): self.lock.release() monkeypatch.setattr(datasource, "_late_lock", lambda: Racing()) pins: dict = {} try: tools = await tools_for([f"db_query__{src}"], data_versions=pins, run_id=run_id) [s2] = reuploaded assert pinned_of(pins, sid) == s2, "补固定用了建工具前读到的旧指针" text = await tools[f"db_query__{src}"].coroutine(sql=SUM) assert not text.startswith(DATA_UNAVAILABLE), text assert total_of(text) == 100 async with SessionLocal() as session: assert pinned_of((await session.get(Run, run_id)).data_versions, sid) == s2 finally: await engines.invalidate(sid)