fix(maps): 画布连线键与目标改名解耦（objective_id 锚定）+ nodeRefs 位置兜底 + 保守反错绑断言（P3）

- map_connections 读侧以 from/to_objective_id 反查当前 dimensions 树定位节点，改名/重排不再静默断线
- 悬停/拖拽期 map_objectives 未返回时按位置兜底（仅同层同序位有效），并记录 stale_objective_id
- 新增 tests/test_conn_key_rename_stability.py（含 2 条保守反错绑负测用例）
- 零 DDL、零 health_score 改动（问3=A）

backend/app/api/maps.py

@@ -681,9 +681,102 @@ def _get_connections(m: StrategicMap) -> list:
     return m.canvas_data["connections"]
 
 
+def _objective_key_index(db: Session, m: StrategicMap):
+    """连线解析锚点：MapObjective.id → 画布渲染键 `${dim}-${idx}`（以当前 dimensions 顺序为准）
+
+    返回 (id2key, name2key, dim_len)：
+      · id2key  : 本表行 id → `${dim}-${idx}`（map_connections.from/to_objective_id 的锚点）
+      · name2key: (dim, 归一化名称) → `${dim}-${idx}`（历史 `层:中文名` 的兜底，改名即失效）
+      · dim_len : {dim: len(objectives)}（`层-序号` 越界判定）
+
+    只按名称把 dimensions 节点对回 map_objectives 行，**不读** dims JSON 里的 `_objective_id`
+    —— 那是 objectives 表（实体级目标）的 id 空间，与 map_objectives 行 id 不是一回事。
+    """
+    dims = _dims_list(m.dimensions)
+    rows = db.query(MapObjective).filter(MapObjective.map_id == m.id).order_by(
+        MapObjective.sort_order, MapObjective.id).all()
+    by_name = {}
+    for o in rows:
+        by_name[(o.dimension_key, _norm_obj_name(o.name))] = o.id
+    id2key, name2key, dim_len = {}, {}, {}
+    for d in dims:
+        dk = d.get("key") or ""
+        if not dk:
+            continue
+        objs = d.get("objectives") or []
+        dim_len[dk] = len(objs)
+        for idx, o in enumerate(objs):
+            if not isinstance(o, dict):
+                continue
+            key = f"{dk}-{idx}"
+            nm = _norm_obj_name(o.get("name"))
+            if nm:
+                name2key[(dk, nm)] = key
+            oid = by_name.get((dk, nm))
+            if oid is not None:
+                id2key[oid] = key
+    return id2key, name2key, dim_len
+
+
+def _key_for_node(node, id2key, name2key, dim_len, objective_id=None):
+    """单端节点解析 → (key, source, reason)
+
+    顺序：objective_id 锚点 → `层:中文名` → `层-序号`。
+    改名后 `objective_id` 仍指向同一节点 ⇒ key 不变、线不断；锚点失效（目标已删/历史脏 id/
+    跨 map）才回退原名字串，仍解析不到才返回 None（调用方计入 unresolved）。
+    """
+    if objective_id:
+        hit = id2key.get(objective_id)
+        if hit:
+            return hit, "objective_id", None
+    ref = parse_node_ref(node)
+    if ref["error"]:
+        return None, None, ref["error"]
+    layer = ref["layer"]
+    stale = "stale_objective_id" if objective_id else None
+    if ref["kind"] == "name":
+        want = _norm_obj_name(ref["value"])
+        hit = name2key.get((layer, want))
+        if hit:
+            return hit, "name", stale
+        return None, None, (stale or "objective_unresolved")
+    idx = ref["value"]
+    if 0 <= idx < dim_len.get(layer, 0):
+        return f"{layer}-{idx}", "index", stale
+    return None, None, "index_out_of_range"
+
+
+def _attach_keys(db: Session, m: StrategicMap, items: list):
+    """给连线条目附 `from_key/to_key`（渲染键）+ 未解析清单。
+
+    条目原地补字段：`{side}_key` / `{side}_key_source` / `{side}_key_reason`，
+    键与中文名解耦的唯一出口（前端按 key 画线，缺失回落原值）。
+    """
+    id2key, name2key, dim_len = _objective_key_index(db, m)
+    unresolved = []
+    for it in items:
+        for side in ("from", "to"):
+            key, source, reason = _key_for_node(
+                it.get(side), id2key, name2key, dim_len, it.get(f"{side}_objective_id"))
+            it[f"{side}_key"] = key
+            it[f"{side}_key_source"] = source
+            it[f"{side}_key_reason"] = reason
+            if key is None:
+                unresolved.append({
+                    "from": it.get("from"), "to": it.get("to"),
+                    "side": side, "node": it.get(side),
+                    "reason": reason, "objective_id": it.get(f"{side}_objective_id"),
+                })
+    return items, unresolved
+
+
 @router.get("/{map_id}/connections")
 def list_connections(map_id: int, db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)):
-    """读连线：表优先，表空回落 canvas_data（兼容期）。"""
+    """读连线：表优先，表空回落 canvas_data（兼容期）。
+
+    P3：每条连线附 `from_key/to_key` = `${dim}-${idx}`（由 objective_id 在当前 dimensions
+    树反查）+ 顶层 `unresolved` 计数与清单。前端优先按 key 画线 ⇒ 目标改名不断线。
+    """
     m = db.query(StrategicMap).filter(StrategicMap.id == map_id, StrategicMap.entity_id == entity_id).first()
     if not m:
         raise HTTPException(404, "战略地图不存在")
@@ -691,25 +784,33 @@ def list_connections(map_id: int, db: Session = Depends(get_db), entity_id: int
     if _table_ready(db):
         rows = _table_rows(db, m)
         if rows:
+            items = [{
+                "from": r.from_node, "to": r.to_node, "style": "solid",
+                "effect": r.effect or "positive", "label": r.label or "",
+                "from_layer": r.from_layer, "to_layer": r.to_layer,
+                "from_objective_id": r.from_objective_id, "to_objective_id": r.to_objective_id,
+            } for r in rows]
+            items, unresolved = _attach_keys(db, m, items)
             return {
-                "source": "table", "total": len(rows),
-                "connections": [{
-                    "from": r.from_node, "to": r.to_node, "style": "solid",
-                    "effect": r.effect or "positive", "label": r.label or "",
-                    "from_layer": r.from_layer, "to_layer": r.to_layer,
-                    "from_objective_id": r.from_objective_id, "to_objective_id": r.to_objective_id,
-                } for r in rows],
+                "source": "table", "total": len(items),
+                "unresolved": len(unresolved), "unresolved_items": unresolved,
+                "connections": items,
             }
 
     conns = _get_connections(m)
+    items = [{
+        "from": c.get("from"), "to": c.get("to"),
+        "style": c.get("style") or "solid",
+        "effect": c.get("effect") or "positive",
+        "label": c.get("label") or "",
+        "from_objective_id": c.get("from_objective_id"),
+        "to_objective_id": c.get("to_objective_id"),
+    } for c in conns if isinstance(c, dict)]
+    items, unresolved = _attach_keys(db, m, items)
     return {
-        "source": "canvas_data", "total": len(conns),
-        "connections": [{
-            "from": c.get("from"), "to": c.get("to"),
-            "style": c.get("style") or "solid",
-            "effect": c.get("effect") or "positive",
-            "label": c.get("label") or "",
-        } for c in conns if isinstance(c, dict)],
+        "source": "canvas_data", "total": len(items),
+        "unresolved": len(unresolved), "unresolved_items": unresolved,
+        "connections": items,
     }
 
 
@@ -751,6 +852,7 @@ def analyze_connections(map_id: int, db: Session = Depends(get_db), entity_id: i
     layer_stats = {k: {"as_from": 0, "as_to": 0} for k in LAYER_ORDER}
     cross_layer, reversed_edges, broken, unknown = {}, [], [], []
     same_layer = 0
+    items, unresolved = _attach_keys(db, m, items)
     for it in items:
         fl, tl = it.get("from_layer"), it.get("to_layer")
         if fl in layer_stats:
@@ -768,12 +870,14 @@ def analyze_connections(map_id: int, db: Session = Depends(get_db), entity_id: i
             elif _LAYER_INDEX[tl] < _LAYER_INDEX[fl]:
                 reversed_edges.append({"from": it["from"], "to": it["to"],
                                        "from_layer": fl, "to_layer": tl})
-        if not it.get("from_objective_id"):
+        if not it.get("from_key"):
             broken.append({"from": it["from"], "to": it["to"], "side": "from",
-                           "node": it["from"], "reason": parse_node_ref(it["from"]).get("error") or "objective_unresolved"})
-        if not it.get("to_objective_id"):
+                           "node": it["from"],
+                           "reason": it.get("from_key_reason") or "objective_unresolved"})
+        if not it.get("to_key"):
             broken.append({"from": it["from"], "to": it["to"], "side": "to",
-                           "node": it["to"], "reason": parse_node_ref(it["to"]).get("error") or "objective_unresolved"})
+                           "node": it["to"],
+                           "reason": it.get("to_key_reason") or "objective_unresolved"})
 
     warnings = []
     if reversed_edges:
@@ -791,6 +895,8 @@ def analyze_connections(map_id: int, db: Session = Depends(get_db), entity_id: i
         "same_layer_count": same_layer,
         "reversed": reversed_edges, "broken": broken,
         "unknown_layer": unknown, "warnings": warnings,
+        "unresolved": len(unresolved), "unresolved_items": unresolved,
+        "items": items,
     }
 
 
@@ -925,26 +1031,98 @@ def m_to_dict(m, db: Session = None):
 
 
 def _sync_map_objectives(m, db):
-    """保存时：将dimensions JSON中的目标同步到map_objectives表"""
-    db.query(MapObjective).filter(MapObjective.map_id == m.id).delete()
-    dims = m.dimensions
-    if isinstance(dims, str):
-        dims = json.loads(dims)
-    dims = dims or []
+    """保存时：将dimensions JSON中的目标同步到map_objectives表（就地 upsert）
+
+    2026-10-22 cma-map-canvas-conn-key-rename-stable（P3）：
+    原实现是 delete+insert —— MySQL 的 AUTO_INCREMENT 计数器不随 DELETE 归零，每次保存
+    都会给同一节点换一批新行 id，`map_connections.from/to_objective_id` 随即变成野指针
+    （读侧再按 id 反查就查不到），「用 objective_id 与中文名解耦」的前提被静默破坏。
+    改为就地 upsert：行 id 全程稳定，连线锚点才真的能扛住改名。
+
+    定位顺序：dims 的 `_mo_id`（读侧由 _merge_map_objectives 回写）→ (dimension_key, 归一化名称)。
+    **不做位置兜底**（口径 B，ASK-20261023-04 任总 2026-10-23 15:13「放行 4」）：
+    未命中名字的项一律视为新目标，未被占用的存量行一律视为已删除。理由见下方 ③。
+    """
+    dims = _dims_list(m.dimensions)
+    entity_id = m.entity_id if m.entity_id is not None else 1
+    rows = db.query(MapObjective).filter(MapObjective.map_id == m.id).order_by(
+        MapObjective.sort_order, MapObjective.id).all()
+    by_id = {r.id: r for r in rows}
+    used = set()
+
+    # 计划项：按 dims 顺序展开
+    planned = []
     for dim in dims:
-        for i, obj in enumerate(dim.get("objectives", [])):
-            mo = MapObjective(
-                map_id=m.id,
-                entity_id=m.entity_id if m.entity_id is not None else 1,  # 强制继承地图企业(OpenCode审查#11)
-                dimension_key=dim.get("key", ""),
-                name=obj.get("name", ""),
-                description=obj.get("description", ""),
-                icon=obj.get("icon", "target"),
-                kpis=obj.get("kpis", []),
-                sort_order=i,
-            )
-            db.add(mo)
+        dk = dim.get("key", "")
+        for i, obj in enumerate(dim.get("objectives") or []):
+            if isinstance(obj, dict):
+                planned.append((dk, i, obj))
+
+    assigned = {}
+
+    # ① 显式锚点：客户端回带的 `_mo_id`（读侧 _merge_map_objectives 写入）优先级最高
+    for idx, (dk, i, obj) in enumerate(planned):
+        try:
+            mo_id = int(obj.get("_mo_id")) if obj.get("_mo_id") is not None else None
+        except (TypeError, ValueError):
+            mo_id = None
+        if mo_id is not None and mo_id in by_id and mo_id not in used:
+            assigned[idx] = by_id[mo_id]
+            used.add(mo_id)
+
+    # ② 名称批：**先全表跑完名字匹配，再做位置兜底**。
+    #   逐项贪心（名→序位→空位）时，改名项在名字层落空后会立刻抢占同序位行，把
+    #   「同一次保存里改名 + 同层调序」的两个目标 id 串位 —— 线静默画到别的节点上，
+    #   比断线更坏（看不见）。批处理顺序把名字仍有效的项先钉住，未命中的项再按顺序配对。
+    for idx, (dk, i, obj) in enumerate(planned):
+        if idx in assigned:
+            continue
+        want = _norm_obj_name(obj.get("name"))
+        cand = [r for r in rows
+                if r.id not in used and r.dimension_key == dk
+                and _norm_obj_name(r.name) == want]
+        if cand:
+            cand.sort(key=lambda r: (r.sort_order or 0, r.id))   # 同名重复取最靠前，结果确定
+            assigned[idx] = cand[0]
+            used.add(cand[0].id)
+
+    # ③ （口径 B · ASK-20261023-04 任总 2026-10-23 15:13「放行 4」）**不做位置兜底**。
+    #   老通道（不带 `_mo_id`）下「删一个 + 加一个」与「改名」数据上不可区分：按位置配对
+    #   在其中一种情况下必然静默错绑（新项继承被删项的锚点，线正常画着却是错的）。
+    #   错绑比断线更坏 —— 断线在读侧 `unresolved` 里可见、可计数。故：
+    #   · 名字未命中的项 → 一律当新目标插入（新行 id）；
+    #   · 未被占用的存量行 → 一律当已删除删除；
+    #   · 连线锚点由此指向被删行 ⇒ 读侧记 `stale_objective_id` 并计入 unresolved。
+    #   UI 通道（客户端回带 `_mo_id`）改名仍走 ① 精确保住锚点，本口径不影响它。
+    #   逐项贪心（名→序位→空位）的串位反证见 tests/test_conn_key_rename_stability.py
+    #   的 test_negative_greedy_matching_would_crossbind。
+
+    created = 0
+    keep = []
+    for idx, (dk, i, obj) in enumerate(planned):
+        target = assigned.get(idx)
+        if target is None:
+            target = MapObjective(map_id=m.id, entity_id=entity_id, dimension_key=dk or "finance")
+            db.add(target)
+            created += 1
+        target.entity_id = entity_id
+        target.dimension_key = dk
+        target.name = (obj.get("name") or "")[:200]
+        target.description = obj.get("description") or ""
+        target.icon = obj.get("icon") or "target"
+        target.kpis = obj.get("kpis") or []
+        target.sort_order = i
+        keep.append(target)
+
+    db.flush()                       # 新增行拿到 id 后才有 id 可比
+    deleted = 0
+    for r in rows:
+        if r.id not in used:
+            db.delete(r)
+            deleted += 1
     db.commit()
+    return {"planned": len(planned), "created": created, "deleted": deleted,
+            "kept": len(used)}
 
 
 def _merge_map_objectives(m, db):
@@ -978,6 +1156,9 @@ def _merge_map_objectives(m, db):
                     "icon": o.icon or existing.get("icon", "target"),
                     "kpis": o.kpis or existing.get("kpis", []),
                     **{k: existing[k] for k in existing if k not in ("name", "description", "icon", "kpis")},
+                    # P3：回写本表行 id（连线锚点的落点）。只在读响应里出现，不落 dimensions JSON；
+                    # 客户端原样带回保存时，_sync_map_objectives 据此就地更新、保住行 id。
+                    "_mo_id": o.id,
                 })
             dim["objectives"] = merged
     m.dimensions = dims

backend/tests/test_conn_key_rename_stability.py

@@ -0,0 +1,415 @@
+"""P3 阶段A · 连线键解耦（读侧解析键）验收测试 —— 沙箱，不连生产。
+
+判据（对应 P3 单验收标准 2/3）：
+  1. 目标改名后 `from_key/to_key` 跟随**同一节点**（key 由 objective_id 反查，不随中文名变）；
+  2. 索引位移（删掉前一个目标）后 key 跟随节点到新位置；
+  3. 负测：拿掉 `objective_id` 锚点（等价于"未实现解析键"）且名字已改 ⇒ 必须解析失败
+     并被计入 `unresolved`，证明判据有区分度、不是空过。
+"""
+import copy
+
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy.orm import Session
+
+from app.models import StrategicMap, MapObjective, MapConnection
+from tests.conftest import create_test_user, get_token_for_user, auth_header
+
+DIMS = [
+    {"key": "finance", "name": "财务层", "icon": "F", "color": "#F56C6C",
+     "objectives": [{"name": "营业收入"}, {"name": "营业利润"}]},
+    {"key": "customer", "name": "客户层", "icon": "C", "color": "#409EFF",
+     "objectives": [{"name": "客户留存"}]},
+    {"key": "process", "name": "流程层", "icon": "P", "color": "#67C23A",
+     "objectives": [{"name": "交付效率"}, {"name": "质量闭环"}]},
+    {"key": "learning", "name": "学习层", "icon": "L", "color": "#E6A23C",
+     "objectives": [{"name": "全员培训"}, {"name": "能力认证"}]},
+]
+
+
+def _seed(db: Session) -> StrategicMap:
+    m = StrategicMap(entity_id=1, title="p3-conn-key", status="draft",
+                     dimensions=copy.deepcopy(DIMS), canvas_data={"connections": []})
+    db.add(m)
+    db.commit()
+    db.refresh(m)
+    for d in DIMS:
+        for i, o in enumerate(d["objectives"]):
+            db.add(MapObjective(entity_id=1, map_id=m.id, dimension_key=d["key"],
+                                name=o["name"], sort_order=i))
+    db.commit()
+    return m
+
+
+def _save_dims(client, headers, db: Session, m: StrategicMap, dims) -> None:
+    """整表 PUT（legacy 通道，批处理/脚本真实会走这条）"""
+    db.refresh(m)
+    body = {"dimensions": dims,
+            "canvas_data": copy.deepcopy(m.canvas_data),
+            "base_updated_at": m.updated_at.isoformat() if m.updated_at else None}
+    r = client.put(f"/api/cma/maps/{m.id}", json=body, headers=headers)
+    assert r.status_code == 200, f"整表 PUT 失败: {r.status_code} {r.text[:300]}"
+
+
+def _conn(client, headers, map_id: int) -> dict:
+    r = client.get(f"/api/cma/maps/{map_id}/connections", headers=headers)
+    assert r.status_code == 200, r.text[:300]
+    return r.json()
+
+
+def _mk_conn(client, headers, map_id: int, frm: str, to: str) -> None:
+    r = client.post(f"/api/cma/maps/{map_id}/connections", headers=headers,
+                    json={"from": frm, "to": to})
+    assert r.status_code == 200, f"建连线失败: {r.status_code} {r.text[:300]}"
+
+
+def test_rename_keeps_render_key(client: TestClient, db: Session):
+    """UI 通道（带回 `_mo_id`）改名 → 连线仍挂同一节点（key 不变、unresolved=0）
+
+    口径 B（ASK-20261023-04，任总 2026-10-23 15:13「放行 4」）：只有带回 `_mo_id` 的
+    通道能精确保住锚点；不带 `_mo_id` 的老通道改名会断线但可见，
+    见 test_rename_without_moid_breaks_visibly。
+    """
+    create_test_user(db)
+    h = auth_header(get_token_for_user(client))
+    m = _seed(db)
+    _mk_conn(client, h, m.id, "learning:全员培训", "process:交付效率")
+
+    base = _conn(client, h, m.id)
+    assert base["total"] == 1 and base["unresolved"] == 0, base
+    c0 = base["connections"][0]
+    assert c0["from_key"] == "learning-0" and c0["to_key"] == "process-0", c0
+    assert c0["from_key_source"] == "objective_id"
+
+    # 真实 UI 通道：先读（读侧 m_to_dict→_merge_map_objectives 回写 `_mo_id`），改名后整表回传
+    r = client.get("/api/cma/maps", headers=h)
+    assert r.status_code == 200, r.text[:300]
+    got_map = next(x for x in r.json()["data"] if x["id"] == m.id)
+    dims = got_map["dimensions"]
+    assert dims[3]["objectives"][0].get("_mo_id"), "读侧未回写 _mo_id，UI 通道无法保锚点"
+    dims[3]["objectives"][0]["name"] = "全员培训A"
+    _save_dims(client, h, db, m, dims)
+
+    after = _conn(client, h, m.id)
+    assert after["total"] == 1, f"连线数减少: {after}"
+    assert after["unresolved"] == 0, after["unresolved_items"]
+    a0 = after["connections"][0]
+    # 原始串仍是旧名（禁回填），但渲染键已由锚点解析到新节点
+    assert a0["from"] == "learning:全员培训", a0
+    assert a0["from_key"] == "learning-0", a0
+    assert a0["from_key_source"] == "objective_id" and a0["from_key_reason"] is None, a0
+
+
+def test_index_shift_follows_node(client: TestClient, db: Session):
+    """删除同层前一个目标 ⇒ 节点索引位移，key 跟随节点而非冻结在旧序号"""
+    create_test_user(db)
+    h = auth_header(get_token_for_user(client))
+    m = _seed(db)
+    _mk_conn(client, h, m.id, "learning:能力认证", "process:交付效率")
+
+    base = _conn(client, h, m.id)["connections"][0]
+    assert base["from_key"] == "learning-1", base
+
+    dims = copy.deepcopy(m.dimensions)
+    dims[3]["objectives"] = dims[3]["objectives"][1:]  # 删掉「全员培训」
+    _save_dims(client, h, db, m, dims)
+
+    after = _conn(client, h, m.id)
+    assert after["unresolved"] == 0, after["unresolved_items"]
+    a0 = after["connections"][0]
+    assert a0["from_key"] == "learning-0", f"key 未跟随节点位移: {a0}"
+    assert a0["from_key_source"] == "objective_id", a0
+
+
+def test_negative_without_anchor_must_fail(client: TestClient, db: Session):
+    """负测：拿掉 objective_id 锚点 + 名字已改 ⇒ 必须解析失败并计入 unresolved
+
+    证明判据有区分度：若解析键能力缺失（=改造前行为），这条必然断线且被计数，
+    不会静默画到错节点上。
+    """
+    create_test_user(db)
+    h = auth_header(get_token_for_user(client))
+    m = _seed(db)
+    _mk_conn(client, h, m.id, "learning:全员培训", "process:交付效率")
+
+    dims = copy.deepcopy(m.dimensions)
+    dims[3]["objectives"][0]["name"] = "全员培训A"
+    _save_dims(client, h, db, m, dims)
+
+    # 模拟"无锚点"（历史脏行 / 未回填）：抹掉 from_objective_id
+    row = db.query(MapConnection).filter(MapConnection.map_id == m.id).first()
+    row.from_objective_id = None
+    db.commit()
+
+    broken = _conn(client, h, m.id)
+    b0 = broken["connections"][0]
+    assert b0["from_key"] is None, f"无锚点 + 旧名不应解析成功: {b0}"
+    assert b0["from_key_reason"] == "objective_unresolved", b0
+    assert broken["unresolved"] == 1, broken["unresolved_items"]
+    assert broken["unresolved_items"][0]["side"] == "from"
+    assert broken["unresolved_items"][0]["reason"] == "objective_unresolved"
+    # 另一端不受影响
+    assert b0["to_key"] == "process-0" and b0["to_key_source"] == "objective_id", b0
+
+
+def test_fallback_name_source_still_works(client: TestClient, db: Session):
+    """兼容期：历史行无 objective_id 但名字未改 ⇒ 仍按 `层:名字` 解析成功（不误报未解析）"""
+    create_test_user(db)
+    h = auth_header(get_token_for_user(client))
+    m = _seed(db)
+    _mk_conn(client, h, m.id, "learning:全员培训", "process:交付效率")
+
+    row = db.query(MapConnection).filter(MapConnection.map_id == m.id).first()
+    row.from_objective_id = None
+    db.commit()
+
+    got = _conn(client, h, m.id)
+    assert got["unresolved"] == 0, got["unresolved_items"]
+    c0 = got["connections"][0]
+    assert c0["from_key"] == "learning-0" and c0["from_key_source"] == "name", c0
+
+
+# ── project 2026-10-22 23:25 裁定的 3 项准入条件（阶段B 开跑前补测）─────────────
+# 条件2-a：连续两次保存（不改名，不带 _mo_id）⇒ map_objectives 行 id 集合不变（幂等）；
+# 条件2-b：同一次保存内「改名 + 同层调序」且不带 _mo_id ⇒ 两个目标的 id 不得串位；
+# 条件3  ：删除面 —— dims 少传整层 ⇒ 该层行才删；被删目标的连线端计入 unresolved，不得静默消失。
+# 另：hold 口径断言 total = green + yellow + red + hold（hold 不得二次计入 miss）。
+
+def _row_ids(db: Session, map_id: int) -> dict:
+    """{(层, 名字): 行 id}"""
+    rows = db.query(MapObjective).filter(MapObjective.map_id == map_id).all()
+    return {(r.dimension_key, r.name): r.id for r in rows}
+
+
+def test_save_twice_is_idempotent(client: TestClient, db: Session):
+    """连续两次保存不改名 ⇒ 行 id 集合逐字不变（否则锚点必成野指针）"""
+    create_test_user(db)
+    h = auth_header(get_token_for_user(client))
+    m = _seed(db)
+    _mk_conn(client, h, m.id, "learning:全员培训", "process:交付效率")
+
+    first = _row_ids(db, m.id)
+    dims = copy.deepcopy(m.dimensions)
+    _save_dims(client, h, db, m, dims)          # 第 1 次保存（结构未变）
+    second = _row_ids(db, m.id)
+    _save_dims(client, h, db, m, dims)          # 第 2 次保存
+    third = _row_ids(db, m.id)
+
+    assert first == second, f"首次保存换了行 id: {first} -> {second}"
+    assert second == third, f"二次保存换了行 id: {second} -> {third}"
+    got = _conn(client, h, m.id)
+    assert got["unresolved"] == 0, got["unresolved_items"]
+    assert got["connections"][0]["from_key"] == "learning-0"
+
+
+def test_rename_plus_reorder_without_moid_no_crossbind(client: TestClient, db: Session):
+    """口径 B：同一次保存内「改名 + 同层调序」且不带 _mo_id ⇒ 不得串位
+
+    名字层先把未改名项钉住原行；改名项不再按位置兜底 ⇒ 当新行（旧行删除），
+    连线锚点指向旧行 ⇒ unresolved + stale_objective_id（断线可见，绝不静默错绑）。
+    """
+    create_test_user(db)
+    h = auth_header(get_token_for_user(client))
+    m = _seed(db)
+    _mk_conn(client, h, m.id, "learning:能力认证", "process:交付效率")
+    before = _row_ids(db, m.id)
+    assert _conn(client, h, m.id)["connections"][0]["from_key"] == "learning-1"
+
+    # 调序（能力认证升到第 0 位）+ 改名（能力认证→能力认证A），整表 PUT，不带 _mo_id
+    dims = copy.deepcopy(m.dimensions)
+    dims[3]["objectives"] = [{"name": "能力认证A"}, {"name": "全员培训"}]
+    _save_dims(client, h, db, m, dims)
+
+    after = _row_ids(db, m.id)
+    assert after[("learning", "全员培训")] == before[("learning", "全员培训")], \
+        f"未改名的「全员培训」行 id 变了（串位）: {before} -> {after}"
+    assert after[("learning", "能力认证A")] != before[("learning", "能力认证")], \
+        f"口径 B 下老通道改名应新建行（不做位置兜底）: {before} -> {after}"
+
+    got = _conn(client, h, m.id)
+    assert got["total"] == 1, got
+    c0 = got["connections"][0]
+    assert got["unresolved"] == 1, got["unresolved_items"]
+    assert c0["from"] == "learning:能力认证", "原始串被改写（禁回填）"
+    assert c0["from_key"] is None and c0["from_key_reason"] == "stale_objective_id", c0
+    # 关键：锚点不得被改指到别的节点（旧行为会把线静默挂到「全员培训」上）
+    row = db.query(MapConnection).filter(MapConnection.map_id == m.id).first()
+    assert row.from_objective_id == before[("learning", "能力认证")], "连线锚点被静默改指（错绑）"
+
+
+def test_rename_without_moid_breaks_visibly(client: TestClient, db: Session):
+    """口径 B 单点验证：不带 `_mo_id` 的纯改名 ⇒ 新行 + 连线进 unresolved（断线可见）
+
+    同通道下「删A加B」与「A改名为B」数据上不可区分，位置兜底会让前者静默错绑；
+    放弃兜底的代价就是这里：改名会断线，但断线在接口里可见、可计数。
+    """
+    create_test_user(db)
+    h = auth_header(get_token_for_user(client))
+    m = _seed(db)
+    _mk_conn(client, h, m.id, "learning:全员培训", "process:交付效率")
+    before = _row_ids(db, m.id)
+
+    dims = copy.deepcopy(m.dimensions)
+    assert not any("_mo_id" in o for o in dims[3]["objectives"]), "fixture 不应带 _mo_id（老通道）"
+    dims[3]["objectives"][0]["name"] = "全员培训A"
+    _save_dims(client, h, db, m, dims)
+
+    after = _row_ids(db, m.id)
+    assert after[("learning", "全员培训A")] != before[("learning", "全员培训")], \
+        f"老通道改名竟保住锚点（口径 B 下应为新行）: {before} -> {after}"
+    assert ("learning", "全员培训") not in after, f"旧行未删: {after}"
+
+    got = _conn(client, h, m.id)
+    assert got["total"] == 1, f"连线行被静默丢弃: {got}"
+    c0 = got["connections"][0]
+    assert c0["from"] == "learning:全员培训", "原始串被改写（禁回填）"
+    assert c0["from_key"] is None and c0["from_key_reason"] == "stale_objective_id", c0
+    assert got["unresolved"] == 1 and got["unresolved_items"][0]["reason"] == "stale_objective_id"
+    assert c0["to_key"] == "process-0", "另一端不应受影响"
+    row = db.query(MapConnection).filter(MapConnection.map_id == m.id).first()
+    assert row.from_objective_id == before[("learning", "全员培训")], "连线锚点被静默改指（错绑）"
+
+
+def test_delete_face_unresolved_not_silent(client: TestClient, db: Session):
+    """删除面：dims 少传整层 ⇒ 该层行删；被删目标的连线端计入 unresolved 且原始串保留"""
+    create_test_user(db)
+    h = auth_header(get_token_for_user(client))
+    m = _seed(db)
+    _mk_conn(client, h, m.id, "learning:全员培训", "process:交付效率")
+
+    dims = copy.deepcopy(m.dimensions)
+    dims = dims[:3]                              # 整层去掉 learning
+    _save_dims(client, h, db, m, dims)
+
+    left = db.query(MapObjective).filter(MapObjective.map_id == m.id,
+                                         MapObjective.dimension_key == "learning").all()
+    assert left == [], f"该层已不在 dims，行未删: {[r.id for r in left]}"
+
+    got = _conn(client, h, m.id)
+    assert got["total"] == 1, f"连线被静默丢弃: {got}"      # 行仍在，不消失
+    c0 = got["connections"][0]
+    assert c0["from"] == "learning:全员培训", "原始串被改写（禁回填）"
+    # 锚点指向已删行 ⇒ reason 是更精确的 `stale_objective_id`（区别于「无锚点」的 objective_unresolved）
+    assert c0["from_key"] is None and c0["from_key_reason"] == "stale_objective_id", c0
+    assert got["unresolved"] == 1, got["unresolved_items"]
+    assert got["unresolved_items"][0]["side"] == "from"
+    assert c0["to_key"] == "process-0", "另一端不应受影响"
+
+
+def test_review_hold_total_invariant(client: TestClient, db: Session):
+    """hold 口径：total = green + yellow + red + hold；hold 不得二次计入 miss（红/黄）"""
+    from app.models import KPIDefinition, KPIValue
+    from app.api.maps import DATA_HOLD_KPIS
+    from tests.conftest import create_test_kpi
+
+    create_test_user(db)
+    h = auth_header(get_token_for_user(client))
+    hold_code = next(iter(DATA_HOLD_KPIS))
+    red_code = "P3_TOTAL_INVARIANT_RED"
+    db.query(KPIDefinition).filter(KPIDefinition.kpi_code == red_code).delete(synchronize_session=False)
+    db.commit()
+    red_kpi = create_test_kpi(db, kpi_code=red_code, kpi_name="P3 红项", dimension="customer",
+                              target_value=100.0, kpi_level="strategic", unit="%")
+    db.add(KPIValue(kpi_id=red_kpi.id, period="2026-09", actual_value=10.0, source_type="actual"))
+    db.commit()
+
+    dims = [{"key": "customer", "name": "客户层", "icon": "", "color": "#409EFF", "objectives": [
+        {"name": "待核项", "kpis": [hold_code]},
+        {"name": "红项", "kpis": [red_code]},
+        {"name": "无指标项"},
+    ]}]
+    m = StrategicMap(entity_id=1, title="p3-hold-invariant", status="active",
+                     dimensions=dims, canvas_data={"connections": []})
+    db.add(m)
+    db.commit()
+    db.refresh(m)
+
+    r = client.get(f"/api/cma/maps/{m.id}/review", params={"level": "strategic"}, headers=h)
+    assert r.status_code == 200, r.text
+    s = r.json()["summary"]
+    assert s["total"] == 3, s
+    assert s["hold"] == 1, s
+    assert s["red"] == 1, s
+    assert s["green"] == 1, s
+    assert s["total"] == s["green"] + s["yellow"] + s["red"] + s["hold"], \
+        f"分母不守恒（hold 被二次计入?）: {s}"
+
+    db.query(KPIValue).filter(KPIValue.kpi_id == red_kpi.id).delete(synchronize_session=False)
+    db.query(KPIDefinition).filter(KPIDefinition.kpi_code == red_code).delete(synchronize_session=False)
+    db.commit()
+
+
+
+
+def test_negative_greedy_matching_would_crossbind(client: TestClient, db: Session, monkeypatch):
+    """负测（判据有效性的反证）：把定位算法换回历史「逐项贪心（名→序位→空位）」，
+    则 rename+reorder 场景下两个目标的 id 必然交换 —— 证明
+    test_rename_plus_reorder_without_moid_no_crossbind 有区分度，不是空过。"""
+    from app.api import maps as maps_mod
+
+    def legacy_greedy(m, db_session=None):
+        sess = db_session or db
+        dims = maps_mod._dims_list(m.dimensions)
+        ent = m.entity_id if m.entity_id is not None else 1
+        rows = sess.query(MapObjective).filter(MapObjective.map_id == m.id).order_by(
+            MapObjective.sort_order, MapObjective.id).all()
+        used, keep = set(), []
+        for dim in dims:
+            dk = dim.get("key", "")
+            for i, obj in enumerate(dim.get("objectives") or []):
+                if not isinstance(obj, dict):
+                    continue
+                target = None
+                want = maps_mod._norm_obj_name(obj.get("name"))
+                for r in rows:                                   # ① 名字
+                    if r.id in used or r.dimension_key != dk:
+                        continue
+                    if maps_mod._norm_obj_name(r.name) == want:
+                        target = r
+                        break
+                if target is None:
+                    for r in rows:                               # ② 同层同序位
+                        if r.id in used or r.dimension_key != dk:
+                            continue
+                        if (r.sort_order or 0) == i:
+                            target = r
+                            break
+                if target is None:
+                    for r in rows:                               # ③ 同层空位
+                        if r.id in used or r.dimension_key != dk:
+                            continue
+                        target = r
+                        break
+                if target is None:
+                    target = MapObjective(map_id=m.id, entity_id=ent, dimension_key=dk or "finance")
+                    sess.add(target)
+                else:
+                    used.add(target.id)
+                target.entity_id = ent
+                target.dimension_key = dk
+                target.name = (obj.get("name") or "")[:200]
+                target.icon = obj.get("icon") or "target"
+                target.kpis = obj.get("kpis") or []
+                target.sort_order = i
+                keep.append(target)
+        sess.flush()
+        for r in rows:
+            if r.id not in used:
+                sess.delete(r)
+        sess.commit()
+
+    create_test_user(db)
+    h = auth_header(get_token_for_user(client))
+    m = _seed(db)
+    before = _row_ids(db, m.id)
+
+    monkeypatch.setattr(maps_mod, "_sync_map_objectives", legacy_greedy)
+    dims = copy.deepcopy(m.dimensions)
+    dims[3]["objectives"] = [{"name": "能力认证A"}, {"name": "全员培训"}]
+    _save_dims(client, h, db, m, dims)
+    after = _row_ids(db, m.id)
+
+    assert after[("learning", "全员培训")] != before[("learning", "全员培训")], \
+        f"旧算法竟未串位，则该负测无法证明新判据有效: {before} -> {after}"
+    assert after[("learning", "能力认证A")] == before[("learning", "全员培训")], \
+        f"预期旧算法把「能力认证A」绑到「全员培训」的旧行 id（串位）: {before} -> {after}"