huskies: merge 1014

server/src/pipeline_state/subscribers.rs

@@ -1,10 +1,63 @@
-//! Concrete subscriber stubs for the event bus.
+//! Concrete subscriber stubs for the event bus, plus the production audit-log subscriber.
 
 use super::Stage;
 use super::events::{TransitionFired, TransitionSubscriber};
 #[allow(unused_imports)]
 use super::{event_label, stage_dir_name, stage_label};
 
+// ── Audit log subscriber ─────────────────────────────────────────────────────
+
+/// Format a `TransitionFired` event as a structured one-line audit log entry.
+///
+/// Fields are in stable `key=value` order separated by spaces:
+/// `audit ts=<ISO8601> id=<story_id> from=<from_stage> to=<to_stage> event=<event_label>`
+pub fn format_audit_entry(f: &TransitionFired) -> String {
+    format!(
+        "audit ts={} id={} from={} to={} event={}",
+        f.at.to_rfc3339_opts(chrono::SecondsFormat::Secs, true),
+        f.story_id,
+        stage_label(&f.before),
+        stage_label(&f.after),
+        event_label(&f.event),
+    )
+}
+
+/// Subscriber that writes structured one-line audit entries for every stage transition.
+pub struct AuditLogSubscriber;
+
+impl TransitionSubscriber for AuditLogSubscriber {
+    fn name(&self) -> &'static str {
+        "audit-log"
+    }
+
+    fn on_transition(&self, f: &TransitionFired) {
+        crate::slog!("{}", format_audit_entry(f));
+    }
+}
+
+/// Spawn a background task that writes a structured audit log entry for every pipeline transition.
+///
+/// Subscribes to the transition broadcast channel.  Every `TransitionFired` event produces
+/// one line via [`format_audit_entry`] and writes it to the shared log buffer.
+pub fn spawn_audit_log_subscriber() {
+    let sub = AuditLogSubscriber;
+    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!(
+                        "[audit-log] Subscriber lagged, skipped {n} event(s); \
+                         some transitions may be absent from the audit log."
+                    );
+                }
+                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