发起时就定下的事件和运行记录同一个事务写进事件流：受管级别那条日志、目录变化、口径升版处置以前在 run_manager.start 返回之后另外写，这时运行已经建好、执行任务已经起了。其中任何一次写失败（库被锁超过等待时长之类），launch_run 抛出普通异常，运行却在执行；定时任务按「LaunchRefused 以外的异常记未能发起」会把一次正在执行的正式运行记成未能发起，封存范围里也少了受管级别那条日志（governed_formal 找不到它就退回看工作流现在的 status，中途改过画布的会按已发布级别出具）。另外这几条和执行抢先后，运行结束得快时可能落到封存之后。

- 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

@@ -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
 
 

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(

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