huskies: merge 1184 bug Stage transitions never reach the gateway relay: StatusEvent::StageTransition is published only in tests

server/src/pipeline_state/mod.rs

@@ -64,7 +64,7 @@ pub use apply::{
 };
 
 pub(crate) use subscribers::reconcile_audit_log;
-pub use subscribers::spawn_audit_log_subscriber;
+pub use subscribers::{spawn_audit_log_subscriber, spawn_status_broadcast_subscriber};
 
 #[allow(unused_imports)]
 pub use subscribers::{

server/src/pipeline_state/subscribers.rs

@@ -4,6 +4,8 @@ use super::Stage;
 use super::events::{TransitionFired, TransitionSubscriber};
 #[allow(unused_imports)]
 use super::{event_label, stage_dir_name, stage_label};
+use crate::service::status::{StatusBroadcaster, StatusEvent};
+use std::sync::Arc;
 
 // ── Audit log subscriber ─────────────────────────────────────────────────────
 
@@ -66,6 +68,58 @@ pub fn spawn_audit_log_subscriber() {
     });
 }
 
+/// Subscriber that publishes a [`StatusEvent::StageTransition`] to a
+/// [`StatusBroadcaster`] for every pipeline transition.
+///
+/// `story_name` is looked up from the CRDT at publish time and defaults to
+/// an empty string when the item has no name set.
+pub struct StatusBroadcastSubscriber {
+    status: Arc<StatusBroadcaster>,
+}
+
+impl TransitionSubscriber for StatusBroadcastSubscriber {
+    fn name(&self) -> &'static str {
+        "status-broadcast"
+    }
+
+    fn on_transition(&self, f: &TransitionFired) {
+        self.status.publish(StatusEvent::StageTransition {
+            story_id: f.story_id.0.clone(),
+            story_name: crate::crdt_state::read_item(&f.story_id.0)
+                .map(|v| v.name().to_string())
+                .unwrap_or_default(),
+            from_stage: stage_dir_name(&f.before).to_string(),
+            to_stage: stage_dir_name(&f.after).to_string(),
+        });
+    }
+}
+
+/// Spawn a background task that publishes a [`StatusEvent::StageTransition`] to
+/// `status` for every pipeline transition.
+///
+/// Subscribes to the transition broadcast channel — the same channel
+/// [`spawn_audit_log_subscriber`] uses — and forwards each fired transition via
+/// [`StatusBroadcastSubscriber::on_transition`] to the given [`StatusBroadcaster`],
+/// which fans it out to the gateway relay and any other status consumer.
+pub fn spawn_status_broadcast_subscriber(status: Arc<StatusBroadcaster>) {
+    let sub = StatusBroadcastSubscriber { status };
+    let mut rx = super::events::subscribe_transitions();
+    tokio::spawn(async move {
+        loop {
+            match rx.recv().await {
+                Ok(fired) => sub.on_transition(&fired),
+                Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
+                    crate::slog_warn!(
+                        "[status-broadcast] Subscriber lagged, skipped {n} event(s); \
+                         some transitions may be missing from the status broadcast."
+                    );
+                }
+                Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
+            }
+        }
+    });
+}
+
 // ── Subscriber stubs (real dispatch uses these as the interface) ─────────────
 //
 // These are ready to wire into the event bus but not yet connected to the
@@ -164,3 +218,126 @@ impl TransitionSubscriber for WebUiBroadcastSubscriber {
         );
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::pipeline_state::PipelineEvent;
+
+    /// Build a `TransitionFired` for a Backlog → Coding transition of `story_id`.
+    fn fired_backlog_to_coding(story_id: &str) -> TransitionFired {
+        TransitionFired {
+            story_id: crate::pipeline_state::StoryId(story_id.to_string()),
+            before: Stage::Backlog,
+            after: Stage::Coding {
+                claim: None,
+                plan: crate::pipeline_state::PlanState::Missing,
+                retries: 0,
+            },
+            event: PipelineEvent::DepsMet,
+            at: chrono::Utc::now(),
+        }
+    }
+
+    /// Calling `on_transition` directly must publish a matching
+    /// `StatusEvent::StageTransition` on the given `StatusBroadcaster`, with
+    /// `story_name` populated from the CRDT.
+    ///
+    /// This drives the subscriber directly (rather than through
+    /// `spawn_status_broadcast_subscriber` + the process-wide transition
+    /// broadcast channel) so the test is deterministic: the global channel is
+    /// shared by every test in the binary, and unrelated concurrently-running
+    /// tests firing their own transitions would otherwise race with this one.
+    #[tokio::test]
+    async fn on_transition_publishes_stage_transition_with_story_name() {
+        crate::db::ensure_content_store();
+
+        let story_id = "99184_story_status_broadcast";
+        crate::db::write_item_with_content(
+            story_id,
+            "1_backlog",
+            "---\nname: Status Broadcast Test\n---\n# Story\n",
+            crate::db::ItemMeta::named("Status Broadcast Test"),
+        );
+
+        let status = Arc::new(StatusBroadcaster::new());
+        let mut sub = status.subscribe();
+        let subscriber = StatusBroadcastSubscriber {
+            status: status.clone(),
+        };
+
+        subscriber.on_transition(&fired_backlog_to_coding(story_id));
+
+        let event = sub.recv().await.expect("channel should not be closed");
+        match event {
+            StatusEvent::StageTransition {
+                story_id: sid,
+                story_name,
+                from_stage,
+                to_stage,
+            } => {
+                assert_eq!(sid, story_id);
+                assert_eq!(story_name, "Status Broadcast Test");
+                assert_eq!(from_stage, "backlog");
+                assert_eq!(to_stage, "coding");
+            }
+            other => panic!("expected StatusEvent::StageTransition, got: {other:?}"),
+        }
+    }
+
+    /// When the CRDT has no item for the transitioning story — e.g. it was
+    /// never written, or was evicted before the subscriber processed the
+    /// event — `story_name` defaults to an empty string.
+    #[tokio::test]
+    async fn on_transition_defaults_story_name_to_empty_for_unknown_story() {
+        let story_id = "99184_story_status_broadcast_noname";
+
+        let status = Arc::new(StatusBroadcaster::new());
+        let mut sub = status.subscribe();
+        let subscriber = StatusBroadcastSubscriber {
+            status: status.clone(),
+        };
+
+        subscriber.on_transition(&fired_backlog_to_coding(story_id));
+
+        let event = sub.recv().await.expect("channel should not be closed");
+        match event {
+            StatusEvent::StageTransition { story_name, .. } => {
+                assert_eq!(story_name, "", "story_name should default to empty string");
+            }
+            other => panic!("expected StatusEvent::StageTransition, got: {other:?}"),
+        }
+    }
+
+    /// `spawn_status_broadcast_subscriber` wires a `StatusBroadcastSubscriber`
+    /// to the real transition broadcast channel: firing a transition through
+    /// `super::events::try_broadcast` must reach the given `StatusBroadcaster`.
+    /// Uses a story_id unique enough that no other concurrently-running test
+    /// will coincidentally publish a matching one.
+    #[tokio::test]
+    async fn spawn_status_broadcast_subscriber_wires_the_real_channel() {
+        let story_id = "99184_story_status_broadcast_spawn_wiring_check";
+
+        let status = Arc::new(StatusBroadcaster::new());
+        let mut sub = status.subscribe();
+        spawn_status_broadcast_subscriber(status.clone());
+
+        crate::pipeline_state::events::try_broadcast(&fired_backlog_to_coding(story_id));
+
+        let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(2);
+        loop {
+            let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
+            let event = tokio::time::timeout(remaining, sub.recv())
+                .await
+                .expect("timed out waiting for spawned subscriber to forward the event")
+                .expect("channel should not be closed");
+            // Ignore unrelated transitions fired by other concurrently-running
+            // tests on the shared global broadcast channel.
+            if let StatusEvent::StageTransition { story_id: sid, .. } = &event
+                && sid == story_id
+            {
+                break;
+            }
+        }
+    }
+}

server/src/startup/tick_loop.rs

@@ -28,6 +28,12 @@ pub(crate) fn spawn_event_bridges(
     // Audit log subscriber: write one structured line per pipeline transition.
     crate::pipeline_state::spawn_audit_log_subscriber();
 
+    // Status-broadcast subscriber: publish every stage transition to the
+    // project's StatusBroadcaster so the gateway relay (and any other
+    // StatusBroadcaster consumer) sees real pipeline transitions, not just
+    // test-injected ones.
+    crate::pipeline_state::spawn_status_broadcast_subscriber(agents.status_broadcaster());
+
     // Pipeline event bus: initialise before the event-log subscriber so that
     // real-time broadcasts are ready before the first transition fires.
     crate::pipeline_event_bus::init();