huskies: merge 1016

server/src/pipeline_state/events.rs

@@ -36,6 +36,32 @@ pub(super) fn try_broadcast(fired: &TransitionFired) {
     let _ = get_or_init_tx().send(fired.clone());
 }
 
+/// Replay the current CRDT pipeline state as a burst of synthetic
+/// [`TransitionFired`] events at server startup.
+///
+/// Reads every item from the CRDT and broadcasts a self-transition
+/// (`before == after`) for each one so that all existing subscribers
+/// (worktree lifecycle, merge-failure auto-spawn, auto-assign) react
+/// identically to a live event.  This replaces the legacy scan-based
+/// `reconcile_on_startup` path.
+///
+/// Idempotent: a second call produces another burst of events, but every
+/// subscriber already guards against duplicate work (e.g.
+/// `is_story_assigned_for_stage` returns true once an agent is running,
+/// and worktree creation is a no-op when the worktree already exists).
+pub fn replay_current_pipeline_state() {
+    for item in super::read_all_typed() {
+        let fired = TransitionFired {
+            story_id: item.story_id.clone(),
+            before: item.stage.clone(),
+            after: item.stage,
+            event: super::PipelineEvent::DepsMet,
+            at: chrono::Utc::now(),
+        };
+        try_broadcast(&fired);
+    }
+}
+
 /// Fired when a pipeline stage transition completes.
 #[derive(Debug, Clone)]
 pub struct TransitionFired {
@@ -151,4 +177,58 @@ mod tests {
     }
 
     // ── TransitionError Display ─────────────────────────────────────────
+
+    // ── replay_current_pipeline_state ──────────────────────────────────
+
+    /// AC1: replay broadcasts a synthetic event for every item in the CRDT.
+    #[test]
+    fn replay_broadcasts_event_for_crdt_item_in_coding_stage() {
+        crate::crdt_state::init_for_test();
+        crate::db::ensure_content_store();
+
+        let story_id = "9901_replay_coding";
+        crate::db::write_item_with_content(
+            story_id,
+            "2_current",
+            "---\nname: Replay Coding\n---\n",
+            crate::db::ItemMeta::named("Replay Coding"),
+        );
+
+        let mut rx = subscribe_transitions();
+        replay_current_pipeline_state();
+
+        let mut found = false;
+        while let Ok(fired) = rx.try_recv() {
+            if fired.story_id.0 == story_id && matches!(fired.after, Stage::Coding { .. }) {
+                found = true;
+            }
+        }
+        assert!(
+            found,
+            "replay must broadcast a Coding event for a story in 2_current"
+        );
+    }
+
+    /// AC3: calling replay_current_pipeline_state twice fires events both times.
+    ///
+    /// Pool-state idempotency (no duplicate agents) is enforced by subscribers,
+    /// not by the replay function itself.  This test verifies that replay is safe
+    /// to call multiple times without panicking.
+    #[test]
+    fn replay_twice_does_not_panic() {
+        crate::crdt_state::init_for_test();
+        crate::db::ensure_content_store();
+
+        let story_id = "9902_replay_idem";
+        crate::db::write_item_with_content(
+            story_id,
+            "3_qa",
+            "---\nname: Replay QA\n---\n",
+            crate::db::ItemMeta::named("Replay QA"),
+        );
+
+        // Two successive replays must not panic.
+        replay_current_pipeline_state();
+        replay_current_pipeline_state();
+    }
 }

server/src/pipeline_state/mod.rs

@@ -50,7 +50,10 @@ pub use transition::{
 };
 
 #[allow(unused_imports)]
-pub use events::{EventBus, TransitionFired, TransitionSubscriber, subscribe_transitions};
+pub use events::{
+    EventBus, TransitionFired, TransitionSubscriber, replay_current_pipeline_state,
+    subscribe_transitions,
+};
 
 #[allow(unused_imports)]
 pub use projection::ProjectionError;