C5E: Fix replay crash-window (PROPAGATING_TOLARIA/READY + all objects propagated) via formal VERIFYING_TOLARIA path

FRESH CHECKER-VERDICT (deleg_3ea8ed5a): FAIL.
Defekt: rq_c5e.py Replay 'alle Objekte bereits propagated' rief
transition_commit(ST_UPDATING_SEARCH) aus PROPAGATING_TOLARIA/READY direkt auf,
was InvalidTransitionError warf (Transition nicht in _ALLOWED_TRANSITIONS).

Reparatur (invarianten-treu): statt den VERIFYING_TOLARIA-Schritt zu ueberspringen
(wuerde Read-Back/DRIFT-Check verletzen), wird der formale State-Pfad
READY->PROPAGATING->VERIFYING->UPDATING_SEARCH durchlaufen. Nutzt nur bereits
erlaubte Transitions; keine _ALLOWED_TRANSITIONS-Aenderung, kein C5A/C5C-Risiko.
Kein Tolaria-Doppel-Write (prop_calls=0), kein verfrühter Search (search_calls=0).

+ 2 Regressionstests (test_c5e.py): crash-window + ready-edge-case.
Volle Suite: C5A 25/0 + B/C/D/E 170/0 = 195 OK. Guarantees true.

tolaria/c5-sync-service/rq_c5e.py

@@ -421,11 +421,20 @@ class C5EEngine:
             pending = self._pending_objects(commit_sha)
             if not pending:
                 # Nichts mehr zu propagieren -> Tolaria-Phase abgeschlossen.
+                # Deterministisch den FORMALEN State-Pfad zum Suchschritt durchlaufen.
+                # KEIN Doppel-Write, KEIN Ueberspringen des VERIFYING_TOLARIA-Schritts
+                # (Read-Back/DRIFT-Invariante: Search nie vor vollstaendigem Tolaria-PASS).
+                # Nutzt nur bereits erlaubte Transitions (READY->PROPAGATING->
+                # VERIFYING->UPDATING_SEARCH); keine _ALLOWED_TRANSITIONS-Aenderung noetig.
+                if cur == ST_READY:
+                    self.store.transition_commit(commit_sha, ST_PROPAGATING_TOLARIA)
+                if cur in (ST_READY, ST_PROPAGATING_TOLARIA):
+                    self.store.transition_commit(commit_sha, ST_VERIFYING_TOLARIA)
                 self.store.transition_commit(commit_sha, ST_UPDATING_SEARCH)
                 return {"commit_sha": commit_sha, "decision": decision["decision"],
                         "status": ST_UPDATING_SEARCH,
                         "result": {"partial": True, "propagated": 0},
-                        "reason": "Tolaria bereits vollstaendig -> direkt Suchschritt"}
+                        "reason": "Tolaria bereits vollstaendig -> formaler Pfad zum Suchschritt"}
             result = self.propagator.propagate_commit(commit_sha)
             return {"commit_sha": commit_sha, "decision": decision["decision"],
                     "status": result.get("status"), "result": result,

tolaria/c5-sync-service/test_c5e.py

@@ -258,6 +258,46 @@ class TestReplayIdempotency(unittest.TestCase):
         self.assertEqual(prop.calls, 0, "kein Tolaria-Doppel-Write nach fertiger Tolaria-Phase")
         self.assertEqual(search.calls, 1, "Search-Rebuild wird ausgefuehrt")
 
+    def test_replay_crash_window_propagating_tolaria_all_objects_propagated(self):
+        # Crash-Window (§3/§4): alle Objekte bereits nach Tolaria propagiert, aber die
+        # Transition PROPAGATING_TOLARIA -> VERIFYING_TOLARIA ist noch nicht erfolgt.
+        # Replay MUSS deterministisch zum Suchschritt weiterlaufen (kein
+        # InvalidTransitionError, kein Tolaria-Doppel-Write, kein verfrühter Search).
+        # Regression für FRESH CHECKER-VERDICT (rq_c5e.py Zeile 424).
+        add_commit(self.s, "c1", status=ST_PROPAGATING_TOLARIA, sequence=1)
+        add_object(self.s, "c1", "obj1")
+        add_object(self.s, "c1", "obj2")
+        for o in self.s.list_object_changes("c1"):
+            self.s.set_object_progress("c1", o["object_id"], o["operation"], OBJ_PROPAGATED)
+        prop = _FakePropagator()
+        search = _FakeSearchEngine()
+        prop._store = self.s
+        search._store = self.s
+        eng = C5EEngine(self.s, propagator=prop, search_engine=search,
+                        allow_writes=True)
+        r = eng.replay("c1")
+        self.assertEqual(r["status"], ST_UPDATING_SEARCH,
+                         "Replay aus PROPAGATING_TOLARIA (alle propagiert) -> UPDATING_SEARCH")
+        self.assertEqual(prop.calls, 0, "kein Tolaria-Doppel-Write im Crash-Window")
+        # Formal über VERIFYING_TOLARIA gelaufen (Read-Back-Invariante), nicht übersprungen.
+        self.assertEqual(self.s.commit_status("c1"), ST_UPDATING_SEARCH)
+
+    def test_replay_ready_with_all_objects_propagated(self):
+        # Edge-Case: Commit noch in READY, aber Objekte bereits alle propagiert
+        # (unwahrscheinlich, aber deterministisch auflösbar -> UPDATING_SEARCH).
+        add_commit(self.s, "c1", status=ST_READY, sequence=1)
+        add_object(self.s, "c1", "obj1")
+        self.s.set_object_progress("c1", "obj1", OP_CREATE, OBJ_PROPAGATED)
+        prop = _FakePropagator()
+        search = _FakeSearchEngine()
+        prop._store = self.s
+        search._store = self.s
+        eng = C5EEngine(self.s, propagator=prop, search_engine=search,
+                        allow_writes=True)
+        r = eng.replay("c1")
+        self.assertEqual(r["status"], ST_UPDATING_SEARCH)
+        self.assertEqual(prop.calls, 0, "kein Tolaria-Doppel-Write")
+
     def test_replay_after_search_rebuild_no_duplicate(self):
         # Search bereits APPLIED -> kein Downstream-Write, idempotent.
         add_commit(self.s, "c1", status=ST_APPLIED, sequence=1)