From 6fca7f076ae39d99668558fa21ffec45b33a7629 Mon Sep 17 00:00:00 2001 From: allen-munsch-bot Date: Mon, 24 Aug 2026 18:34:30 -0500 Subject: [PATCH] vigil: Janet plugin bridge and hooks (slice 4 of 5) Expose vigil/* Janet functions (live?, emit, list, get, set-state) to plugins via a process-global OnceLock bridge in the plugin worker, and register the three vigil lifecycle hooks (on-vigil-event, on-vigil-reap, on-vigil-observance) in the loader. The keeper's plugin event channel is installed into the bridge at startup so (vigil/emit ...) reaches the keeper router across the worker/runtime thread boundary. --- src/main.rs | 11 ++ src/plugin/loader.rs | 4 + src/plugin/worker.rs | 435 +++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 450 insertions(+) diff --git a/src/main.rs b/src/main.rs index 9caf8036e..762c5d410 100644 --- a/src/main.rs +++ b/src/main.rs @@ -2138,6 +2138,17 @@ async fn main() -> anyhow::Result<()> { (None, None, None, None, None) } else { let n = keeper.vigils.len(); + #[cfg(feature = "plugin")] + { + if let Some(ref vig_tx) = keeper.vigil_plugin_tx { + crate::plugin::worker::vigil_bridge::install_vigil_tx( + vig_tx.clone(), + ); + } + let names: Vec = + keeper.vigils.iter().map(|v| v.name.clone()).collect(); + crate::plugin::worker::vigil_bridge::install_vigil_names(names); + } eprintln!("info: vigil-keeper started with {n} vigil(s)"); let wake = keeper.wake_rx.take(); let obs = keeper.observance_rx.take(); diff --git a/src/plugin/loader.rs b/src/plugin/loader.rs index e6f6a43f4..0db2eaef7 100644 --- a/src/plugin/loader.rs +++ b/src/plugin/loader.rs @@ -37,6 +37,10 @@ pub const HOOK_NAMES: &[&str] = &[ // ctx :messages (JSON); may call harness/set-compact-summary to // supply a summary instead of the LLM summarizer. "on-compact", + // --- vigil hooks (dirge-vigil) --- + "on-vigil-event", + "on-vigil-reap", + "on-vigil-observance", ]; /// Filter an input candidate list to only paths that exist as diff --git a/src/plugin/worker.rs b/src/plugin/worker.rs index 9fc3af00c..34e6846f6 100644 --- a/src/plugin/worker.rs +++ b/src/plugin/worker.rs @@ -1039,6 +1039,83 @@ const HARNESS_TOOL_INIT: &str = r#" (harness/__call-tool name payload)) "#; +/// Vigil Janet prelude — exposes vigil/emit, vigil/list, vigil/set-state, +/// vigil/get, vigil/live? for plugins running inside a vigil-keeper. +#[cfg(all(feature = "plugin", feature = "vigil"))] +const HARNESS_VIGIL_INIT: &str = r#" +(defn vigil/live? + "True when the vigil bridge is active. False on builds without the + vigil feature, and also when vigil is not running — so a true result + guarantees that vigil/emit will actually reach the keeper." + [] + (if-let [entry (get (curenv) 'harness/__vigil-live)] + (truthy? ((entry :value))) + false)) + +(defn- json-escape + "Escape a string for embedding inside a JSON string literal." + [s] + (->> s + (string/replace-all "\\" "\\\\") + (string/replace-all "\"" "\\\"") + (string/replace-all "\n" "\\n") + (string/replace-all "\r" "\\r") + (string/replace-all "\t" "\\t"))) + +(defn- json-encode + "Serialize a Janet value to JSON. Handles strings, numbers, booleans, + nil, indexed arrays, and dictionaries (tables/structs)." + [x] + (cond + (string? x) (string "\"" (json-escape x) "\"") + (number? x) (string x) + (= x true) "true" + (= x false) "false" + (nil? x) "null" + (dictionary? x) + (string "{" + (string/join + (map (fn [[k v]] + (string "\"" k "\":" (json-encode v))) + (pairs x)) + ",") + "}") + (indexed? x) + (string "[" + (string/join (map json-encode x) ",") + "]") + (string x))) + +(defn vigil/emit + "Push an event into the vigil-keeper. `event-name` is a string key; + `data` is a dict or JSON string with event context." + [event-name &opt data] + (when (and (vigil/live?) (string? event-name)) + (let [payload (if data + (if (string? data) data (json-encode data)) + "") + msg (string event-name "\t" payload)] + (harness/__vigil-emit msg)))) + +(defn vigil/list + "Return an array of all active vigil names." + [] + (when (vigil/live?) + (harness/__vigil-list))) + +(defn vigil/set-state + "Set a state key for a named vigil. (vigil/set-state name key value)" + [name key value] + (when (and (vigil/live?) (string? name) (string? key)) + (harness/__vigil-set-state name key (string value)))) + +(defn vigil/get + "Get the state table for a named vigil. (vigil/get name)" + [name] + (when (and (vigil/live?) (string? name)) + (harness/__vigil-get name))) +"#; + /// dirge-l6bf: neuter the Janet escape hatches that can terminate or /// destabilize the HOST process. Every hook / command / tool handler is /// already run inside a Janet `(try ...)` (see `mod.rs`), so an ordinary @@ -1943,6 +2020,30 @@ fn worker_loop( env.add_c_fn(CFunOptions::new(c"__lsp", janet_lsp_cfn).namespace(c"harness")); env.add_c_fn(CFunOptions::new(c"__lsp-live", janet_lsp_live_cfn).namespace(c"harness")); } + // Vigil bridge: expose vigil/live? and vigil/emit C functions + // so Janet plugins can interact with the vigil-keeper at runtime. + #[cfg(all(feature = "plugin", feature = "vigil"))] + { + env.add_c_fn( + CFunOptions::new(c"__vigil-live", vigil_bridge::vigil_live_cfn) + .namespace(c"harness"), + ); + env.add_c_fn( + CFunOptions::new(c"__vigil-emit", vigil_bridge::vigil_emit_cfn) + .namespace(c"harness"), + ); + env.add_c_fn( + CFunOptions::new(c"__vigil-list", vigil_bridge::vigil_list_cfn) + .namespace(c"harness"), + ); + env.add_c_fn( + CFunOptions::new(c"__vigil-get", vigil_bridge::vigil_get_cfn).namespace(c"harness"), + ); + env.add_c_fn( + CFunOptions::new(c"__vigil-set-state", vigil_bridge::vigil_set_state_cfn) + .namespace(c"harness"), + ); + } // Computer-use exec: forwards actions to the sandbox drainer. // The C function reads SANDBOX_EXEC_TX; if the channel wasn't // installed (e.g. --sandbox off), it returns nil gracefully. @@ -1992,6 +2093,15 @@ fn worker_loop( let _ = init_tx.send(Err(format!("harness tool-bridge init failed: {e}"))); return; } + // Vigil Janet prelude — defines (vigil/emit), (vigil/list), + // (vigil/set-state), (vigil/get), (vigil/live?). + #[cfg(all(feature = "plugin", feature = "vigil"))] + { + if let Err(e) = client.run(HARNESS_VIGIL_INIT) { + let _ = init_tx.send(Err(format!("vigil init failed: {e}"))); + return; + } + } // dirge-l6bf: disable host-terminating Janet functions. MUST run after // the harness preludes and before any plugin loads, so plugin code // compiles against the neutered bindings. @@ -3582,6 +3692,216 @@ unsafe fn get_dict_int_array(v: janetrs::lowlevel::Janet, key: &str) -> Option> = + std::sync::OnceLock::new(); + + /// Active vigil names, populated at keeper startup via install_vigil_names. + static VIGIL_NAMES: std::sync::OnceLock> = std::sync::OnceLock::new(); + + /// Per-vigil state map (name → JSON value). Janet code can + /// read/write this via vigil/get and vigil/set-state. Lazy-initialized + /// because `HashMap::new` is not const. + static VIGIL_STATE: std::sync::OnceLock>> = + std::sync::OnceLock::new(); + + fn vigil_state() -> &'static std::sync::Mutex> { + VIGIL_STATE.get_or_init(|| std::sync::Mutex::new(HashMap::new())) + } + + /// Install the vigil bridge sender. Called from the tokio runtime + /// after the vigil-keeper is created. Panics if called twice. + pub fn install_vigil_tx(tx: tokio::sync::mpsc::Sender) { + assert!(VIGIL_TX.set(tx).is_ok(), "vigil bridge already installed"); + } + + /// Install vigil names into the bridge. Called at keeper startup + /// after install_vigil_tx. + pub fn install_vigil_names(names: Vec) { + let _ = VIGIL_NAMES.set(names); + } + + /// Return a Janet boolean — true if the vigil bridge has been installed. + pub unsafe extern "C-unwind" fn vigil_live_cfn( + _argc: i32, + _argv: *mut janetrs::lowlevel::Janet, + ) -> janetrs::lowlevel::Janet { + use janetrs::lowlevel::*; + let live = VIGIL_TX.get().is_some(); + unsafe { janet_wrap_boolean(if live { 1 } else { 0 }) } + } + + /// Push a vigil event from Janet into the keeper. Takes one + /// string argument (the event payload). + pub unsafe extern "C-unwind" fn vigil_emit_cfn( + argc: i32, + argv: *mut janetrs::lowlevel::Janet, + ) -> janetrs::lowlevel::Janet { + use janetrs::lowlevel::*; + if argc < 1 { + return unsafe { janet_wrap_nil() }; + } + let msg = match unsafe { read_string_arg(argv, 0) } { + Some(s) => s, + None => return unsafe { janet_wrap_nil() }, + }; + if let Some(tx) = VIGIL_TX.get() { + let _ = tx.try_send(msg); + } + unsafe { janet_wrap_nil() } + } + + /// (vigil/list) — return a Janet array of all active vigil names. + #[allow(clippy::ptr_offset_with_cast)] + pub unsafe extern "C-unwind" fn vigil_list_cfn( + _argc: i32, + _argv: *mut janetrs::lowlevel::Janet, + ) -> janetrs::lowlevel::Janet { + use janetrs::lowlevel::*; + let names = VIGIL_NAMES.get().map(|v| v.as_slice()).unwrap_or(&[]); + let tup = unsafe { janet_tuple_begin(names.len() as i32) }; + for (i, name) in names.iter().enumerate() { + unsafe { + let c_str = std::ffi::CString::new(name.as_str()).unwrap(); + let s = janet_wrap_string(janet_cstring(c_str.as_ptr())); + *tup.offset(i as isize) = s; + } + } + unsafe { janet_wrap_tuple(janet_tuple_end(tup)) } + } + + /// (vigil/get name) — return a Janet table of state for the named vigil. + /// Returns nil if the vigil is not found. + pub unsafe extern "C-unwind" fn vigil_get_cfn( + argc: i32, + argv: *mut janetrs::lowlevel::Janet, + ) -> janetrs::lowlevel::Janet { + use janetrs::lowlevel::*; + if argc < 1 { + return unsafe { janet_wrap_nil() }; + } + let name = match unsafe { read_string_arg(argv, 0) } { + Some(s) => s, + None => return unsafe { janet_wrap_nil() }, + }; + let state = vigil_state().lock().unwrap(); + if let Some(value) = state.get(&name) { + json_to_janet(value) + } else { + let known = VIGIL_NAMES.get().is_some_and(|n| n.contains(&name)); + if known { + // Vigil exists but has no state yet — return empty table. + let tab = unsafe { janet_table(0) }; + unsafe { janet_wrap_table(tab) } + } else { + unsafe { janet_wrap_nil() } + } + } + } + + /// (vigil/set-state name key value) — set a key-value pair on a vigil's + /// state. Returns the vigil name on success, nil on failure. + pub unsafe extern "C-unwind" fn vigil_set_state_cfn( + argc: i32, + argv: *mut janetrs::lowlevel::Janet, + ) -> janetrs::lowlevel::Janet { + use janetrs::lowlevel::*; + if argc < 3 { + return unsafe { janet_wrap_nil() }; + } + let name = match unsafe { read_string_arg(argv, 0) } { + Some(s) => s, + None => return unsafe { janet_wrap_nil() }, + }; + let key = match unsafe { read_string_arg(argv, 1) } { + Some(s) => s, + None => return unsafe { janet_wrap_nil() }, + }; + let value_str = match unsafe { read_string_arg(argv, 2) } { + Some(s) => s, + None => return unsafe { janet_wrap_nil() }, + }; + // Parse as JSON — if the value looks like JSON, use it; otherwise + // treat it as a raw string. + let value: serde_json::Value = + serde_json::from_str(&value_str).unwrap_or(serde_json::Value::String(value_str)); + let mut state = vigil_state().lock().unwrap(); + let entry = state + .entry(name.clone()) + .or_insert(serde_json::Value::Object(serde_json::Map::new())); + if let serde_json::Value::Object(map) = entry { + map.insert(key, value); + } + drop(state); + let c_str = std::ffi::CString::new(name.as_str()).unwrap(); + unsafe { janet_wrap_string(janet_cstring(c_str.as_ptr())) } + } + + /// Convert a serde_json::Value to a Janet value. + #[allow(clippy::ptr_offset_with_cast)] + pub(super) fn json_to_janet(value: &serde_json::Value) -> janetrs::lowlevel::Janet { + use janetrs::lowlevel::*; + match value { + serde_json::Value::Null => unsafe { janet_wrap_nil() }, + serde_json::Value::Bool(b) => unsafe { janet_wrap_boolean(if *b { 1 } else { 0 }) }, + serde_json::Value::Number(n) => { + if let Some(i) = n.as_i64() { + unsafe { janet_wrap_number(i as f64) } + } else if let Some(f) = n.as_f64() { + unsafe { janet_wrap_number(f) } + } else { + unsafe { janet_wrap_nil() } + } + } + serde_json::Value::String(s) => { + let c_str = std::ffi::CString::new(s.as_str()).unwrap(); + unsafe { janet_wrap_string(janet_cstring(c_str.as_ptr())) } + } + serde_json::Value::Array(arr) => { + let tup = unsafe { janet_tuple_begin(arr.len() as i32) }; + for (i, v) in arr.iter().enumerate() { + unsafe { + *tup.offset(i as isize) = json_to_janet(v); + } + } + unsafe { janet_wrap_tuple(janet_tuple_end(tup)) } + } + serde_json::Value::Object(map) => { + let tab = unsafe { janet_table(map.len() as i32) }; + for (k, v) in map { + let c_str = std::ffi::CString::new(k.as_str()).unwrap(); + let key = unsafe { janet_wrap_string(janet_cstring(c_str.as_ptr())) }; + let val = json_to_janet(v); + unsafe { janet_table_put(tab, key, val) }; + } + unsafe { janet_wrap_table(tab) } + } + } + } +} + #[cfg(all(test, feature = "plugin"))] mod tests { use super::*; @@ -4712,4 +5032,119 @@ mod tests { assert!(r.contains("Привет"), "lost Cyrillic: {r:?}"); helper.join().unwrap(); } + + /// vigil/emit uses json-encode to serialize event data. Verify + /// the Janet dict → JSON round-trip produces valid JSON that + /// serde_json can parse back into structured fields — the keeper + /// path at src/extras/vigil/mod.rs:171. + #[cfg(feature = "vigil")] + #[test] + fn json_encode_produces_valid_json_for_vigil_emit() { + let (mut worker, _dialog_rx, _lsp_rx) = Worker::try_spawn().unwrap(); + let json_str = worker + .eval( + r#"(json-encode {:job "my-pipeline" + :build_number "42" + :url "http://jenkins:8080/job/my-pipeline/42" + :status "FAILURE"})"#, + ) + .unwrap(); + + let parsed: serde_json::Value = + serde_json::from_str(&json_str).expect("json-encode must produce valid JSON"); + + assert_eq!(parsed["job"], "my-pipeline"); + assert_eq!(parsed["build_number"], "42"); + assert_eq!(parsed["url"], "http://jenkins:8080/job/my-pipeline/42"); + assert_eq!(parsed["status"], "FAILURE"); + } + + /// Regression: verify that the full vigil/emit message format + /// (name\tjson) can be split and parsed by the keeper router. + #[cfg(feature = "vigil")] + #[test] + fn vigil_emit_message_format_is_parseable_by_keeper() { + let (mut worker, _dialog_rx, _lsp_rx) = Worker::try_spawn().unwrap(); + + // Simulate what (vigil/emit "jenkins-remediate" {...}) sends + // through harness/__vigil-emit. We can't call vigil/emit directly + // (vigil/live? is false in tests), so we call json-encode and + // format the message manually. + let payload = worker + .eval( + r#"(json-encode {:job "my-pipeline" + :build_number "42" + :url "http://jenkins:8080/job/my-pipeline/42" + :status "FAILURE"})"#, + ) + .unwrap(); + let msg = format!("jenkins-remediate\t{payload}"); + + // Simulate the keeper router (src/extras/vigil/mod.rs:167-173) + let (name, payload_str) = msg.split_once('\t').expect("tab-separated message"); + assert_eq!(name, "jenkins-remediate"); + + let context: serde_json::Value = + serde_json::from_str(payload_str).expect("payload must be valid JSON"); + + assert_eq!(context["job"], "my-pipeline"); + assert_eq!(context["build_number"], "42"); + assert_eq!(context["url"], "http://jenkins:8080/job/my-pipeline/42"); + assert_eq!(context["status"], "FAILURE"); + } + + /// harness/json-decode parses nested JSON into a Janet table keyed by + /// string (not keyword) — the poller fixtures rely on this to read + /// `jobs[0].lastBuild.result` from a Jenkins API response. + #[cfg(feature = "vigil")] + #[test] + fn json_decode_parses_nested_json_with_string_keys() { + let (mut worker, _dialog_rx, _lsp_rx) = Worker::try_spawn().unwrap(); + let r = worker + .eval( + r#"(let [d (harness/json-decode "{\"jobs\":[{\"name\":\"test-pipeline\",\"lastBuild\":{\"number\":1,\"result\":\"FAILURE\"}}]}")] + (let [job (get (get d "jobs") 0)] + (string (get job "name") "|" (get-in job ["lastBuild" "result"]))))"#, + ) + .unwrap(); + assert!(r.contains("test-pipeline"), "got {r:?}"); + assert!(r.contains("FAILURE"), "got {r:?}"); + } + + /// harness/json-decode returns nil (not a panic) for malformed JSON. + #[cfg(feature = "vigil")] + #[test] + fn json_decode_returns_nil_on_invalid_json() { + let (mut worker, _dialog_rx, _lsp_rx) = Worker::try_spawn().unwrap(); + let r = worker.eval(r#"(harness/json-decode "not json")"#).unwrap(); + assert_eq!(r, "nil"); + } + + /// Regression: the vigil bridge sender installed on one thread must be + /// visible to `vigil_emit_cfn` running on the Janet worker thread. The + /// original `thread_local!` bridge wrote VIGIL_TX on the tokio runtime + /// thread and read it on the worker thread, so (vigil/emit ...) was a + /// silent no-op. This pins the cross-thread contract with a real worker. + #[cfg(feature = "vigil")] + #[test] + fn vigil_bridge_delivers_emit_across_threads() { + // Install on the TEST thread; the worker evaluates on its own thread. + let (tx, mut rx) = tokio::sync::mpsc::channel::(16); + vigil_bridge::install_vigil_tx(tx); + + let (mut worker, _dialog_rx, _lsp_rx) = Worker::try_spawn().unwrap(); + let r = worker + .eval(r#"(vigil/emit "test-vigil" {:job "my-pipeline"})"#) + .unwrap(); + assert_eq!(r, "nil"); + + let msg = rx + .blocking_recv() + .expect("vigil/emit must reach the bridge tx across threads"); + let (name, payload) = msg.split_once('\t').expect("name\tpayload"); + assert_eq!(name, "test-vigil"); + let context: serde_json::Value = + serde_json::from_str(payload).expect("payload must be valid JSON"); + assert_eq!(context["job"], "my-pipeline"); + } }