From c8363206bd98879335db53c04a83ff794fde8f51 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Mon, 5 Oct 2026 01:18:43 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8F=91=E8=B5=B7=E6=97=B6=E5=B0=B1=E5=AE=9A?= =?UTF-8?q?=E4=B8=8B=E7=9A=84=E4=BA=8B=E4=BB=B6=E5=92=8C=E8=BF=90=E8=A1=8C?= =?UTF-8?q?=E8=AE=B0=E5=BD=95=E5=90=8C=E4=B8=80=E4=B8=AA=E4=BA=8B=E5=8A=A1?= =?UTF-8?q?=E5=86=99=E8=BF=9B=E4=BA=8B=E4=BB=B6=E6=B5=81=EF=BC=9A=E5=8F=97?= =?UTF-8?q?=E7=AE=A1=E7=BA=A7=E5=88=AB=E9=82=A3=E6=9D=A1=E6=97=A5=E5=BF=97?= =?UTF-8?q?=E3=80=81=E7=9B=AE=E5=BD=95=E5=8F=98=E5=8C=96=E3=80=81=E5=8F=A3?= =?UTF-8?q?=E5=BE=84=E5=8D=87=E7=89=88=E5=A4=84=E7=BD=AE=E4=BB=A5=E5=89=8D?= =?UTF-8?q?=E5=9C=A8=20run=5Fmanager.start=20=E8=BF=94=E5=9B=9E=E4=B9=8B?= =?UTF-8?q?=E5=90=8E=E5=8F=A6=E5=A4=96=E5=86=99=EF=BC=8C=E8=BF=99=E6=97=B6?= =?UTF-8?q?=E8=BF=90=E8=A1=8C=E5=B7=B2=E7=BB=8F=E5=BB=BA=E5=A5=BD=E3=80=81?= =?UTF-8?q?=E6=89=A7=E8=A1=8C=E4=BB=BB=E5=8A=A1=E5=B7=B2=E7=BB=8F=E8=B5=B7?= =?UTF-8?q?=E4=BA=86=E3=80=82=E5=85=B6=E4=B8=AD=E4=BB=BB=E4=BD=95=E4=B8=80?= =?UTF-8?q?=E6=AC=A1=E5=86=99=E5=A4=B1=E8=B4=A5=EF=BC=88=E5=BA=93=E8=A2=AB?= =?UTF-8?q?=E9=94=81=E8=B6=85=E8=BF=87=E7=AD=89=E5=BE=85=E6=97=B6=E9=95=BF?= =?UTF-8?q?=E4=B9=8B=E7=B1=BB=EF=BC=89=EF=BC=8Claunch=5Frun=20=E6=8A=9B?= =?UTF-8?q?=E5=87=BA=E6=99=AE=E9=80=9A=E5=BC=82=E5=B8=B8=EF=BC=8C=E8=BF=90?= =?UTF-8?q?=E8=A1=8C=E5=8D=B4=E5=9C=A8=E6=89=A7=E8=A1=8C=EF=BC=9B=E5=AE=9A?= =?UTF-8?q?=E6=97=B6=E4=BB=BB=E5=8A=A1=E6=8C=89=E3=80=8CLaunchRefused=20?= =?UTF-8?q?=E4=BB=A5=E5=A4=96=E7=9A=84=E5=BC=82=E5=B8=B8=E8=AE=B0=E6=9C=AA?= =?UTF-8?q?=E8=83=BD=E5=8F=91=E8=B5=B7=E3=80=8D=E4=BC=9A=E6=8A=8A=E4=B8=80?= =?UTF-8?q?=E6=AC=A1=E6=AD=A3=E5=9C=A8=E6=89=A7=E8=A1=8C=E7=9A=84=E6=AD=A3?= =?UTF-8?q?=E5=BC=8F=E8=BF=90=E8=A1=8C=E8=AE=B0=E6=88=90=E6=9C=AA=E8=83=BD?= =?UTF-8?q?=E5=8F=91=E8=B5=B7=EF=BC=8C=E5=B0=81=E5=AD=98=E8=8C=83=E5=9B=B4?= =?UTF-8?q?=E9=87=8C=E4=B9=9F=E5=B0=91=E4=BA=86=E5=8F=97=E7=AE=A1=E7=BA=A7?= =?UTF-8?q?=E5=88=AB=E9=82=A3=E6=9D=A1=E6=97=A5=E5=BF=97=EF=BC=88governed?= =?UTF-8?q?=5Fformal=20=E6=89=BE=E4=B8=8D=E5=88=B0=E5=AE=83=E5=B0=B1?= =?UTF-8?q?=E9=80=80=E5=9B=9E=E7=9C=8B=E5=B7=A5=E4=BD=9C=E6=B5=81=E7=8E=B0?= =?UTF-8?q?=E5=9C=A8=E7=9A=84=20status=EF=BC=8C=E4=B8=AD=E9=80=94=E6=94=B9?= =?UTF-8?q?=E8=BF=87=E7=94=BB=E5=B8=83=E7=9A=84=E4=BC=9A=E6=8C=89=E5=B7=B2?= =?UTF-8?q?=E5=8F=91=E5=B8=83=E7=BA=A7=E5=88=AB=E5=87=BA=E5=85=B7=EF=BC=89?= =?UTF-8?q?=E3=80=82=E5=8F=A6=E5=A4=96=E8=BF=99=E5=87=A0=E6=9D=A1=E5=92=8C?= =?UTF-8?q?=E6=89=A7=E8=A1=8C=E6=8A=A2=E5=85=88=E5=90=8E=EF=BC=8C=E8=BF=90?= =?UTF-8?q?=E8=A1=8C=E7=BB=93=E6=9D=9F=E5=BE=97=E5=BF=AB=E6=97=B6=E5=8F=AF?= =?UTF-8?q?=E8=83=BD=E8=90=BD=E5=88=B0=E5=B0=81=E5=AD=98=E4=B9=8B=E5=90=8E?= =?UTF-8?q?=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - run_manager.start 加 notes：(类型, node_id, 数据) 的列表，先 flush 拿到运行的 id，和运行记录同一个事务提交，提交之后广播，最后才建执行任务。写不进去就不建运行，launch_run 抛出的任何异常都说明运行没有开始执行 - launch_run 在发起前把这几条事件备好交给 start，不再事后用 run_manager.note 写。顺序和以前实际落库的一样（日志、目录变化、口径升版处置，然后 run.started），现在是确定的 - 没有 notes 的发起（探索运行、问数据）和以前完全相同 没有照复核建议的「写失败就记日志、照常返回运行」：那样正式运行会在缺了受管级别记录的情况下照常出具，可能悄悄降成已发布级别。 测试：用触发器让口径升版处置那条写不进去，launch_run 抛出 IntegrityError、库里没有多出运行（旧代码留下一条正在执行的运行，这条变红）；服务发起的正式运行里，受管级别日志和口径升版处置排在 run.started 前面，都在封存范围里 --- backend/app/engine/launch.py | 41 ++++++++++++++++++-------------- backend/app/engine/runner.py | 22 ++++++++++++++--- backend/tests/test_run_launch.py | 38 ++++++++++++++++++++++++++--- 3 files changed, 77 insertions(+), 24 deletions(-) diff --git a/backend/app/engine/launch.py b/backend/app/engine/launch.py index 71d7a9aa..087be199 100644 --- a/backend/app/engine/launch.py +++ b/backend/app/engine/launch.py @@ -3,7 +3,8 @@ 以前这些规则全写在 start_run 里、只认 HTTP 请求：定时任务要发起正式运行，要么另抄一份（两份规则迟早 走样，正式运行「只能从当前发布版本发起」这条最要紧的规则就可能在一边被绕过），要么假装成一次 HTTP 请求。 现在两边都调 launch_run。拒绝时抛 LaunchRefused：start_run 把它换回以前的 HTTP 响应（状态码、detail、 -code 逐字不变），调度器据此记一次未能发起。 +code 逐字不变），调度器据此记一次未能发起。别的异常（库出错之类）也说明运行没有开始执行：发起时定下的事件 +（受管级别、目录变化、口径升版处置）和运行记录同一个事务提交，执行任务是最后一步才建的。 发起前检查绑定的工具（missing_tools），以及它和 runner 固定数据版本共用的取绑定的办法（tool_bindings、 node_setting）也放在这里：runner 以前要从接口层导入它们。 @@ -55,7 +56,8 @@ async def launch_run( schedule_id: str | None = None, options: dict[str, Any] | None = None, ) -> Run: - """发起一次运行，返回刚建好的运行记录（执行已经交给 run_manager）。不发起就抛 LaunchRefused。 + """发起一次运行，返回刚建好的运行记录（执行已经交给 run_manager）。不发起就抛 LaunchRefused；抛别的异常时 + 运行也没有开始执行。 参数和 POST /api/runs 的请求体一一对应（input 叫 input_payload）；actor 是记在运行上的发起人。 schedule_id 是发起它的定时任务，options 是整次运行沿用的选项快照，都原样交给 run_manager.start。 @@ -137,6 +139,24 @@ async def launch_run( from app.api.settings import resolve_run_scope memory_scope, collection = await resolve_run_scope(session, memory_scope, collection) + # 发起时就定下的事件交给 run_manager.start，和运行记录同一个事务写进事件流（在封存范围内、排在 run.started + # 前面）。以前运行建好、开始执行之后才另外写：写失败时运行已经在执行，这里却抛出异常，定时任务会把一次 + # 正在执行的正式运行记成未能发起 + notes: list[tuple[EventType | str, str | None, dict[str, Any]]] = [] + if formal: + # 按不按受管出具在发起这一刻定下来：status 反映的是当前画布，中途改图、审批停下再恢复时再去读它， + # 报告节点和出口复核可能各读到不同的值 + notes.append(("log", None, governance_note(governed=governed, workflow_name=name, version=run_version))) + if catalog_changes: + # 之后回看这次正式出具，能知道当时的目录和发布时不一样、差在哪几张表。每张表带 impact：direct 写死的 + # SQL 用到的，possible Agent 可能查询的（界面分开说） + notes.append((EventType.CATALOG_DRIFT, None, + {"version": run_version, "count": len(catalog_changes), "tables": catalog_changes})) + # 已声明策略的升版记入事件流，出具物上能看到"这期换口径了、怎么处置的" + for ev in upgrade_events: + payload_ev = dict(ev) + node = payload_ev.pop("node_id", None) + notes.append(("caliber.upgrade", node, payload_ev)) try: run = await run_manager.start( graph=graph, @@ -151,25 +171,10 @@ async def launch_run( started_by=actor, options=options, schedule_id=schedule_id, + notes=notes, ) except ValueError as e: raise LaunchRefused(400, str(e)) from e - - if formal: - # 按不按受管出具在发起这一刻定下来、记进封存范围：status 反映的是当前画布，中途改图、 - # 审批停下再恢复时再去读它，报告节点和出口复核可能各读到不同的值 - await run_manager.note(run.id, "log", **governance_note( - governed=governed, workflow_name=name, version=run_version)) - if catalog_changes: - # 落在封存范围内：之后回看这次正式出具，能知道当时的目录和发布时不一样、差在哪几张表。 - # 每张表带 impact：direct 写死的 SQL 用到的，possible Agent 可能查询的（界面分开说） - await run_manager.note(run.id, EventType.CATALOG_DRIFT, version=run_version, count=len(catalog_changes), - tables=catalog_changes) - # 已声明策略的升版记入事件流，出具物上能看到"这期换口径了、怎么处置的" - for ev in upgrade_events: - payload_ev = dict(ev) - node = payload_ev.pop("node_id", None) - await run_manager.note(run.id, "caliber.upgrade", node_id=node, **payload_ev) return run diff --git a/backend/app/engine/runner.py b/backend/app/engine/runner.py index a1dd51f8..4677078b 100644 --- a/backend/app/engine/runner.py +++ b/backend/app/engine/runner.py @@ -6,7 +6,7 @@ import copy import json import logging import time -from collections.abc import Callable +from collections.abc import Callable, Sequence from datetime import datetime, timezone from typing import Any @@ -172,7 +172,7 @@ class RunManager: self, run_id: str, event_type: EventType | str, *, node_id: str | None = None, **data: Any, ) -> None: - """给 API 层用的事件入口（比如 formal 启动时记录口径升版处置）。""" + """给运行之外的代码往运行里记一条事件。发起时就定下的事件不走这里，交给 start 的 notes。""" await self._emit(run_id, event_type, node_id=node_id, data=data) # ---------------- 启动 ---------------- @@ -193,8 +193,14 @@ class RunManager: approval_default: str | None = None, options: dict[str, Any] | None = None, schedule_id: str | None = None, + notes: Sequence[tuple[EventType | str, str | None, dict[str, Any]]] = (), ) -> Run: - """建运行记录、交给执行任务。options 是这次运行的选项快照，schedule_id 是发起它的定时任务（可空）。""" + """建运行记录、交给执行任务。options 是这次运行的选项快照，schedule_id 是发起它的定时任务（可空）。 + + notes 是发起时就定下的事件 (类型, node_id, 数据)，比如正式运行按什么级别出具、口径升版怎么处置。它们和 + 运行记录同一个事务提交，排在 run.started 前面：写不进去就不建运行，不会出现运行已经在执行、调用方却 + 只拿到一个异常（定时任务会把它记成未能发起）；也不会和执行抢先后，运行结束得快时落到封存之后。 + """ spec = GraphSpec.model_validate(graph) report = validate_graph(spec) if not report.ok: @@ -238,8 +244,18 @@ class RunManager: ) run.thread_id = run.id # 一个 run 一条 checkpoint 线程 session.add(run) + staged: list[RunEventModel] = [] + if notes: + await session.flush() # 事件要带上运行的 id + staged = [self._event(run.id, kind, node_id=node_id, data=data) + for kind, node_id, data in notes] + for event in staged: + await _stage(session, event, run) await session.commit() await session.refresh(run) + # 先落库再广播，理由同 _emit；执行任务还没建，run.started 必定排在它们后面 + for event in staged: + await bus.publish(event) self._tasks[run.id] = asyncio.create_task( self._drive( diff --git a/backend/tests/test_run_launch.py b/backend/tests/test_run_launch.py index 4fb698a1..edd38cf0 100644 --- a/backend/tests/test_run_launch.py +++ b/backend/tests/test_run_launch.py @@ -2,7 +2,8 @@ - 每一种拒绝：POST /api/runs 的状态码、detail、code 先按搬家之前的原话写死（这一组先在旧代码上通过）； 直接调 launch_run 抛 LaunchRefused，和 HTTP 的返回逐条一致 -- 发起之后的治理事件（受管级别那条日志、口径升版处置）在服务里写：不经 HTTP 发起的正式运行一样有 +- 发起时定下的治理事件（受管级别那条日志、口径升版处置）在服务里写：不经 HTTP 发起的正式运行一样有； + 它们和运行记录同一个事务提交，排在 run.started 前面、在封存范围里，写不进去就不建运行 - 旧的导入路径（app.api.runs 的 missing_tools、node_setting、tool_bindings、TOOL_MISSING）照旧可用； runner 不再从接口层导入 """ @@ -299,10 +300,41 @@ async def test_a_formal_launch_records_the_governance_notes(): assert run.version_hash and run.input == {"week": "第 40 周"} row = await _settle(run.id) assert row.status == "succeeded", row.error - notes = [e.data for e in await _events(run.id, "log") if e.data.get("code") == "run_governance"] - assert len(notes) == 1 and notes[0]["governed"] is False and "v1" in notes[0]["message"] + notes = [e for e in await _events(run.id, "log") if e.data.get("code") == "run_governance"] + assert len(notes) == 1 and notes[0].data["governed"] is False and "v1" in notes[0].data["message"] [upgrade] = await _events(run.id, "caliber.upgrade") assert upgrade.node_id == "card" and upgrade.data["policy"] == "dual" and upgrade.data["latest"] == 2 + # 发起时就有的事件排在 run.started 前面，都在封存范围里 + [started] = await _events(run.id, "run.started") + assert notes[0].seq < upgrade.seq < started.seq <= row.manifest_seq + + +async def test_launch_events_are_written_with_the_run_or_not_at_all(): + """发起时定下的事件和运行记录同一个事务提交：写不进去就不建运行，launch_run 抛出异常就是没有发起。以前运行 + 建好、开始执行之后才另外写，写失败时运行已经在执行，调用方（定时任务）只拿到一个异常，会把一次正在执行的 + 正式运行记成未能发起，封存范围里也少了受管级别那条日志。""" + from sqlalchemy import text + from sqlalchemy.exc import IntegrityError + + from app.db.base import engine + from app.engine.launch import launch_run + + lib = await upstream_with_two_versions() + wf, _ = await workflow(weekly(lib, upgrade_policy="dual"), published=1, versions=(1,)) + before = await _run_count() + # 口径升版处置那条写不进去。走异步引擎建删触发器：同步连接等写锁时会卡住事件循环，而持锁的正是这个循环里的写入 + async with engine.begin() as conn: + await conn.execute(text( + "CREATE TRIGGER refuse_upgrade_note BEFORE INSERT ON run_events WHEN NEW.type = 'caliber.upgrade' " + "BEGIN SELECT RAISE(ABORT, 'refused by test'); END")) + try: + async with SessionLocal() as session: + with pytest.raises(IntegrityError): + await launch_run(session, workflow_id=wf, run_class="formal", actor="定时任务") + assert await _run_count() == before, "没写进治理事件，却留下了一条运行" + finally: + async with engine.begin() as conn: + await conn.execute(text("DROP TRIGGER IF EXISTS refuse_upgrade_note")) async def test_an_exploratory_launch_takes_the_saved_graph_and_the_run_defaults():