diff --git a/server/src/pipeline_state/mod.rs b/server/src/pipeline_state/mod.rs index 9d7b4ac1..0136582e 100644 --- a/server/src/pipeline_state/mod.rs +++ b/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::{ diff --git a/server/src/pipeline_state/subscribers.rs b/server/src/pipeline_state/subscribers.rs index d27298d1..5bc208d3 100644 --- a/server/src/pipeline_state/subscribers.rs +++ b/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, +} + +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) { + 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; + } + } + } +} diff --git a/server/src/startup/tick_loop.rs b/server/src/startup/tick_loop.rs index 614c176e..8608afb5 100644 --- a/server/src/startup/tick_loop.rs +++ b/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();