From 0eaaf713469e84876a3f838b789437b158fc6f0f Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Mon, 5 Oct 2026 00:07:07 +0800 Subject: [PATCH] =?UTF-8?q?=E8=BF=90=E8=A1=8C=E8=AE=B0=E4=B8=8B=E9=80=89?= =?UTF-8?q?=E9=A1=B9=E5=BF=AB=E7=85=A7=E5=92=8C=E6=9D=A5=E6=BA=90=EF=BC=9A?= =?UTF-8?q?runs=20=E5=8A=A0=20options=EF=BC=88JSON=EF=BC=8C=E9=BB=98?= =?UTF-8?q?=E8=AE=A4=20{}=EF=BC=89=E5=92=8C=20schedule=5Fid=EF=BC=88?= =?UTF-8?q?=E5=8F=AF=E7=A9=BA=EF=BC=8C=E5=BB=BA=E7=B4=A2=E5=BC=95=EF=BC=89?= =?UTF-8?q?=E4=B8=A4=E5=88=97=E3=80=82=E5=90=8E=E9=9D=A2=E7=9A=84=E6=A0=B8?= =?UTF-8?q?=E6=9F=A5=E8=AE=BE=E7=BD=AE=E8=A6=81=E5=9C=A8=E5=8F=91=E8=B5=B7?= =?UTF-8?q?=E6=97=B6=E5=AE=9A=E4=B8=8B=E3=80=81=E6=95=B4=E6=AC=A1=E8=BF=90?= =?UTF-8?q?=E8=A1=8C=E6=B2=BF=E7=94=A8=EF=BC=8C=E5=AE=A1=E6=89=B9=E5=81=9C?= =?UTF-8?q?=E4=B8=8B=E6=9C=9F=E9=97=B4=E6=94=B9=E4=BA=86=E8=AE=BE=E7=BD=AE?= =?UTF-8?q?=EF=BC=8C=E6=81=A2=E5=A4=8D=E5=92=8C=E9=87=8D=E6=94=BE=E4=BB=8D?= =?UTF-8?q?=E6=8C=89=E5=8F=91=E8=B5=B7=E6=97=B6=E7=9A=84=EF=BC=88=E5=90=8C?= =?UTF-8?q?=20approval=5Fdefault=20=E7=9A=84=E9=81=93=E7=90=86=EF=BC=89?= =?UTF-8?q?=EF=BC=9B=E5=AE=9A=E6=97=B6=E4=BB=BB=E5=8A=A1=E8=A6=81=E8=83=BD?= =?UTF-8?q?=E6=89=BE=E5=9B=9E=E8=87=AA=E5=B7=B1=E5=8F=91=E8=B5=B7=E7=9A=84?= =?UTF-8?q?=E8=BF=90=E8=A1=8C=E3=80=82RunManager.start=20=E6=94=B6=20optio?= =?UTF-8?q?ns=E3=80=81schedule=5Fid=20=E5=B9=B6=E8=90=BD=E5=BA=93=EF=BC=8C?= =?UTF-8?q?=E9=80=89=E9=A1=B9=E5=85=88=E6=B7=B1=E6=8B=B7=E4=B8=80=E4=BB=BD?= =?UTF-8?q?=EF=BC=8C=E8=B0=83=E7=94=A8=E6=96=B9=E4=B9=8B=E5=90=8E=E6=94=B9?= =?UTF-8?q?=E8=87=AA=E5=B7=B1=E7=9A=84=20dict=20=E4=B8=8D=E5=BD=B1?= =?UTF-8?q?=E5=93=8D=E5=BF=AB=E7=85=A7=EF=BC=9B=E9=80=89=E9=A1=B9=E9=9D=9E?= =?UTF-8?q?=E7=A9=BA=E6=97=B6=E5=86=99=E8=BF=9B=E6=AF=8F=E4=B8=80=E6=AE=B5?= =?UTF-8?q?=E7=9A=84=20run.started=EF=BC=88=E8=BF=9B=E5=B0=81=E5=AD=98?= =?UTF-8?q?=E8=8C=83=E5=9B=B4=EF=BC=89=EF=BC=8C=E4=B8=BA=E7=A9=BA=E6=97=B6?= =?UTF-8?q?=20run.started=20=E5=92=8C=E4=BB=A5=E5=89=8D=E9=80=90=E5=AD=97?= =?UTF-8?q?=E7=9B=B8=E5=90=8C=EF=BC=9B=E6=81=A2=E5=A4=8D=E3=80=81=E7=BB=A7?= =?UTF-8?q?=E7=BB=AD=E8=BF=90=E8=A1=8C=E7=85=A7=E8=BF=90=E8=A1=8C=E8=AE=B0?= =?UTF-8?q?=E5=BD=95=E6=B2=BF=E7=94=A8=EF=BC=8C=E8=8A=82=E7=82=B9=E4=BB=8E?= =?UTF-8?q?=20RunContext.extra=20=E7=9A=84=20run=5Foptions=20=E8=AF=BB?= =?UTF-8?q?=E5=90=8C=E4=B8=80=E4=BB=BD=E6=8B=B7=E8=B4=9D=E3=80=82RunOut=20?= =?UTF-8?q?=E5=B8=A6=20schedule=5Fid=EF=BC=8CGET=20/api/runs=20=E5=8F=AF?= =?UTF-8?q?=E6=8C=89=20schedule=5Fid=20=E7=AD=9B=EF=BC=8Claunch=5Frun=20?= =?UTF-8?q?=E5=8E=9F=E6=A0=B7=E5=BE=80=E4=B8=8B=E4=BA=A4=E3=80=82=E6=97=A7?= =?UTF-8?q?=E5=BA=93=E8=B5=B0=20=5FCOLUMN=5FMIGRATIONS=20=E8=A1=A5?= =?UTF-8?q?=E5=88=97=EF=BC=8C=E5=AD=98=E9=87=8F=E8=BF=90=E8=A1=8C=E5=9B=9E?= =?UTF-8?q?=E5=A1=AB=20{}=20=E5=92=8C=20NULL=EF=BC=9Bcreate=5Fall=20?= =?UTF-8?q?=E4=B8=8D=E7=A2=B0=E5=B7=B2=E6=9C=89=E7=9A=84=E8=A1=A8=EF=BC=8C?= =?UTF-8?q?=E6=96=B0=E5=88=97=E7=9A=84=E7=B4=A2=E5=BC=95=E5=8F=A6=E7=94=B1?= =?UTF-8?q?=20=5FINDEX=5FMIGRATIONS=20=E8=A1=A5=E5=BB=BA=EF=BC=8C=E7=B4=A2?= =?UTF-8?q?=E5=BC=95=E5=90=8D=E5=92=8C=E6=96=B0=E5=BA=93=20create=5Fall=20?= =?UTF-8?q?=E5=BB=BA=E5=87=BA=E6=9D=A5=E7=9A=84=E7=9B=B8=E5=90=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/runs.py | 5 + backend/app/db/base.py | 15 ++ backend/app/db/models.py | 5 + backend/app/engine/launch.py | 5 + backend/app/engine/runner.py | 25 ++- backend/tests/test_run_options.py | 268 ++++++++++++++++++++++++++++++ 6 files changed, 321 insertions(+), 2 deletions(-) create mode 100644 backend/tests/test_run_options.py diff --git a/backend/app/api/runs.py b/backend/app/api/runs.py index 058ad518..00fd3637 100644 --- a/backend/app/api/runs.py +++ b/backend/app/api/runs.py @@ -68,6 +68,8 @@ class RunOut(BaseModel): #: 发起时实际生效的记忆域和知识库 memory_scope: str | None = None collection: str | None = None + #: 发起它的定时任务。为空 = 手动发起 + schedule_id: str | None = None # 时间一律带时区（db.base.UTCDateTime 读出来就是 UTC aware） created_at: Any = None started_at: Any = None @@ -118,6 +120,7 @@ async def list_runs( status: str | None = Query(default=None, description="逗号分隔，如 failed,cancelled"), run_class: str | None = None, q: str | None = Query(default=None, description="按工作流名模糊匹配"), + schedule_id: str | None = Query(default=None, description="只要这个定时任务发起的运行"), limit: int = Query(default=50, ge=1, le=200), before: datetime | None = Query(default=None, description="翻页游标：只要 created_at 早于它的"), session: AsyncSession = Depends(get_session), @@ -138,6 +141,8 @@ async def list_runs( stmt = stmt.where(Run.run_class.in_(classes)) if q and q.strip(): stmt = stmt.where(Run.workflow_name.ilike(f"%{_like(q.strip())}%", escape="\\")) + if schedule_id: + stmt = stmt.where(Run.schedule_id == schedule_id) if before is not None: stmt = stmt.where(Run.created_at < before) return list((await session.execute(stmt)).scalars()) diff --git a/backend/app/db/base.py b/backend/app/db/base.py index 2f1e5a4a..68ee478b 100644 --- a/backend/app/db/base.py +++ b/backend/app/db/base.py @@ -200,6 +200,17 @@ _COLUMN_MIGRATIONS: list[tuple[str, str, str]] = [ ("import_stagings", "edits", "JSON"), ("import_stagings", "answers_dropped", "JSON"), ("import_stagings", "redraft_sha256", "VARCHAR(64)"), + # 运行选项（发起时的快照，整次运行沿用）和发起它的定时任务。存量运行回填 {} 和 NULL： + # 没有选项，也不是定时任务发起的 + ("runs", "options", "JSON DEFAULT '{}'"), + ("runs", "schedule_id", "VARCHAR(32)"), +] + +#: 迁移补上的列要建的索引：(索引名, 表, 列)。create_all 不碰已有的表，表上新列的索引也就不会建。 +#: 索引名照 NAMING_CONVENTION 取，和新库 create_all 建出来的同名，IF NOT EXISTS 才不会重复建 +_INDEX_MIGRATIONS: list[tuple[str, str, str]] = [ + # 定时任务的触发历史按它找这个任务发起的运行 + ("ix_runs_schedule_id", "runs", "schedule_id"), ] @@ -210,6 +221,10 @@ def _migrate(conn) -> None: # pragma: no cover - 同步回调 cols = {row[1] for row in conn.execute(text(f"PRAGMA table_info({table})"))} if cols and column not in cols: conn.execute(text(f"ALTER TABLE {table} ADD COLUMN {column} {ddl}")) + for name, table, column in _INDEX_MIGRATIONS: + cols = {row[1] for row in conn.execute(text(f"PRAGMA table_info({table})"))} + if column in cols: + conn.execute(text(f"CREATE INDEX IF NOT EXISTS {name} ON {table} ({column})")) async def init_db() -> None: diff --git a/backend/app/db/models.py b/backend/app/db/models.py index 403b9177..3d7cc860 100644 --- a/backend/app/db/models.py +++ b/backend/app/db/models.py @@ -491,6 +491,11 @@ class Run(Base, TimestampMixin): # 发起时钉住的上传表格版本：{source_id: {"snapshot": 快照 id, "name": 源名}}。续跑、恢复沿用， # 回收（table_versions.gc）不删任何运行引用的快照。None 是升级前发起的运行 data_versions: Mapped[dict[str, Any] | None] = mapped_column(default=None) + # 发起时定下、整次运行沿用的运行选项（比如核查设置的快照）。审批停下期间改了设置，恢复和重放 + # 仍按发起时的这份——同 approval_default 的道理。非空时写进 run.started，进封存范围。存量运行是 {} + options: Mapped[dict[str, Any]] = mapped_column(default=dict) + # 发起它的定时任务，手动发起的为空。不设外键：定时任务删掉之后，运行记录照旧是追溯凭证，来源 id 留着 + schedule_id: Mapped[str | None] = mapped_column(String(32), default=None, index=True) events: Mapped[list["RunEvent"]] = relationship( back_populates="run", cascade="all, delete-orphan" diff --git a/backend/app/engine/launch.py b/backend/app/engine/launch.py index 3186a690..71d7a9aa 100644 --- a/backend/app/engine/launch.py +++ b/backend/app/engine/launch.py @@ -52,10 +52,13 @@ async def launch_run( run_class: str = "exploratory", version: int | None = None, actor: str | None, + schedule_id: str | None = None, + options: dict[str, Any] | None = None, ) -> Run: """发起一次运行，返回刚建好的运行记录（执行已经交给 run_manager）。不发起就抛 LaunchRefused。 参数和 POST /api/runs 的请求体一一对应（input 叫 input_payload）；actor 是记在运行上的发起人。 + schedule_id 是发起它的定时任务，options 是整次运行沿用的选项快照，都原样交给 run_manager.start。 """ # 没挂在工作流上（画布没保存就发起、问数据当场搭的）。老数据里是「临时图」，前端两种都认 name = "未保存的工作流" @@ -146,6 +149,8 @@ async def launch_run( version=run_version, version_hash=version_hash, started_by=actor, + options=options, + schedule_id=schedule_id, ) except ValueError as e: raise LaunchRefused(400, str(e)) from e diff --git a/backend/app/engine/runner.py b/backend/app/engine/runner.py index 74151f86..b5462af4 100644 --- a/backend/app/engine/runner.py +++ b/backend/app/engine/runner.py @@ -2,6 +2,7 @@ from __future__ import annotations import asyncio import contextlib +import copy import json import logging import time @@ -33,6 +34,10 @@ logger = logging.getLogger(__name__) # 轨迹里正文靠 node.finished 的 preview，思考靠调用结束时的单条 llm.thinking 汇总 _EPHEMERAL = {EventType.LLM_TOKEN, EventType.LLM_THINKING_DELTA} +#: RunContext.extra 里放这次运行的选项（Run.options）的键。节点从这里读发起时定下的快照、不读设置的 +#: 当前值：审批停下期间改了设置，恢复后的那一段也按发起时的，重放和正式运行才可复现 +RUN_OPTIONS_KEY = "run_options" + class RunManager: """负责一次运行的完整生命周期：启动、串流、中断、恢复、取消。""" @@ -177,7 +182,10 @@ class RunManager: version_hash: str | None = None, started_by: str | None = None, approval_default: str | None = None, + options: dict[str, Any] | None = None, + schedule_id: str | None = None, ) -> Run: + """建运行记录、交给执行任务。options 是这次运行的选项快照，schedule_id 是发起它的定时任务（可空）。""" spec = GraphSpec.model_validate(graph) report = validate_graph(spec) if not report.ok: @@ -187,6 +195,8 @@ class RunManager: ) if approval_default is None: approval_default = await _approval_default(run_class=run_class) + # 拷一份：调用方手上的 dict 之后再改，不该改到这次运行的快照 + options = copy.deepcopy(options) if options else {} async with SessionLocal() as session: from app.data.table_versions import pin_lock @@ -214,6 +224,8 @@ class RunManager: tool_trust=tool_trust, agent_limits=agent_limits, data_versions=data_versions, + options=options, + schedule_id=schedule_id, ) run.thread_id = run.id # 一个 run 一条 checkpoint 线程 session.add(run) @@ -232,6 +244,7 @@ class RunManager: tool_trust=tool_trust, agent_limits=agent_limits, data_versions=data_versions, + options=options, ) ) return run @@ -585,6 +598,7 @@ class RunManager: tool_trust: dict[str, str] | None = None, agent_limits: dict[str, Any] | None = None, data_versions: dict[str, Any] | None = None, + options: dict[str, Any] | None = None, resumed: list[str] | tuple[str, ...] = (), actors: dict[str, str | None] | None = None, ) -> None: @@ -631,7 +645,10 @@ class RunManager: # 运行记录里的那份，含运行中途第一次用到时补上的；None 是升级前的运行 "data_versions": data_versions, # 这一段是按哪一版重放协议跑的，恢复时据此认出升级前做下的工作 - "replay_protocol": PROTOCOL}, + "replay_protocol": PROTOCOL, + # 发起时定下的运行选项（核查设置之类），每一段都记、记的是同一份。没有选项的 + # 运行不带这个键，和加选项之前的 run.started 逐字相同 + **({"options": options} if options else {})}, ) run_ctx = RunContext( @@ -648,7 +665,9 @@ class RunManager: # 的 run_versions）。复制一份：运行中途补固定的源就地写进这个 dict，同一段执行里 # 后面的节点都看得到，但不该碰到别处（运行记录、事件）手上的那个对象 extra={"resumed": set(resumed), "actors": dict(actors or {}), - RUN_VERSIONS_KEY: dict(data_versions) if data_versions is not None else None}, + RUN_VERSIONS_KEY: dict(data_versions) if data_versions is not None else None, + # 同理也是一份拷贝：节点读它，不该改到运行记录那份 + RUN_OPTIONS_KEY: copy.deepcopy(options or {})}, ) app = compile_graph(spec, run_ctx).compile(checkpointer=self.checkpointer) # 循环按自己声明的轮数另记一份预算，全局上限只管没人把关的环 @@ -1025,6 +1044,8 @@ def _carried(run: Run) -> dict[str, Any]: "agent_limits": run.agent_limits, # 固定的数据版本照运行记录来，不重新读数据源的当前指针：中途重传过，续跑查的仍是发起时那一版 "data_versions": run.data_versions, + # 运行选项同理：照发起时的快照，不重新读设置 + "options": run.options or {}, } diff --git a/backend/tests/test_run_options.py b/backend/tests/test_run_options.py new file mode 100644 index 00000000..ef456a27 --- /dev/null +++ b/backend/tests/test_run_options.py @@ -0,0 +1,268 @@ +"""运行选项和来源：Run.options 是发起时定下、整次运行沿用的选项快照，Run.schedule_id 是发起它的定时任务。 + +- 旧库升级时补这两列和 schedule_id 的索引；存量运行的选项读出来是 {}，来源为空 +- RunManager.start 收下两者并落库；选项非空时写进 run.started（进封存范围），为空时 run.started 和以前 + 逐字相同；恢复、继续运行沿用发起时的选项，节点从 RunContext.extra 读得到同一份 +- 接口带出 schedule_id，运行列表能按定时任务筛；launch_run 原样往下交 +""" +from __future__ import annotations + +import asyncio +import uuid + +import pytest +from httpx import ASGITransport, AsyncClient +from sqlalchemy import create_engine, select, text + +from app.db.base import SessionLocal, engine as app_engine +from app.db.models import Run, RunEvent +from app.engine import compiler +from app.engine.runner import run_manager +from app.engine.schema import NodeType +from app.main import app + +#: 加 options 之前 run.started 就有的键：选项为空时一个不多、一个不少 +LEGACY_STARTED_KEYS = {"nodes", "resumed", "memory_scope", "collection", "approval_default", "tool_trust", + "agent_limits", "data_versions", "replay_protocol"} +OPTIONS = {"verify": {"dual_check": True, "max_queries": 6}, "note": "第 40 周例行核查"} + + +@pytest.fixture(autouse=True) +async def engine_up(): + await run_manager.setup() + yield + await run_manager.shutdown() + + +@pytest.fixture +async def client(): + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + yield c + + +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:])]} + + +SIMPLE = chain(node("start", "input"), node("out", "output", fields=[{"name": "r", "value": "完成"}])) +GATED = chain(node("start", "input"), + node("before", "transform", mode="template", template="前"), + node("gate", "human", mode="approve", title="放行吗"), + node("after", "transform", mode="template", template="后"), + node("out", "output", fields=[{"name": "r", "value": "{{ nodes.after }}"}])) + + +async def _settle(run_id: str, statuses=("succeeded", "failed", "cancelled")) -> Run: + for _ in range(400): + async with SessionLocal() as session: + row = await session.get(Run, run_id) + if row.status in statuses and (row.status == "interrupted" or row.manifest_seq is not None): + await run_manager.wait_idle(run_id) + return row + await asyncio.sleep(0.03) + raise AssertionError(f"运行 {run_id} 停在 {row.status}") + + +async def _started(run_id: str) -> list[dict]: + async with SessionLocal() as session: + return [e.data for e in (await session.execute( + select(RunEvent).where(RunEvent.run_id == run_id, RunEvent.type == "run.started").order_by(RunEvent.seq) + )).scalars()] + + +def _schedule() -> str: + return uuid.uuid4().hex + + +# -------------------------------------------------------------------------- +# 旧库升级 +# -------------------------------------------------------------------------- + + +def test_old_databases_get_both_columns_and_the_index(): + from app.db.base import _migrate + + db = create_engine("sqlite://") + with db.begin() as conn: + conn.execute(text("CREATE TABLE runs (id VARCHAR(32) PRIMARY KEY, workflow_name VARCHAR(200), " + "status VARCHAR(20), last_seq INTEGER)")) + conn.execute(text("INSERT INTO runs (id, workflow_name, status, last_seq) VALUES ('r1', '周报', 'succeeded', 9)")) + _migrate(conn) + _migrate(conn) # 再执行一次不出错：列和索引都已经在了 + cols = {row[1] for row in conn.execute(text("PRAGMA table_info(runs)"))} + assert {"options", "schedule_id"} <= cols + assert conn.execute(text("SELECT options, schedule_id FROM runs WHERE id = 'r1'")).one() == ("{}", None) + indexes = {row[1]: row for row in conn.execute(text("PRAGMA index_list(runs)"))} + assert "ix_runs_schedule_id" in indexes + assert [r[2] for r in conn.execute(text("PRAGMA index_info(ix_runs_schedule_id)"))] == ["schedule_id"] + + +def test_a_missing_table_is_left_alone(): + """还没有 runs 表的库（create_all 之前不会这样，这里只确认迁移不替它建索引、不报错）。""" + from app.db.base import _migrate + + db = create_engine("sqlite://") + with db.begin() as conn: + _migrate(conn) + assert conn.execute(text("SELECT name FROM sqlite_master WHERE name = 'ix_runs_schedule_id'")).first() is None + + +async def test_new_databases_index_schedule_id(): + async with app_engine.connect() as conn: + names = {row[1] for row in (await conn.execute(text("PRAGMA index_list(runs)"))).all()} + assert "ix_runs_schedule_id" in names + + +async def test_a_migrated_row_reads_back_empty_options(): + """迁移回填的是字符串 '{}'：经 ORM 读出来是空 dict，不是字符串，也不是 None。""" + async with SessionLocal() as session: + run = Run(workflow_name="迁移回填", status="succeeded", graph={}, input={}) + session.add(run) + await session.commit() + run_id = run.id + async with app_engine.begin() as conn: + await conn.execute(text("UPDATE runs SET options = '{}' WHERE id = :id"), {"id": run_id}) + async with SessionLocal() as session: + row = await session.get(Run, run_id) + assert row.options == {} and row.schedule_id is None + + +# -------------------------------------------------------------------------- +# 发起：落库、run.started、沿用 +# -------------------------------------------------------------------------- + + +async def test_options_and_schedule_are_stored_and_sealed(): + sid = _schedule() + run = await run_manager.start(graph=SIMPLE, input_payload={}, options=OPTIONS, schedule_id=sid) + assert (run.options, run.schedule_id) == (OPTIONS, sid) + row = await _settle(run.id) + assert row.status == "succeeded", row.error + assert (row.options, row.schedule_id) == (OPTIONS, sid) + [started] = await _started(run.id) + assert started["options"] == OPTIONS + assert set(started) == LEGACY_STARTED_KEYS | {"options"} + + +@pytest.mark.parametrize("options", [None, {}]) +async def test_without_options_run_started_is_unchanged(options): + run = await run_manager.start(graph=SIMPLE, input_payload={}, options=options) + row = await _settle(run.id) + assert row.options == {} and row.schedule_id is None + [started] = await _started(run.id) + assert set(started) == LEGACY_STARTED_KEYS, "选项为空时 run.started 多出了键" + + +async def test_the_options_belong_to_the_run_not_to_the_callers_dict(): + mine = {"verify": {"dual_check": True}} + run = await run_manager.start(graph=SIMPLE, input_payload={}, options=mine) + mine["verify"]["dual_check"] = False + mine["late"] = 1 + await _settle(run.id) + async with SessionLocal() as session: + assert (await session.get(Run, run.id)).options == {"verify": {"dual_check": True}} + + +@pytest.fixture +def seen_options(monkeypatch): + """记下每次执行「数据整形」节点时它从运行上下文里读到的选项。""" + from app.engine.runner import RUN_OPTIONS_KEY + + seen: list = [] + original = compiler.RUNNERS[NodeType.TRANSFORM] + + async def spy(state, ctx): + seen.append(ctx.run.extra.get(RUN_OPTIONS_KEY)) + return await original(state, ctx) + + monkeypatch.setitem(compiler.RUNNERS, NodeType.TRANSFORM, spy) + return seen + + +async def test_resume_keeps_the_options_the_run_started_with(seen_options): + run = await run_manager.start(graph=GATED, input_payload={}, options=OPTIONS) + assert (await _settle(run.id, ("interrupted",))).status == "interrupted" + await run_manager.resume(run.id, {"approved": True}) + row = await _settle(run.id) + assert row.status == "succeeded", row.error + assert seen_options == [OPTIONS, OPTIONS], "恢复之后节点读到的选项变了" + first, second = await _started(run.id) + assert first["options"] == second["options"] == OPTIONS and second["resumed"] is True + + +async def test_continue_keeps_the_options_too(seen_options, monkeypatch): + original = compiler.RUNNERS[NodeType.TRANSFORM] + calls = {"n": 0} + + async def flaky(state, ctx): + calls["n"] += 1 + if ctx.node.id == "after" and calls["n"] <= 2: + raise RuntimeError("临时故障") + return await original(state, ctx) + + monkeypatch.setitem(compiler.RUNNERS, NodeType.TRANSFORM, flaky) + graph = chain(node("start", "input"), + node("before", "transform", mode="template", template="前"), + node("after", "transform", mode="template", template="后"), + node("out", "output", fields=[{"name": "r", "value": "{{ nodes.after }}"}])) + run = await run_manager.start(graph=graph, input_payload={}, options=OPTIONS) + assert (await _settle(run.id)).status == "failed" + await run_manager.continue_failed(run.id) + row = await _settle(run.id) + assert row.status == "succeeded", row.error + assert [s.get("options") for s in await _started(run.id)] == [OPTIONS, OPTIONS] + + +async def test_nodes_read_an_empty_dict_when_there_are_no_options(seen_options): + run = await run_manager.start(graph=GATED, input_payload={}) + await _settle(run.id, ("interrupted",)) + assert seen_options == [{}] + + +# -------------------------------------------------------------------------- +# 接口：带出来源、按定时任务筛、launch_run 照转 +# -------------------------------------------------------------------------- + + +async def test_the_api_shows_and_filters_by_schedule(client): + sid, other = _schedule(), _schedule() + mine = [await run_manager.start(graph=SIMPLE, input_payload={}, schedule_id=sid) for _ in range(2)] + theirs = await run_manager.start(graph=SIMPLE, input_payload={}, schedule_id=other) + manual = await run_manager.start(graph=SIMPLE, input_payload={}) + for run in (*mine, theirs, manual): + await _settle(run.id) + + got = (await client.get(f"/api/runs/{mine[0].id}")).json() + assert got["schedule_id"] == sid + assert (await client.get(f"/api/runs/{manual.id}")).json()["schedule_id"] is None + + listed = (await client.get("/api/runs", params={"schedule_id": sid})).json() + assert sorted(r["id"] for r in listed) == sorted(r.id for r in mine) + assert all(r["schedule_id"] == sid for r in listed) + # 不带这个参数时照旧全列 + everything = {r["id"] for r in (await client.get("/api/runs", params={"limit": 200})).json()} + assert {theirs.id, manual.id} <= everything + + +async def test_launch_run_passes_both_through(): + from app.engine.launch import launch_run + + sid = _schedule() + async with SessionLocal() as session: + run = await launch_run(session, graph=SIMPLE, actor="定时任务", schedule_id=sid, options=OPTIONS) + row = await _settle(run.id) + assert (row.schedule_id, row.options, row.started_by) == (sid, OPTIONS, "定时任务") + [started] = await _started(run.id) + assert started["options"] == OPTIONS + + +async def test_an_http_launch_has_no_schedule_and_no_options(client): + res = await client.post("/api/runs", json={"graph": SIMPLE}) + assert res.status_code == 201, res.text + assert res.json()["schedule_id"] is None + row = await _settle(res.json()["id"]) + assert row.options == {}