fix(console): 草稿乐观锁是「读-比-写」非原子——并发保存双方都成功并静默丢一次编辑（SQLite 实测已可复现）

The race sits in _save_flow_once in plaita-console/backend/services/flow_store.py (around lines 405-422), where a separate SELECT via _stored_updated_at is followed by an ORM update keyed only on the primary key. Read _touch_updated_at and the optimistic-concurrency comment near line 58 first, then the existing sequential tests in tests/console/test_flow_store_concurrency.py. Done means that two saves with the same base_updated_at yield one success and one VersionConflictError, the definition UPDATE carries an updated_at predicate, and the existing sequential tests still pass.

优先级 P2 · 依赖：无

基线 HEAD = 65f48f2

Summary

草稿版本保存（PUT /api/flows/{flow_id}/versions/{version}）对外承诺「客户端回传 base_updated_at，服务端不一致即 409 version_stale」（flows.py:93、flows.py:313-322；前端 FlowEditor.tsx:341/395 正是这样回传的）。但服务端实现是 「读-比-写」三步非原子：先用一个独立 SELECT 读回当前 updated_at 比对，比对通过后再用只按主键定位的 ORM UPDATE 落库，两步之间没有行锁、没有条件更新、没有版本号 CAS。两个持同一基准的并发保存会双双通过冲突检查，后提交者静默覆盖先提交者，客户端拿到的是两个 200，没有任何一方收到 409。

1. 读了什么、比了什么、写了什么

冲突检查与写入在同一 session、同一事务里，但检查所依据的值来自一个独立的 SELECT（_stored_updated_at，flow_store.py:470-487），其后是 ORM 属性赋值 + flush() + 独立 _touch_updated_at：

plaita-console/backend/services/flow_store.py:405-422
                stored = self._stored_updated_at(session, flow_id, version, tenant_id)
                if (
                    base_updated_at is not None
                    and stored is not None
                    and stored != base_updated_at
                    and not force
                ):
                    raise VersionConflictError(
                        f"画布所基于的版本已被他人更新（{flow_id}@{version}），"
                        "请加载最新或选择强制覆盖",
                        latest_updated_at=stored,
                    )
                existing.definition = definition
                existing.layout = layout
                existing.status = status
                session.flush()
                self._touch_updated_at(session, flow_id, version, tenant_id)
                session.commit()

比对所用的 SELECT 本身不带任何锁定：

plaita-console/backend/services/flow_store.py:476-482
        sql = "SELECT updated_at FROM flow_versions WHERE flow_id = :fid AND version = :ver"
        params: Dict[str, Any] = {"fid": flow_id, "ver": version}
        if tenant_id is not None:
            sql += " AND tenant_id = :tid"
            params["tid"] = tenant_id
        try:
            row = session.execute(text(sql), params).first()

紧跟的 _touch_updated_at（flow_store.py:456-468）刷新的是同一行的 updated_at，它的 WHERE 也只按 tenant_id/flow_id/version 定位，不带「旧值等于 base」的前置条件：

plaita-console/backend/services/flow_store.py:462-468
        session.execute(
            text(
                "UPDATE flow_versions SET updated_at = :ts "
                "WHERE tenant_id = :tid AND flow_id = :fid AND version = :ver"
            ),
            {"ts": datetime.utcnow(), "tid": tenant_id, "fid": flow_id, "ver": version},
        )

模块头注释把这一机制明确写成「乐观并发」（flow_store.py:58-60）：

plaita-console/backend/services/flow_store.py:58-60
# ---- C5-1 乐观并发：flow_versions.updated_at ----
# models/flow.py 的 FlowVersion 未声明 updated_at（历史遗留：草稿更新时间无处
# 可查，保存是 last-write-wins）。本服务层把列追加到映射表的 Table 元数据上：

2. 反证：全服务层零行锁、零条件更新

$ grep -rn "with_for_update\|FOR UPDATE" --include='*.py' plaita-console/backend/
（零命中，退出码 1）

比对通过的判定条件里没有任何「写入时再校验一次」的谓词：flow_store.py 内所有含 updated_at 的 UPDATE/WHERE 命中共 7 处，逐一核对后只有 :464 那条无条件 _touch_updated_at；existing.definition = ... 走的是 SQLAlchemy ORM 按主键的 UPDATE，没有任何 updated_at 参与 WHERE。

3. API 层确为同步 def（可并发进入）

plaita-console/backend/api/flows.py:290-292
@router.put("/flows/{flow_id}/versions/{version}", response_model=VersionView)
def save_version(flow_id: str, version: str, req: SaveVersionRequest,
                 request: Request = None):  # noqa: ANN001 — FastAPI 注入

同步 def 由 FastAPI 投到线程池执行，同一进程内多个请求体可同时走到 _save_flow_once。协商契约与 409 映射见 flows.py:301-322（store.save_flow_definition(...) → except flow_store.VersionConflictError → "conflict": "version_stale"）。

4. 实跑复现（SQLAlchemy 事件钩子 + 双线程，SQLite 后端）

$ /tmp/plaita-audit-venv/bin/python /tmp/verif/repro_flow_lock.py
== 一次成功保存实际执行的语句 ==
   SELECT flows.id, flows.tenant_id, flows.flow_id, flows.author, flows."desc", flows.created_at, flows.updated_at FROM flows WHERE flows.tenant_id = ? AND flows.flow_id = ?
   SELECT flow_versions.id, flow_versions.tenant_id, flow_versions.flow_id, flow_versions.version, flow_versions.status, flow_versions.definition, flow_versions.layout, flow_versions.
   SELECT updated_at FROM flow_versions WHERE flow_id = ? AND version = ? AND tenant_id = ?
   UPDATE flow_versions SET definition=? WHERE flow_versions.id = ?
   UPDATE flow_versions SET updated_at = ? WHERE tenant_id = ? AND flow_id = ? AND version = ?
→ UPDATE 语句内含 updated_at 条件: True
→ SELECT 带 FOR UPDATE: False
== 双方持同一 base 并发保存 ==
  结果: {'threadA': 'OK', 'threadB': 'OK'}
  最终 desc: threadB

（脚本末行 UPDATE 语句内含 updated_at 条件: True 是它把紧随其后的 _touch_updated_at 也一并匹配了，与本条无关——真正落 definition 的那条是 UPDATE flow_versions SET definition=? WHERE flow_versions.id = ?，WHERE 只有主键，没有 updated_at 谓词。）

脚本第 41-50 行给 _stored_updated_at 装了 threading.Barrier(2)，只是把「双方都先读到旧基准」这一必然存在的窗口稳定复现出来，没有人为制造竞态：真实并发下两请求同样会在各自读到旧 updated_at 之后才写库。

5. 既有测试全是顺序用例，不覆盖竞争

plaita-console/backend/tests/console/test_flow_store_concurrency.py:65-78
def test_stale_base_conflicts(store: FlowStore):
    # 「标签页 A」写入后，「标签页 B」仍持旧基准保存 → 冲突
    store.save_flow_definition("echo", "0.0.1", _echo_definition(tag="a"))
    stale_base = store.get_version("echo", "0.0.1").updated_at.isoformat()
    store.save_flow_definition("echo", "0.0.1", _echo_definition(tag="b"))  # A 的更新

    with pytest.raises(VersionConflictError) as ei:
        store.save_flow_definition(
            "echo", "0.0.1", _echo_definition(tag="c"), base_updated_at=stale_base
        )

test_flow_store_concurrency.py 全文件（:52-256）的冲突用例都在 A 的写入完成并提交之后才发起 B 的保存，即只验证「顺序过期 → 409」，没有任何并发竞争断言。

影响

• 多标签页 / 多人同编一条流程（console 默认部署形态）：两个人（或同一人两个标签页）在各自的编辑窗口里保存同一 flow_id@version，都会走「读旧基准 → 比对相同 → 各自 UPDATE」的路径，双方都收到 200，前端不弹冲突提示。谁最后提交谁的定义生效，另一人的整份画布编辑静默丢失且无从追溯——用户以为保存成功，实际已被覆盖。乐观锁存在的唯一目的（把「静默覆盖」变成「显式 409」）在此场景下完全失效。
• 代价不对称：这是用户手写的流程定义，丢失的是一整个版本的 definition + layout，没有历史副本（同版本被原地覆盖，不是新增版本），恢复只能靠用户记忆或外部备份。相比一次明确的 409（用户点「加载最新」或「强制覆盖」）成本高得多。
• PG 部署（仓内官方迁移脚本 plaita-console/scripts/migrate_sqlite_to_pg.py 可达，--pg 直接指向 PG 库）：READ COMMITTED 下两事务互不阻塞，两方 UPDATE 都成功，覆盖照旧发生——这是本条最典型的暴露面。
• SQLite 部署同样可复现：实测双方均返回成功（本仓 README 默认的单机形态），并非「SQLite 会 database is locked 所以侥幸挡住」。
• 不修的后果：并发保存恒为 last-write-wins，409 门只在「顺序晚到」时触发，用户对「保存成功」的信任被错误地维持。

建议

把「比对」与「写入」合并为一次原子条件写，二选一（任一即可闭合本条）：

1. 条件 UPDATE（推荐，改动最小）：在 _save_flow_once（flow_store.py:405-422）里放弃「先 SELECT 比对、再 ORM 赋值」的写法，改为单条
   UPDATE flow_versions SET definition=:d, layout=:l, status=:s, updated_at=:ts WHERE tenant_id=:tid AND flow_id=:fid AND version=:ver AND updated_at = :base_updated_at，
   以 rowcount == 0 判定冲突（再 SELECT 一次取 latest_updated_at 填进 VersionConflictError）；force=True 时去掉 updated_at = :base 谓词。注意 _updated_at_supported() 为假（旧库缺列）时该守卫退化为无条件更新——此时应至少 logger.warning 一次，让「无并发保护」可见。
2. 行锁：在比对前对目标行加锁，PG 用 SELECT ... FOR UPDATE；SQLite 不支持该语法，只能靠 BEGIN IMMEDIATE 把整库写事务提前串行化，代价是拖慢所有并发写，故方案 1 对两种后端都更合适。全仓当前零 with_for_update（见 Summary §2）。

验收条件：

• 新增单测（并入 tests/console/test_flow_store_concurrency.py）：两线程/两 session 持同一 base_updated_at 并发调 save_flow_definition("echo","0.0.1",...)，断言恰好一个成功、另一个抛 VersionConflictError，且最终 definition 等于成功那一方的内容；当前复现脚本 /tmp/verif/repro_flow_lock.py 的结果应从 {'threadA': 'OK', 'threadB': 'OK'} 变为一 OK 一 VersionConflictError。
• 断言实际 SQL：对落 definition 的 UPDATE，WHERE 子句必须包含 updated_at 谓词（可用 SQLAlchemy 事件钩子或 rowcount 语义间接验证）；即 Summary §4 里 UPDATE flow_versions SET definition=? WHERE flow_versions.id = ? 这条语句不得再出现。
• 保持既有顺序用例全绿（test_stale_base_conflicts / test_matching_base_saves / test_force_overrides_conflict / test_no_base_keeps_legacy_behavior，:65-106），即不带 base_updated_at 的存量调用方行为不变。

边界

• 本条只修 flow_store._save_flow_once 的「读-比-写」原子性，以及它导致的双写静默覆盖。不包含：_updated_at_supported() 对旧库缺列的整体降级策略是否合理（属能力探测设计，另议）；allocate_version 的并发插入重试路径（flow_store.py:385-392，走 IntegrityError 重试，是另一套机制且当前工作正常）；Api.flows 的鉴权/租户维度；前端 FlowEditor.tsx 的本地基准缓存策略。
• 与 #40（list_executions 丢 tenant_id/flow_hash、redis 后端列表无索引）无交集：那是执行记录查询的字段与性能问题，本条是版本存储写入的并发原子性问题，代码位置与后果都不同。
• 与 #36（flow_hash 基于解析后 model_dump 导致跨版本漂移）无交集：本条不涉及 flow_hash 计算，也不涉及 worker 侧执行语义。
• 本条在单进程内即可触发（FastAPI 同步端点 + 线程池 + 共享 SQLAlchemy 引擎），无需多机部署；SQLite 与 PG 均可复现，PG 的 READ COMMITTED 是代价最高、最容易被真实用户踩到的形态。