fix: fail closed on divergent human same-source unify

src/localizer/web/review_coordinator.py

@@ -1,16 +1,19 @@
 """Review mutation coordination at the local Dashboard boundary.
 
-The legacy :class:`ReviewService` owns validation/business semantics. This module
-adds only two cross-cutting guarantees that became necessary once Recovery grew:
+The legacy :class:`ReviewService` owns the baseline validation/business semantics.
+This module adds three production-boundary guarantees that need the current TM
+snapshot under one process-local lock:
 
-1. every Dashboard Review mutation shares the same process-local maintenance RLock
-   as task submission / TM maintenance, so ``check -> TM -> decision log`` cannot be
+1. every Dashboard Review mutation shares the same maintenance RLock as task
+   submission / TM maintenance, so ``check -> TM -> decision log`` cannot be
    interleaved by another Dashboard write;
-2. newly appended TM-mutation decisions carry a complete ``after`` row snapshot.
+2. newly appended TM-mutation decisions carry a complete ``after`` row snapshot;
+3. same-source ``unify`` fails closed before overwriting an existing divergent
+   Review-owned human finalization.
 
-It deliberately does not add a second audit store or duplicate commit/unify rules.
-Old ReviewService callers remain compatible; production Dashboard services use the
-coordinated subclass.
+It deliberately does not add a second audit store, override workflow, or duplicate
+commit/unify implementations. Old ReviewService callers remain compatible;
+production Dashboard services use the coordinated subclass.
 """
 from __future__ import annotations
 
@@ -78,7 +81,7 @@ def append(
 
 
 class CoordinatedReviewService(ReviewService):
-    """ReviewService with one explicit mutation critical section.
+    """ReviewService with one explicit production mutation critical section.
 
     ``RLock`` is intentional: ``unify_majorities`` enters the outer guard and then
     calls ``self.commit()``, which re-enters the same lock. The Dashboard passes the
@@ -111,18 +114,47 @@ def _matches_intended_human_write(row, translation: str) -> bool:
         return True
 
     def commit(self, run_id: str, edits: Mapping[str, str], **kwargs):
-        """Serialize commit and compensate the narrow optimistic-race failure.
+        """Serialize Review writes and keep same-source convenience actions conservative.
 
         Under the shared lock, a local Dashboard writer cannot change the Review log
         revision between TM write and log append. If a non-coordinated/external writer
         still causes ``LogRevisionMismatch``, compensation only touches coordinates
         that (a) differ from their captured before-image and (b) still exactly match
-        this commit's intended fixed human-write shape. Anything else fails closed
-        rather than overwriting a possible newer external state.
+        this commit's intended fixed human-write shape. Anything else fails closed.
+
+        ``unify`` has one additional KISS guard: if any target already contains a
+        different local ``human + reviewed + formal`` value, reject the whole unify
+        before writing anything. There is deliberately no override flag in this slice.
         """
         with self._mutation_lock:
             with SQLiteTranslationMemory(self.config.tm.database) as tm:
                 before = tm.rows_for(list(edits))
+
+            if kwargs.get("action") == "unify":
+                divergent = [
+                    identity
+                    for identity, translation in edits.items()
+                    if (row := before.get(identity)) is not None
+                    and row.get("origin") == "human"
+                    and row.get("review_state") == "reviewed"
+                    and bool(row.get("is_formal"))
+                    and row.get("translation") != translation
+                ]
+                if divergent:
+                    labels = []
+                    for identity in divergent[:10]:
+                        row = before[identity]
+                        path = str(row.get("relative_path") or "")
+                        key = str(row.get("logical_key") or identity)
+                        labels.append(f"{path}:{key}" if path else key)
+                    more = "" if len(divergent) <= 10 else f" 等 {len(divergent)} 条"
+                    raise ReviewConflict(
+                        "同源统一会覆盖已有且译文不同的人工定稿；已拒绝整个操作。"
+                        "请逐条确认或保留该语境译法："
+                        + ", ".join(labels)
+                        + more
+                    )
+
             try:
                 return super().commit(run_id, edits, **kwargs)
             except LogRevisionMismatch:

tests/test_review_convergence.py

@@ -14,12 +14,18 @@
 if str(ROOT / "tests") not in sys.path:
     sys.path.insert(0, str(ROOT / "tests"))
 
+from localizer.adapters.storage.sqlite_tm import SQLiteTranslationMemory
 from localizer.application.review_log import ReviewDecisionEvent
 from localizer.web import DashboardServer as ProductDashboardServer
 from localizer.web.collector import DashboardCollector
 from localizer.web.project_history_detail import project_history_coordinates
+from localizer.web.review import ReviewConflict
 from localizer.web.review_coordinator import CoordinatedReviewService
-from localizer.web.review_recovery import project_change_history, safe_revert
+from localizer.web.review_recovery import (
+    project_change_history,
+    recovery_operations,
+    safe_revert,
+)
 from test_review_recovery import _RecoveryCase
 from test_web_review import _Project
 
@@ -54,22 +60,38 @@ def setUp(self) -> None:
         )
 
     def test_new_tm_decisions_capture_after_image_and_use_it_without_run_index(self) -> None:
-        identities, outcome, payload, _operation, rows = self._prepare_bad_unify()
-        decision_id = rows[identities["g2"]]["decision_id"]
+        run_id = self.project.RUN_ID
+        group = next(
+            item
+            for item in self.service.groups(run_id)["groups"]
+            if item["source"] == "Общий текст"
+        )
+        outcome = self.service.unify(
+            run_id,
+            group["group_id"],
+            "这就对了！",
+            reason="安全统一用于 after-image 回归",
+        )
+        payload = recovery_operations(self.service, run_id, action="unify")
+        operation = next(
+            item for item in payload["operations"] if item["audit_id"] == outcome.audit_id
+        )
+        row = next(
+            item for item in operation["coordinates"] if item["logical_key"] == "g2"
+        )
+        decision_id = row["decision_id"]
         event = next(
             item for item in self.service._log().read_all()
             if item.decision_id == decision_id
         )
-        self.assertEqual(
-            "这就对了！", event.after[identities["g2"]]["translation"]
-        )
+        self.assertEqual("这就对了！", event.after[row["stable_identity"]]["translation"])
 
-        index_path = self.service._index_path(self.project.RUN_ID)
+        index_path = self.service._index_path(run_id)
         self.assertIsNotNone(index_path)
         index_path.unlink()
         detail = project_history_coordinates(
             self.service,
-            run_id=self.project.RUN_ID,
+            run_id=run_id,
             action="unify",
             audit_id=outcome.audit_id,
             query="b.mo",
@@ -81,7 +103,7 @@ def test_new_tm_decisions_capture_after_image_and_use_it_without_run_index(self)
 
         result = safe_revert(
             self.service,
-            self.project.RUN_ID,
+            run_id,
             [decision_id],
             reason="after-image recovery",
             expected_log_revision=payload["log_revision"],
@@ -124,6 +146,75 @@ def test_large_project_history_operation_is_summary_only(self) -> None:
         self.assertIsNone(operation["revertible_count"])
 
 
+class DivergentHumanPreventionTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self._temp = tempfile.TemporaryDirectory()
+        self.project = _Project(Path(self._temp.name))
+        self.service = CoordinatedReviewService(
+            self.project.config,
+            output_root=self.project.config.paths.output,
+            workspace_root=self.project.config.paths.workspace,
+        )
+
+    def tearDown(self) -> None:
+        self._temp.cleanup()
+
+    def _rows(self, identities):
+        with SQLiteTranslationMemory(self.project.config.tm.database) as tm:
+            return tm.rows_for(list(identities))
+
+    def test_group_unify_refuses_divergent_existing_human_finalization(self) -> None:
+        run_id = self.project.RUN_ID
+        group = next(
+            item
+            for item in self.service.groups(run_id)["groups"]
+            if item["source"] == "Общий текст"
+        )
+        protected = self.project.identity("g2")
+        self.service.commit(
+            run_id,
+            {protected: "收到！"},
+            reason="无线电语境人工定稿",
+        )
+
+        with self.assertRaises(ReviewConflict) as ctx:
+            self.service.unify(
+                run_id,
+                group["group_id"],
+                "这就对了！",
+                reason="不应覆盖语境译法",
+            )
+        self.assertIn("人工定稿", str(ctx.exception))
+        self.assertIn("b.mo:g2", str(ctx.exception))
+
+        identities = [member["stable_identity"] for member in group["members"]]
+        rows = self._rows(identities)
+        self.assertEqual({protected}, set(rows))
+        self.assertEqual("收到！", rows[protected]["translation"])
+        self.assertEqual(1, len(self.service._log().read_all()))
+
+    def test_majority_bulk_refuses_before_any_partial_write(self) -> None:
+        run_id = self.project.RUN_ID
+        protected = self.project.identity("m5")
+        majority_ids = [self.project.identity(f"m{i}") for i in range(1, 6)]
+        self.service.commit(
+            run_id,
+            {protected: "语境译法"},
+            reason="保留上下文差异",
+        )
+
+        with self.assertRaises(ReviewConflict):
+            self.service.unify_majorities(
+                run_id,
+                reason="多数派便利操作必须尊重人工定稿",
+            )
+
+        rows = self._rows(majority_ids)
+        self.assertEqual({protected}, set(rows))
+        self.assertEqual("语境译法", rows[protected]["translation"])
+        self.assertEqual(1, len(self.service._log().read_all()))
+
+
 class DashboardCompositionTests(unittest.TestCase):
     def setUp(self) -> None:
         self._temp = tempfile.TemporaryDirectory()