huskies: merge 1038

server/src/service/event_triggers/mod.rs

@@ -0,0 +1,174 @@
+//! Event-based pipeline triggers: register conditions on [`TransitionFired`] events
+//! and automatically execute MCP tool calls or agent prompts when they fire.
+
+/// Persistent storage for event triggers backed by a JSON file.
+pub mod store;
+
+use crate::pipeline_state::{TransitionFired, event_label, stage_label};
+use chrono::{DateTime, Utc};
+use serde::{Deserialize, Serialize};
+
+/// Predicate for matching a [`TransitionFired`] event.
+///
+/// Every `Some` field must match for the predicate to pass; `None` fields
+/// are wildcards.  All comparisons are case-insensitive.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct TriggerPredicate {
+    /// Match only transitions for this story (e.g. `"42_my_feature"`).
+    pub story_id: Option<String>,
+    /// Match only when `fired.before` has this label (e.g. `"Merge"`, `"Coding"`).
+    pub from_stage: Option<String>,
+    /// Match only when `fired.after` has this label (e.g. `"Done"`, `"MergeFailure"`).
+    pub to_stage: Option<String>,
+    /// Match only when the event has this label (e.g. `"MergeFailed"`, `"Block"`).
+    pub event_kind: Option<String>,
+}
+
+impl TriggerPredicate {
+    /// Returns `true` if every non-`None` field matches `fired`.
+    pub fn matches(&self, fired: &TransitionFired) -> bool {
+        if let Some(sid) = &self.story_id
+            && !fired.story_id.0.eq_ignore_ascii_case(sid)
+        {
+            return false;
+        }
+        if let Some(from) = &self.from_stage
+            && !stage_label(&fired.before).eq_ignore_ascii_case(from)
+        {
+            return false;
+        }
+        if let Some(to) = &self.to_stage
+            && !stage_label(&fired.after).eq_ignore_ascii_case(to)
+        {
+            return false;
+        }
+        if let Some(kind) = &self.event_kind
+            && !event_label(&fired.event).eq_ignore_ascii_case(kind)
+        {
+            return false;
+        }
+        true
+    }
+}
+
+/// Action to execute when a trigger fires.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(tag = "type", rename_all = "snake_case")]
+pub enum TriggerAction {
+    /// Call an MCP tool deterministically (no LLM in the loop).
+    Mcp {
+        method: String,
+        args: serde_json::Value,
+    },
+    /// Spawn a short-lived focused agent with this text as its task prompt.
+    Prompt { text: String },
+}
+
+/// Whether a trigger fires once then is removed, or fires repeatedly until cancelled.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum FireMode {
+    /// Remove the trigger after its first match.
+    Once,
+    /// Keep the trigger active until explicitly cancelled.
+    Persistent,
+}
+
+/// A registered event trigger.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct EventTrigger {
+    /// Unique identifier (UUIDv4).
+    pub id: String,
+    /// Predicate applied to every [`TransitionFired`].
+    pub predicate: TriggerPredicate,
+    /// Action to execute on match.
+    pub action: TriggerAction,
+    /// Whether this trigger fires once or persists.
+    pub mode: FireMode,
+    /// When the trigger was registered.
+    pub created_at: DateTime<Utc>,
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::pipeline_state::{PipelineEvent, Stage, StoryId, TransitionFired};
+
+    fn fired(story: &str, before: Stage, after: Stage, event: PipelineEvent) -> TransitionFired {
+        TransitionFired {
+            story_id: StoryId(story.to_string()),
+            before,
+            after,
+            event,
+            at: Utc::now(),
+        }
+    }
+
+    #[test]
+    fn predicate_wildcard_matches_anything() {
+        let p = TriggerPredicate {
+            story_id: None,
+            from_stage: None,
+            to_stage: None,
+            event_kind: None,
+        };
+        let f = fired("42_foo", Stage::Backlog, Stage::Qa, PipelineEvent::DepsMet);
+        assert!(p.matches(&f));
+    }
+
+    #[test]
+    fn predicate_story_id_filter() {
+        let p = TriggerPredicate {
+            story_id: Some("42_foo".to_string()),
+            from_stage: None,
+            to_stage: None,
+            event_kind: None,
+        };
+        let yes = fired("42_foo", Stage::Backlog, Stage::Qa, PipelineEvent::DepsMet);
+        let no = fired("99_bar", Stage::Backlog, Stage::Qa, PipelineEvent::DepsMet);
+        assert!(p.matches(&yes));
+        assert!(!p.matches(&no));
+    }
+
+    #[test]
+    fn predicate_to_stage_filter() {
+        let p = TriggerPredicate {
+            story_id: None,
+            from_stage: None,
+            to_stage: Some("Done".to_string()),
+            event_kind: None,
+        };
+        let yes = fired(
+            "1",
+            Stage::Merge {
+                feature_branch: crate::pipeline_state::BranchName("b".into()),
+                commits_ahead: std::num::NonZeroU32::new(1).unwrap(),
+                claim: None,
+                retries: 0,
+                server_start_time: None,
+            },
+            Stage::Done {
+                merged_at: Utc::now(),
+                merge_commit: crate::pipeline_state::GitSha("abc".into()),
+            },
+            PipelineEvent::MergeSucceeded {
+                merge_commit: crate::pipeline_state::GitSha("abc".into()),
+            },
+        );
+        let no = fired("1", Stage::Backlog, Stage::Qa, PipelineEvent::DepsMet);
+        assert!(p.matches(&yes));
+        assert!(!p.matches(&no));
+    }
+
+    #[test]
+    fn predicate_event_kind_case_insensitive() {
+        let p = TriggerPredicate {
+            story_id: None,
+            from_stage: None,
+            to_stage: None,
+            event_kind: Some("depsmet".to_string()),
+        };
+        let f = fired("1", Stage::Backlog, Stage::Qa, PipelineEvent::DepsMet);
+        assert!(p.matches(&f));
+    }
+}

server/src/service/event_triggers/store.rs

@@ -0,0 +1,332 @@
+//! Persistent store for registered event triggers, backed by a JSON file.
+//!
+//! Loaded at server startup and kept in sync on every mutation.  Thread-safe
+//! via an internal `Mutex`.
+
+use std::path::{Path, PathBuf};
+use std::sync::Mutex;
+
+use chrono::Utc;
+use serde_json::Value;
+
+use super::{EventTrigger, FireMode, TriggerAction, TriggerPredicate};
+
+/// Persistent store for [`EventTrigger`] entries.
+pub struct EventTriggerStore {
+    path: PathBuf,
+    triggers: Mutex<Vec<EventTrigger>>,
+}
+
+impl EventTriggerStore {
+    /// Load the store from `path`.  Returns an empty store if the file does
+    /// not exist or cannot be parsed.
+    pub fn load(path: PathBuf) -> Self {
+        let triggers = if path.exists() {
+            std::fs::read_to_string(&path)
+                .ok()
+                .and_then(|s| serde_json::from_str::<Vec<EventTrigger>>(&s).ok())
+                .unwrap_or_default()
+        } else {
+            Vec::new()
+        };
+        Self {
+            path,
+            triggers: Mutex::new(triggers),
+        }
+    }
+
+    fn persist(path: &Path, triggers: &[EventTrigger]) -> Result<(), String> {
+        if let Some(parent) = path.parent() {
+            std::fs::create_dir_all(parent)
+                .map_err(|e| format!("Failed to create directory: {e}"))?;
+        }
+        let json = serde_json::to_string_pretty(triggers)
+            .map_err(|e| format!("Serialization error: {e}"))?;
+        std::fs::write(path, json).map_err(|e| format!("Failed to write triggers: {e}"))
+    }
+
+    /// Register a new trigger and persist to disk.
+    pub fn add(
+        &self,
+        predicate: TriggerPredicate,
+        action: TriggerAction,
+        mode: FireMode,
+    ) -> Result<EventTrigger, String> {
+        let trigger = EventTrigger {
+            id: uuid_v4(),
+            predicate,
+            action,
+            mode,
+            created_at: Utc::now(),
+        };
+        let mut triggers = self.triggers.lock().unwrap();
+        triggers.push(trigger.clone());
+        Self::persist(&self.path, &triggers)?;
+        Ok(trigger)
+    }
+
+    /// Return a snapshot of all registered triggers.
+    pub fn list(&self) -> Vec<EventTrigger> {
+        self.triggers.lock().unwrap().clone()
+    }
+
+    /// Remove the trigger with `id`.  Returns `true` if it was found and removed.
+    pub fn cancel(&self, id: &str) -> bool {
+        let mut triggers = self.triggers.lock().unwrap();
+        let before = triggers.len();
+        triggers.retain(|t| t.id != id);
+        let removed = triggers.len() < before;
+        if removed {
+            let _ = Self::persist(&self.path, &triggers);
+        }
+        removed
+    }
+
+    /// Remove all triggers whose ids are in `ids` and return how many were removed.
+    ///
+    /// Used by the subscriber to delete `Once` triggers after they fire.
+    pub fn cancel_batch(&self, ids: &[String]) -> usize {
+        if ids.is_empty() {
+            return 0;
+        }
+        let mut triggers = self.triggers.lock().unwrap();
+        let before = triggers.len();
+        triggers.retain(|t| !ids.contains(&t.id));
+        let removed = before - triggers.len();
+        if removed > 0 {
+            let _ = Self::persist(&self.path, &triggers);
+        }
+        removed
+    }
+}
+
+/// Generate a random UUIDv4-style identifier without pulling in the full uuid crate.
+///
+/// Uses [`std::time`] entropy mixed with a thread-local counter. Not cryptographically
+/// strong, but unique enough for trigger IDs.
+fn uuid_v4() -> String {
+    use std::time::{SystemTime, UNIX_EPOCH};
+    let nanos = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap_or_default()
+        .subsec_nanos();
+    let hi = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_secs();
+    format!("{hi:016x}-{nanos:08x}-4000-0000-{hi:012x}")
+}
+
+/// Parse a [`TriggerAction`] from the raw JSON supplied in an MCP args object.
+pub fn parse_action(args: &Value) -> Result<TriggerAction, String> {
+    let action_obj = args
+        .get("action")
+        .ok_or("Missing required argument: action")?;
+    let action_type = action_obj
+        .get("type")
+        .and_then(|v| v.as_str())
+        .ok_or("action.type must be a string (\"mcp\" or \"prompt\")")?;
+
+    match action_type {
+        "mcp" => {
+            let method = action_obj
+                .get("method")
+                .and_then(|v| v.as_str())
+                .ok_or("action.method is required for type=mcp")?
+                .to_string();
+            let action_args = action_obj
+                .get("args")
+                .cloned()
+                .unwrap_or(serde_json::Value::Object(Default::default()));
+            Ok(TriggerAction::Mcp {
+                method,
+                args: action_args,
+            })
+        }
+        "prompt" => {
+            let text = action_obj
+                .get("text")
+                .and_then(|v| v.as_str())
+                .ok_or("action.text is required for type=prompt")?
+                .to_string();
+            Ok(TriggerAction::Prompt { text })
+        }
+        other => Err(format!(
+            "Unknown action type '{other}'; expected \"mcp\" or \"prompt\""
+        )),
+    }
+}
+
+/// Parse a [`TriggerPredicate`] from the raw JSON supplied in an MCP args object.
+pub fn parse_predicate(args: &Value) -> Result<TriggerPredicate, String> {
+    let pred = args
+        .get("predicate")
+        .ok_or("Missing required argument: predicate")?;
+    Ok(TriggerPredicate {
+        story_id: pred
+            .get("story_id")
+            .and_then(|v| v.as_str())
+            .map(str::to_string),
+        from_stage: pred
+            .get("from_stage")
+            .and_then(|v| v.as_str())
+            .map(str::to_string),
+        to_stage: pred
+            .get("to_stage")
+            .and_then(|v| v.as_str())
+            .map(str::to_string),
+        event_kind: pred
+            .get("event_kind")
+            .and_then(|v| v.as_str())
+            .map(str::to_string),
+    })
+}
+
+/// Parse a [`FireMode`] from the raw JSON supplied in an MCP args object.
+pub fn parse_mode(args: &Value) -> FireMode {
+    match args.get("mode").and_then(|v| v.as_str()) {
+        Some("persistent") => FireMode::Persistent,
+        _ => FireMode::Once,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tempfile::TempDir;
+
+    fn tmp_store() -> (TempDir, EventTriggerStore) {
+        let dir = TempDir::new().unwrap();
+        let store = EventTriggerStore::load(dir.path().join("triggers.json"));
+        (dir, store)
+    }
+
+    fn basic_pred() -> TriggerPredicate {
+        TriggerPredicate {
+            story_id: None,
+            from_stage: None,
+            to_stage: Some("Done".to_string()),
+            event_kind: None,
+        }
+    }
+
+    fn basic_action() -> TriggerAction {
+        TriggerAction::Mcp {
+            method: "get_pipeline_status".to_string(),
+            args: serde_json::json!({}),
+        }
+    }
+
+    #[test]
+    fn store_empty_on_missing_file() {
+        let (_dir, store) = tmp_store();
+        assert!(store.list().is_empty());
+    }
+
+    #[test]
+    fn store_add_and_list() {
+        let (_dir, store) = tmp_store();
+        let t = store
+            .add(basic_pred(), basic_action(), FireMode::Once)
+            .unwrap();
+        let list = store.list();
+        assert_eq!(list.len(), 1);
+        assert_eq!(list[0].id, t.id);
+        assert!(matches!(list[0].mode, FireMode::Once));
+    }
+
+    #[test]
+    fn store_cancel_removes_entry() {
+        let (_dir, store) = tmp_store();
+        let t = store
+            .add(basic_pred(), basic_action(), FireMode::Persistent)
+            .unwrap();
+        assert!(store.cancel(&t.id));
+        assert!(!store.cancel(&t.id));
+        assert!(store.list().is_empty());
+    }
+
+    #[test]
+    fn store_persists_and_reloads() {
+        let dir = TempDir::new().unwrap();
+        let path = dir.path().join("triggers.json");
+        let id = {
+            let store = EventTriggerStore::load(path.clone());
+            let t = store
+                .add(basic_pred(), basic_action(), FireMode::Persistent)
+                .unwrap();
+            t.id.clone()
+        };
+        let store2 = EventTriggerStore::load(path);
+        let list = store2.list();
+        assert_eq!(list.len(), 1);
+        assert_eq!(list[0].id, id);
+    }
+
+    #[test]
+    fn store_cancel_batch() {
+        let (_dir, store) = tmp_store();
+        let a = store
+            .add(basic_pred(), basic_action(), FireMode::Once)
+            .unwrap();
+        let b = store
+            .add(basic_pred(), basic_action(), FireMode::Once)
+            .unwrap();
+        let removed = store.cancel_batch(&[a.id, b.id]);
+        assert_eq!(removed, 2);
+        assert!(store.list().is_empty());
+    }
+
+    #[test]
+    fn parse_action_mcp() {
+        let args = serde_json::json!({
+            "action": { "type": "mcp", "method": "get_version", "args": {} }
+        });
+        let action = parse_action(&args).unwrap();
+        assert!(matches!(action, TriggerAction::Mcp { .. }));
+    }
+
+    #[test]
+    fn parse_action_prompt() {
+        let args = serde_json::json!({
+            "action": { "type": "prompt", "text": "investigate the merge failure" }
+        });
+        let action = parse_action(&args).unwrap();
+        assert!(matches!(action, TriggerAction::Prompt { .. }));
+    }
+
+    #[test]
+    fn parse_action_unknown_type_errors() {
+        let args = serde_json::json!({ "action": { "type": "webhook" } });
+        assert!(parse_action(&args).is_err());
+    }
+
+    #[test]
+    fn parse_predicate_all_fields() {
+        let args = serde_json::json!({
+            "predicate": {
+                "story_id": "42_foo",
+                "from_stage": "Coding",
+                "to_stage": "Done",
+                "event_kind": "MergeSucceeded"
+            }
+        });
+        let pred = parse_predicate(&args).unwrap();
+        assert_eq!(pred.story_id.as_deref(), Some("42_foo"));
+        assert_eq!(pred.from_stage.as_deref(), Some("Coding"));
+        assert_eq!(pred.to_stage.as_deref(), Some("Done"));
+        assert_eq!(pred.event_kind.as_deref(), Some("MergeSucceeded"));
+    }
+
+    #[test]
+    fn parse_mode_defaults_to_once() {
+        let args = serde_json::json!({});
+        assert_eq!(parse_mode(&args), FireMode::Once);
+    }
+
+    #[test]
+    fn parse_mode_persistent() {
+        let args = serde_json::json!({ "mode": "persistent" });
+        assert_eq!(parse_mode(&args), FireMode::Persistent);
+    }
+}

server/src/service/mod.rs

@@ -15,6 +15,8 @@ pub mod bot_command;
 pub mod common;
 /// Diagnostics — server logs, CRDT dump, and permission management.
 pub mod diagnostics;
+/// Event-based pipeline triggers: register, list, cancel, and execute on TransitionFired events.
+pub mod event_triggers;
 /// Pipeline event buffer for SSE streaming.
 pub mod events;
 /// File I/O — path validation, read, write, and listing.