运行记下选项快照和来源：runs 加 options（JSON，默认 {}）和 schedule_id（可空，建索引）两列。后面的核查设置要在发起时定下、整次运行沿用，审批停下期间改了设置，恢复和重放仍按发起时的（同 approval_default 的道理）；定时任务要能找回自己发起的运行。RunManager.start 收 options、schedule_id 并落库，选项先深拷一份，调用方之后改自己的 dict 不影响快照；选项非空时写进每一段的 run.started（进封存范围），为空时 run.started 和以前逐字相同；恢复、继续运行照运行记录沿用，节点从 RunContext.extra 的 run_options 读同一份拷贝。RunOut 带 schedule_id，GET /api/runs 可按 schedule_id 筛，launch_run 原样往下交。旧库走 _COLUMN_MIGRATIONS 补列，存量运行回填 {} 和 NULL；create_all 不碰已有的表，新列的索引另由 _INDEX_MIGRATIONS 补建，索引名和新库 create_all 建出来的相同

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())

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:

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"

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

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 {},
     }
 
 

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 == {}