fix(runtime): keep candidate feedback topology stable

plugin.py

@@ -93,9 +93,9 @@ async def apply(ctx: Context, config: object) -> None:
         ),
     )
 
-    # 2. Candidate Roots expose read-only projections without formal side effects.
+    # 2. Candidate Root 保留同一 listener 拓扑，但不启动持久 worker。
+    await ctx.on(AFTER_TURN_COMMITTED, runtime.observe_committed)
     if session_read.formal:
-        await ctx.on(AFTER_TURN_COMMITTED, runtime.enqueue)
         await ctx.spawn(runtime.run_worker(), name="proactive_feedback_worker")
     await ui_slots.register_mobile(
         ctx,
@@ -134,6 +134,13 @@ def __init__(
         self._embedder: Embedder | None = None
         self._discovery_done = False
 
+    def observe_committed(self, event: TurnCommitted) -> None:
+        """忽略候选事件，或入队一个正式提交的 Turn。"""
+
+        if not self._session_read.formal:
+            return
+        self.enqueue(event)
+
     def enqueue(self, event: TurnCommitted) -> None:
         """Durably record one committed Turn identity and wake the worker."""
 

tests/test_plugin.py

@@ -1081,11 +1081,18 @@ async def test_manager_stable_candidate_ui_dashboard_and_cleanup(tmp_path: Path)
         assert not (binding.runtime_data_root / "proactive_feedback.db").exists()
         candidate_root = candidate_snapshot.composition_root
         assert candidate_root is not None
+        assert candidate_snapshot.composition_topology is not None
+        assert stable.composition_topology is not None
+        assert (
+            candidate_snapshot.composition_topology.identity
+            == stable.composition_topology.identity
+        )
         candidate_root.context.emit(AFTER_TURN_COMMITTED, _event())
         assert not (binding.runtime_data_root / "proactive_feedback.db").exists()
         assert hashlib.sha256(formal_database.read_bytes()).hexdigest() == formal_digest
-        await manager.discard_prepared("proactive_feedback")
-        assert manager.current_snapshot is stable
+        result = await manager.publish_prepared("proactive_feedback")
+        assert result["publication_state"] == "committed"
+        assert manager.current_snapshot is not stable
     finally:
         await manager.terminate_all()
     receipt = stable.composition_root.receipt()