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