diff --git a/backend/app/data/catalog_profile_observe.py b/backend/app/data/catalog_profile_observe.py
index ac5d4968..044355e1 100644
--- a/backend/app/data/catalog_profile_observe.py
+++ b/backend/app/data/catalog_profile_observe.py
@@ -23,6 +23,11 @@
 列）留最近 26 条；拿同口径的历史比出取值变化写进 source_changes。机器唯一能写进目录的事：数据证明码值表不再列全时撤销
 「已列全」（目录版本 +1，现有的目录漂移、冻结快照、SQL 检查自然跟上）；新取值本身、含义、状态都不改，交给人在变化里处理。
 
+**观察出错不挡剖析**：每条观察（连同它比出的变化、它撤销的已列全）、每条覆盖率下降各占一个保存点（SAVEPOINT）。一条写库
+出错（flush 报错之类）只回滚它自己的保存点，这张表的剖析结果、写进目录的结论和别的观察照常提交；只靠 try/except 不够：
+没有保存点时 flush 一出错 SQLAlchemy 就把整个事务回滚了，外层提交这张表的结果时跟着抛错。撤销已列全按库里的目录版本写
+（见 _revoke_complete）：catalog.write_entry 乐观锁对不上时回滚的是整个会话，保存点挡不住。
+
 **什么时候观察**：剖析任务的范围 observe（只观察）和 maintenance（需要剖析的表 + 观察到期的表，定时维护用）才带观察检查；
 人点的「需要剖析 / 全部 / 指定的表」照旧不观察。观察到期（observe_due）：有数据目录、不是视图的表，从没观察过，或者上一次
 观察在 OBSERVE_EVERY_H 小时以前。
@@ -39,6 +44,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.data import catalog
 from app.db.base import utcnow
+from app.db.models import CatalogNote
 from app.db.models_sources import TableObservation
 
 logger = logging.getLogger(__name__)
@@ -311,76 +317,99 @@ def _rules_columns(notes: Mapping[str, Any], meta: Mapping[str, Any] | None, tab
 
 
 async def _revoke_complete(session: AsyncSession, source_id: str, table: str, column: str) -> int | None:
-    """撤销码值项上的「已列全」（目录版本 +1，不提交）：值、含义、状态、来源都不动。撞上别人刚改过重读重来，最多三次。
-    已经不是已列全了（别人刚改过）返回 None。"""
-    for _attempt in range(3):
-        entry = await catalog.read_entry(session, source_id, table)
-        if entry is None:
-            return None
-        notes = entry.notes
-        item = ((notes.get("columns") or {}).get(column) or {}).get("codes")
-        if not catalog.codes_complete(item):
-            return None
-        item.pop("complete", None)
-        item["updated_at"] = catalog.now_iso()
-        try:
-            written = await catalog.write_entry(session, source_id, table, notes, if_version=entry.version,
-                                                actor=OBSERVER, commit=False)
-        except catalog.CatalogConflict:
-            continue
-        return written.version
-    return None
+    """撤销码值项上的「已列全」（目录版本 +1，不提交）：值、含义、状态、来源都不动。已经不是已列全了（别人刚改过）返回
+    None。
+
+    目录从库里重读（populate_existing），不用会话里缓存的那个对象：前一条观察撤销过、随后它的保存点回滚了，那个对象上还是
+    回滚掉的版本（只 refresh 过、没改过的对象，保存点回滚时 SQLAlchemy 不让它过期）。身份表对这种干净对象只是弱引用，平常
+    早被回收、下次查询从库里重建；可只要别处还握着它（异常的回溯、调用方的变量），拿它的版本去写，乐观锁就对不上，而
+    write_entry 冲突时回滚的是整个会话、不只是保存点，这张表的剖析结果跟着丢掉——所以不赌它有没有被回收，也不在同一个
+    事务里撞了重试。重读之后到写进去之间别人提交不了新的目录版本：这张表的结果在同一个事务里已经写过（剖析任务的
+    _save_table），这个事务拿着 SQLite 的写锁。"""
+    await session.execute(select(CatalogNote).where(CatalogNote.source_id == source_id, CatalogNote.table_name == table)
+                          .execution_options(populate_existing=True))
+    entry = await catalog.read_entry(session, source_id, table)
+    if entry is None:
+        return None
+    notes = entry.notes
+    item = ((notes.get("columns") or {}).get(column) or {}).get("codes")
+    if not catalog.codes_complete(item):
+        return None
+    item.pop("complete", None)
+    item["updated_at"] = catalog.now_iso()
+    written = await catalog.write_entry(session, source_id, table, notes, if_version=entry.version, actor=OBSERVER,
+                                        commit=False)
+    return written.version
 
 
 async def save(session: AsyncSession, source: Any, table: str, result: Any, *, job_id: str | None,
                seq: int | None, detected: Any) -> int:
     """这张表这一次的观察写进 table_observations、和同口径的历史比，比出来的取值变化写进 source_changes（不提交：和剖析
-    结果、目录在同一个事务里）。返回记了几条新变化。出错只记日志、不挡剖析（观察是辅助步骤）。"""
-    from app.data import source_changes as sc
-    from app.data import value_drift as vd
-
+    结果、目录在同一个事务里）。返回记了几条新变化。每条观察一个保存点，出错只回滚那一条、记日志，不挡剖析（观察是辅助
+    步骤，见模块说明）。"""
     observations = list(getattr(result, "observations", None) or [])
     if not observations:
         return 0
-    fresh = 0
+    # 剖析这张表自己的写入先落下去：那里出错是剖析的事，照常往外抛，不算在观察头上
+    await session.flush()
     try:
-        entry = await catalog.read_entry(session, source.id, table)
-        notes = entry.notes if entry else {}
-        meta = ((getattr(source, "schema_cache", None) or {}).get("tables") or {}).get(table)
-        rules = _rules_columns(notes, meta, table, getattr(source, "kind", None))
-        now = utcnow()
-        today = now.date()
-        for obs in observations:
-            kind, column = obs["kind"], obs["column"]
-            history = await _history(session, source.id, table, kind, column)
-            session.add(TableObservation(source_id=source.id, table_name=table, column_name=column, kind=kind,
-                                         basis=obs["basis"], basis_key=obs["basis_key"], data=obs["data"],
-                                         job_id=job_id, task_seq=seq, at=now))
-            await session.flush()
-            await _prune(session, source.id, table, kind, column)
-            item = ((notes.get("columns") or {}).get(column) or {}).get("codes") if column else None
-            known = list((item or {}).get("value") or {}) if isinstance(item, Mapping) and \
-                item.get("status") != "rejected" else []
-            current = vd.Obs(kind, obs["basis_key"], obs["data"], now)
-            past = [vd.Obs(h.kind, h.basis_key, h.data or {}, h.at) for h in history]
-            findings = vd.compare(current, past, today=today, known=known,
-                                  complete=catalog.codes_complete(item))
-            for f in findings:
-                after = dict(f.after) if isinstance(f.after, Mapping) else {"value": f.after}
-                if f.revoke_complete:
-                    version = await _revoke_complete(session, source.id, table, column)
-                    after.update(complete_revoked=version is not None, catalog_version=version)
-                touched = bool(column) and column.lower() in rules
-                change = sc.NewChange(kind=f.kind, table=table, column=column or None, severity=f.severity,
-                                      before=f.before, after=after,
-                                      basis={"method": obs["basis"].get("method"), "key": obs["basis_key"],
-                                             "query_refs": obs["basis"].get("query_refs") or []},
-                                      affects={"catalog": [], "templates": [], "rules_touched": touched},
-                                      key=(obs["basis_key"], f.key))
-                _row, new = await sc.record(session, source, change, detected)
-                fresh += int(new)
-    except Exception:  # noqa: BLE001 - 观察是辅助步骤：落库、比较出错不挡剖析结果
-        logger.exception("保存 %s 的取值观察时出错", table)
+        async with session.begin_nested():
+            entry = await catalog.read_entry(session, source.id, table)
+            notes = entry.notes if entry else {}
+            meta = ((getattr(source, "schema_cache", None) or {}).get("tables") or {}).get(table)
+            rules = _rules_columns(notes, meta, table, getattr(source, "kind", None))
+    except Exception:  # noqa: BLE001 - 观察是辅助步骤
+        logger.exception("保存 %s 的取值观察时读目录出错", table)
+        return 0
+    now = utcnow()
+    fresh = 0
+    for obs in observations:
+        try:
+            async with session.begin_nested():
+                new = await _save_one(session, source, table, obs, notes=notes, rules=rules, now=now, job_id=job_id,
+                                      seq=seq, detected=detected)
+        except Exception:  # noqa: BLE001 - 这一条回滚了，别的观察和剖析结果照常
+            logger.exception("保存 %s 的取值观察（%s %s）时出错", table, obs.get("kind"), obs.get("column") or "")
+            continue
+        fresh += new
+    return fresh
+
+
+async def _save_one(session: AsyncSession, source: Any, table: str, obs: Mapping[str, Any], *,
+                    notes: Mapping[str, Any], rules: set[str], now: datetime, job_id: str | None, seq: int | None,
+                    detected: Any) -> int:
+    """一条观察：落库、修剪、和同口径的历史比，比出来的变化逐条记下（要撤销已列全的先撤销）。在调用方的保存点里：
+    中间哪一步出错，这条观察、它撤销的已列全、它记的变化一起回滚，不会只撤了已列全却没有那条变化。返回记了几条新变化。"""
+    from app.data import source_changes as sc
+    from app.data import value_drift as vd
+
+    kind, column = obs["kind"], obs["column"]
+    history = await _history(session, source.id, table, kind, column)
+    session.add(TableObservation(source_id=source.id, table_name=table, column_name=column, kind=kind,
+                                 basis=obs["basis"], basis_key=obs["basis_key"], data=obs["data"],
+                                 job_id=job_id, task_seq=seq, at=now))
+    await session.flush()
+    await _prune(session, source.id, table, kind, column)
+    item = ((notes.get("columns") or {}).get(column) or {}).get("codes") if column else None
+    known = list((item or {}).get("value") or {}) if isinstance(item, Mapping) and \
+        item.get("status") != "rejected" else []
+    current = vd.Obs(kind, obs["basis_key"], obs["data"], now)
+    past = [vd.Obs(h.kind, h.basis_key, h.data or {}, h.at) for h in history]
+    fresh = 0
+    for f in vd.compare(current, past, today=now.date(), known=known, complete=catalog.codes_complete(item)):
+        after = dict(f.after) if isinstance(f.after, Mapping) else {"value": f.after}
+        if f.revoke_complete:
+            version = await _revoke_complete(session, source.id, table, column)
+            after.update(complete_revoked=version is not None, catalog_version=version)
+        touched = bool(column) and column.lower() in rules
+        change = sc.NewChange(kind=f.kind, table=table, column=column or None, severity=f.severity,
+                              before=f.before, after=after,
+                              basis={"method": obs["basis"].get("method"), "key": obs["basis_key"],
+                                     "query_refs": obs["basis"].get("query_refs") or []},
+                              affects={"catalog": [], "templates": [], "rules_touched": touched},
+                              key=(obs["basis_key"], f.key))
+        _row, new = await sc.record(session, source, change, detected)
+        fresh += int(new)
     return fresh
 
 
@@ -401,26 +430,32 @@ def coverage_drops(findings: Iterable[Mapping[str, Any]]) -> list[dict[str, Any]
 
 async def record_coverage_drops(session: AsyncSession, source: Any, table: str, result: Any, *, detected: Any) -> int:
     """这张表这次剖析里覆盖率掉下来的关系记成 coverage_drop（不提交，和剖析结果同一个事务）。人点的剖析也记：变化记录是
-    全站一份。出错只记日志。"""
+    全站一份。每条一个保存点，出错只回滚那一条、记日志（同 save）。"""
     from app.data import source_changes as sc
 
-    try:
-        fresh = 0
-        for drop in coverage_drops(getattr(result, "findings", None) or []):
-            _row, new = await sc.record(session, source, sc.NewChange(
-                kind=sc.COVERAGE_DROP, table=table, severity="warning",
-                before={"coverage": drop["before"], "path": drop["path"], "target": drop["target"]},
-                after={"coverage": drop["after"], "confirmed": drop["confirmed"]},
-                basis={"method": "relation_sample", "key": "relation"},
-                affects={"catalog": [{"table": table, "path": drop["path"], "label": drop["target"],
-                                      "status": "confirmed" if drop["confirmed"] else "proposed",
-                                      "source": None, "needs_action": False}], "templates": [], "rules_touched": False},
-                key=(drop["path"], round(float(drop["after"]), 3))), detected)
-            fresh += int(new)
-        return fresh
-    except Exception:  # noqa: BLE001
-        logger.exception("记录 %s 的关联覆盖率下降时出错", table)
+    drops = coverage_drops(getattr(result, "findings", None) or [])
+    if not drops:
         return 0
+    await session.flush()                # 同 save：剖析自己的写入先落下去
+    fresh = 0
+    for drop in drops:
+        try:
+            async with session.begin_nested():
+                _row, new = await sc.record(session, source, sc.NewChange(
+                    kind=sc.COVERAGE_DROP, table=table, severity="warning",
+                    before={"coverage": drop["before"], "path": drop["path"], "target": drop["target"]},
+                    after={"coverage": drop["after"], "confirmed": drop["confirmed"]},
+                    basis={"method": "relation_sample", "key": "relation"},
+                    affects={"catalog": [{"table": table, "path": drop["path"], "label": drop["target"],
+                                          "status": "confirmed" if drop["confirmed"] else "proposed",
+                                          "source": None, "needs_action": False}],
+                             "templates": [], "rules_touched": False},
+                    key=(drop["path"], round(float(drop["after"]), 3))), detected)
+        except Exception:  # noqa: BLE001
+            logger.exception("记录 %s 的关联覆盖率下降（%s）时出错", table, drop.get("path"))
+            continue
+        fresh += int(new)
+    return fresh
 
 
 # ==========================================================================
diff --git a/backend/tests/test_catalog_observe.py b/backend/tests/test_catalog_observe.py
index 73260c8b..68d07455 100644
--- a/backend/tests/test_catalog_observe.py
+++ b/backend/tests/test_catalog_observe.py
@@ -23,7 +23,7 @@ from app.data import catalog, catalog_profile, catalog_profile_jobs, catalog_pro
 from app.data import table_versions
 from app.data.engine import MASK_COLUMNS_OPTION
 from app.db.base import SessionLocal
-from app.db.models import CatalogProfileTask, DataSource
+from app.db.models import CatalogProfileJob, CatalogProfileTask, DataSource
 from app.db.models_sources import SourceChange, TableObservation
 from app.main import app
 from tests.fixtures.sources import drop_source, introspect_now
@@ -300,3 +300,159 @@ def test_coverage_drop_only_when_crossing_the_verified_line():
         {"kind": "codes", "path": "columns.x.codes"},
     ])
     assert [(d["path"], d["before"], d["after"]) for d in drops] == [("relations.r1", 0.99, 0.6)]
+
+
+# ==========================================================================
+# 观察写库出错只回滚它自己：辅助步骤失败不挡剖析（保存点）
+# ==========================================================================
+
+
+def _unserializable(monkeypatch, kind: str) -> None:
+    """让某一种观察带上序列化不了的值：写 table_observations 那一刻 flush 报错（模拟观察写库出错）。"""
+    real = observe._observation
+
+    def broken(k, column, basis, data, refs):
+        out = real(k, column, basis, data, refs)
+        if k == kind:
+            out["data"] = {**out["data"], "bad": object()}
+        return out
+
+    monkeypatch.setattr(observe, "_observation", broken)
+
+
+async def _job_and_task(job_id: str):
+    async with SessionLocal() as session:
+        job = await session.get(CatalogProfileJob, job_id)
+        [task] = (await session.execute(select(CatalogProfileTask).where(CatalogProfileTask.job_id == job_id))).scalars()
+        return job, task
+
+
+async def test_observation_write_error_does_not_take_the_profile_down(shop, monkeypatch):
+    """一条观察写库出错（flush 报错）只回滚这条观察的保存点：同一张表的剖析结果、写进目录的结论照常提交，别的观察照常
+    落库，任务照常结束。以前和剖析共用一个事务又没有保存点：这条一错整个事务就回滚了，外层提交时抛错，剖析写的目录跟着
+    丢掉。"""
+    sid, _ = shop
+    async with SessionLocal() as session:
+        # secret 记成状态类的度量、去掉码值：剖析会给它取值分布、往目录里提一份码值（这次剖析有东西写进目录）
+        entry = await catalog.read_entry(session, sid, "orders")
+        notes = entry.notes
+        notes["columns"]["secret"] = {"measure": catalog.make_item("status", "llm")}
+        before = await catalog.write_entry(session, sid, "orders", notes, if_version=entry.version, actor="测试员")
+    _unserializable(monkeypatch, "freshness")
+    job_id = await _observe(sid, scope="maintenance")
+    job, task = await _job_and_task(job_id)
+    async with SessionLocal() as session:
+        after = await catalog.read_entry(session, sid, "orders")
+    assert job.status == "succeeded", job.error
+    assert task.status == "ok" and task.error is None
+    assert task.added == 1 and after.version == before.version + 1 and task.version == after.version
+    assert after.notes["columns"]["secret"]["codes"]["source"] == "profile", "剖析写进目录的结论照常提交"
+    got = {(o.kind, o.column_name) for o in await _obs(sid)}
+    assert ("freshness", "paid_on") not in got, "出错的那条观察回滚了"
+    assert {("rows", ""), ("codes_observe", "status"), ("daily_volume", "paid_on")} <= got, "别的观察照常落库"
+
+
+async def test_a_failing_observation_takes_its_revoke_back_with_it(shop, monkeypatch):
+    """前一条观察撤销了「已列全」、随后它记变化时出错：整条观察连同撤销一起回滚，不会只撤了已列全却没有那条变化；后一条
+    观察照常撤销另一列的已列全、记下变化，这张表的结果照常提交。以前没有保存点：撤销留在事务里、变化没记上，后面的观察
+    也不再保存。"""
+    from app.data import source_changes
+
+    sid, path = shop
+    async with SessionLocal() as session:
+        entry = await catalog.read_entry(session, sid, "orders")
+        notes = entry.notes
+        notes["columns"]["secret"]["codes"] = catalog.make_item({"y": "保密"}, "human", complete=True)
+        await catalog.write_entry(session, sid, "orders", notes, if_version=entry.version, actor="测试员")
+    db = sqlite3.connect(path)
+    db.execute("INSERT INTO orders VALUES (999, 'held', ?, 1, 'x')", ((TODAY - timedelta(days=2)).isoformat(),))
+    db.commit()
+    db.close()
+    async with SessionLocal() as session:
+        base = (await catalog.read_entry(session, sid, "orders")).version
+    real = source_changes.record
+
+    async def broken(session, source, change, detected):
+        if change.column == "status":
+            change.after = {**(change.after or {}), "bad": object()}
+        return await real(session, source, change, detected)
+
+    monkeypatch.setattr(source_changes, "record", broken)
+    job_id = await _observe(sid)
+    job, task = await _job_and_task(job_id)
+    async with SessionLocal() as session:
+        after = await catalog.read_entry(session, sid, "orders")
+        changes = list((await session.execute(select(SourceChange).where(SourceChange.source_id == sid))).scalars())
+    assert job.status == "succeeded", job.error
+    assert task.status == "ok" and task.error is None, "这张表的结果照常提交"
+    assert [(c.kind, c.column_name) for c in changes] == [("codes_new", "secret")]
+    assert after.version == base + 1 and after.updated_by == observe.OBSERVER
+    assert catalog.codes_complete(after.notes["columns"]["status"]["codes"]), "status 那条观察回滚了：已列全没撤"
+    assert not catalog.codes_complete(after.notes["columns"]["secret"]["codes"])
+    got = {(o.kind, o.column_name) for o in await _obs(sid)}
+    assert ("codes_observe", "status") not in got and ("codes_observe", "secret") in got
+
+
+async def test_coverage_drop_write_error_rolls_back_only_itself(shop):
+    """记关联覆盖率下降出错（这里是更新已有那一条时 flush 报错）只回滚它自己：同一个事务里这张表的剖析结果照常提交。"""
+    from app.data.source_changes import Detected
+
+    sid, _ = shop
+    drop = {"kind": "relation", "path": "relations.r1", "target": "orders → stores", "previous_coverage": 0.99,
+            "coverage": 0.6}
+    detected = Detected(job_kind="profile")
+    async with SessionLocal() as session:
+        row = await session.get(DataSource, sid)
+        assert await observe.record_coverage_drops(session, row, "orders", SimpleNamespace(findings=[drop]),
+                                                   detected=detected) == 1
+        await session.commit()
+    async with SessionLocal() as session:
+        row = await session.get(DataSource, sid)
+        entry = await catalog.read_entry(session, sid, "orders")
+        await catalog.write_entry(session, sid, "orders", {**entry.notes, "description": catalog.make_item(
+            "订单", "human")}, if_version=entry.version, actor="测试员", commit=False)
+        # 同一个指纹再测到（更新那一条），这次带上序列化不了的值
+        bad = {**drop, "target": object()}
+        assert await observe.record_coverage_drops(session, row, "orders", SimpleNamespace(findings=[bad]),
+                                                   detected=detected) == 0
+        await session.commit()
+    async with SessionLocal() as session:
+        after = await catalog.read_entry(session, sid, "orders")
+        [change] = (await session.execute(select(SourceChange).where(SourceChange.source_id == sid))).scalars()
+    assert after.version == entry.version + 1 and after.notes["description"]["value"] == "订单", "同一事务的写入照常提交"
+    assert change.seen_count == 1 and change.before["target"] == "orders → stores", "出错的那次更新回滚了"
+
+
+async def test_revoke_writes_against_the_database_version_not_the_session_cache(shop):
+    """撤销已列全按库里的目录版本写。会话里还握着保存点回滚前的那个目录对象时（干净对象在身份表里只是弱引用，平常早被
+    回收，但异常的回溯、调用方的变量都会握住它），它上面是回滚掉的版本：拿它去写乐观锁对不上，write_entry 回滚的是整个
+    会话，同一个事务里先写的东西（这里是一条观察，代表这张表的剖析结果）跟着丢掉。"""
+    import contextlib
+
+    from app.db.models import CatalogNote
+
+    sid, _ = shop
+    async with SessionLocal() as session:
+        entry = await catalog.read_entry(session, sid, "orders")
+        notes = entry.notes
+        notes["columns"]["secret"]["codes"] = catalog.make_item({"y": "保密"}, "human", complete=True)
+        base = (await catalog.write_entry(session, sid, "orders", notes, if_version=entry.version,
+                                          actor="测试员")).version
+    async with SessionLocal() as session:
+        session.add(TableObservation(source_id=sid, table_name="orders", column_name="", kind="rows",
+                                     basis={"method": "count"}, basis_key="count", data={"rows": 1}))
+        await session.flush()
+        held = (await session.execute(select(CatalogNote).where(CatalogNote.source_id == sid))).scalar_one()
+        with contextlib.suppress(RuntimeError):
+            async with session.begin_nested():
+                assert await observe._revoke_complete(session, sid, "orders", "status") == base + 1
+                raise RuntimeError("这条观察随后出错")
+        assert held.version == base + 1, "前提：会话里握着的对象上还是回滚掉的版本"
+        assert await observe._revoke_complete(session, sid, "orders", "secret") == base + 1
+        await session.commit()
+    async with SessionLocal() as session:
+        after = await catalog.read_entry(session, sid, "orders")
+    assert after.version == base + 1 and after.updated_by == observe.OBSERVER
+    assert catalog.codes_complete(after.notes["columns"]["status"]["codes"]), "回滚掉的那次撤销没有生效"
+    assert not catalog.codes_complete(after.notes["columns"]["secret"]["codes"])
+    assert [(o.kind, o.data) for o in await _obs(sid)] == [("rows", {"rows": 1})], "同一个事务里先写的照常提交"