变化记录的 affects 先占住那一行再读改写（复核 L5）：口径试算记没试算的那几版（ledger/recall._probe_skipped）和探查刷新影响面（data/source_changes.record）以前都是先读再整份写，交错时互相盖掉。现在两处都先发一句按条件的 UPDATE（record 顺手记又见到一次，SQLite 上拿到写锁、PostgreSQL 上锁住这一行），再读出来改。补并发测试（六版口径同时记、探查同时刷新，一条不丢）

backend/app/data/source_changes.py

@@ -29,7 +29,7 @@ from dataclasses import dataclass
 from datetime import datetime
 from typing import Any
 
-from sqlalchemy import func, or_, select
+from sqlalchemy import func, or_, select, update
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -150,11 +150,16 @@ async def record(session: AsyncSession, source: Any, change: NewChange, detected
         raise ValueError(f"不认识的严重程度：{change.severity}")
     fp = fingerprint(change.kind, change.table, change.column, change.key)
     now = utcnow()
+    # 已经有了的：先按条件 UPDATE 记下又见到一次（同时占住那一行：SQLite 上拿到写锁、PostgreSQL 上锁住这一行），再读出来改
+    # 影响面——口径试算同时往 affects 里记没试算的那几版（ledger/recall._probe_skipped，同样先占住再读改写），谁也不盖掉谁
+    seen = await session.execute(update(SourceChange).where(
+        SourceChange.source_id == source.id, SourceChange.fingerprint == fp)
+        .values(last_seen_at=now, seen_count=func.coalesce(SourceChange.seen_count, 0) + 1)
+        .execution_options(synchronize_session=False))
     found = (await session.execute(select(SourceChange).where(
-        SourceChange.source_id == source.id, SourceChange.fingerprint == fp))).scalar_one_or_none()
+        SourceChange.source_id == source.id, SourceChange.fingerprint == fp)
+        .execution_options(populate_existing=True))).scalar_one_or_none() if seen.rowcount else None
     if found is not None:
-        found.last_seen_at = now
-        found.seen_count = int(found.seen_count or 0) + 1
         if found.status == "open":
             # 影响按这一次刷新；表结构变了之后口径试算的情况（ledger/recall 记的 probes）不是探查算的，留着
             probes = (found.affects or {}).get("probes") if isinstance(found.affects, dict) else None

backend/app/ledger/recall.py

@@ -66,7 +66,7 @@ from dataclasses import dataclass, field
 from datetime import datetime, timedelta, timezone
 from typing import Any
 
-from sqlalchemy import func, select
+from sqlalchemy import func, select, update
 
 from app.core.events import EventType
 from app.db.base import SessionLocal, utcnow
@@ -1585,8 +1585,14 @@ async def _probe_skipped(change: Mapping[str, Any], caliber: Any, pinned: Any, g
     note = {"workflow_id": caliber.workflow_id, "version": caliber.version, "name": str(name), "outcome": "skipped",
             "missing": missing, "reason": reason, "inputs_from": inputs_from}
     async with SessionLocal() as session:
-        row = (await session.execute(select(SourceChange).where(SourceChange.id == change.get("id")))).scalar_one_or_none()
+        # 先占住那一行再读改写（SQLite 上拿到写锁、PostgreSQL 上锁住这一行）：几版口径同时记、探查同时刷新影响
+        # （source_changes.record）时一个接一个来，谁也不盖掉谁
+        await session.execute(update(SourceChange).where(SourceChange.id == change.get("id"))
+                              .values(seen_count=SourceChange.seen_count).execution_options(synchronize_session=False))
+        row = (await session.execute(select(SourceChange).where(SourceChange.id == change.get("id"))
+                                     .execution_options(populate_existing=True))).scalar_one_or_none()
         if row is None:
+            await session.rollback()
             return
         affects = dict(row.affects or {})
         kept = [p for p in affects.get(PROBES_KEY) or [] if isinstance(p, dict)

backend/tests/test_ledger_recall.py

@@ -954,3 +954,40 @@ async def test_a_probe_that_cannot_assemble_its_inputs_says_which_are_missing(en
         await session.commit()
         again = (await session.execute(select(SourceChange).where(SourceChange.id == out["id"]))).scalar_one()
     assert again.affects["probes"] == [note]
+
+
+async def test_probe_notes_and_change_refreshes_do_not_overwrite_each_other(engine_up):
+    """同一条变化上几处同时写 affects（几版口径各记一条没试算、探查又刷新了一次影响）：先占住那一行再读改写，谁也不盖掉谁。"""
+    import asyncio
+    from types import SimpleNamespace
+
+    from app.data import source_changes as sc
+    from app.db.models import DataSource
+    from app.db.models_sources import SourceChange
+
+    async with SessionLocal() as session:
+        source = (await session.execute(select(DataSource).where(DataSource.name == S.CURRENT["source"]))).scalar_one()
+        change, _ = await sc.record(session, source, sc.NewChange(kind="column_removed", table="orders", column="amount",
+                                                                  severity="action", key="v4",
+                                                                  affects={"catalog": [], "templates": [],
+                                                                           "rules_touched": False}),
+                                    sc.Detected(job_kind="introspect"))
+        await session.commit()
+        out = sc.change_out(change)
+        source_id = source.id
+
+    async def refresh() -> None:
+        async with SessionLocal() as session:
+            src = await session.get(DataSource, source_id)
+            await sc.record(session, src, sc.NewChange(kind="column_removed", table="orders", column="amount",
+                                                       severity="action", key="v4",
+                                                       affects={"catalog": [], "templates": [], "rules_touched": True}),
+                            sc.Detected(job_kind="introspect"))
+            await session.commit()
+
+    calibers = [SimpleNamespace(workflow_id=f"wf-{i}", version=1) for i in range(6)]
+    await asyncio.gather(*(recall._probe_skipped(out, c, None, {}, "defaults", "缺参数") for c in calibers), refresh())
+    async with SessionLocal() as session:
+        row = (await session.execute(select(SourceChange).where(SourceChange.id == out["id"]))).scalar_one()
+    assert sorted(p["workflow_id"] for p in row.affects["probes"]) == [f"wf-{i}" for i in range(6)], row.affects
+    assert row.affects["rules_touched"] is True, "探查刷新的那一份也在"