diff --git a/docs/reference/cli.md b/docs/reference/cli.md index cdc1b80..137e746 100644 --- a/docs/reference/cli.md +++ b/docs/reference/cli.md @@ -324,7 +324,52 @@ These messages are handled by the gateway before backend dispatch: | `/clear`, `/new`, `/reset` | Start a fresh backend session for that conversation | | `/stop` | Stop the active request; already queued messages continue in order | | `/help` | Return the available chat commands | +| any command in `[command_hooks]` | Run the configured shell command; stdout is the reply, no backend turn | Starting a fresh session preserves canonical history. Push can seed the new backend session with bounded recent turns from the exact channel-qualified conversation. + +### Command hooks + +`[command_hooks]` maps chat slash commands to deterministic shell commands. +The reply is the command's trimmed stdout, relayed verbatim — no agent turn, +no tokens. A mapped command owns all its forms: message arguments are +appended to the command line as trailing positional parameters (`$1`, `$2`, +... after the command's own arguments), so command-shaped input stays +deterministic and never becomes prompt content. Names are normalized to +lowercase at load and must be ASCII letters, digits, `-` or `_` — names +shadowing built-in commands (`/clear`, `/stop`, `/help`) are rejected: + +```toml +[command_hooks] +# Quick sanity check: send `/ping` in chat, expect `pong` back instantly. +ping = "echo pong" +# A longer report: the typing indicator stays on for the whole run, +# and /report reaches the script as $1. +report = "~/bin/report.sh" +``` + +With this config, `/report` runs `~/bin/report.sh` and `/report agents` runs +`~/bin/report.sh agents`. Built-in commands keep exact matching: `/clear +typo` reaches the backend as a regular message, unchanged. Hooks run with a +timeout and a stdout/stderr cap (output beyond 64 KiB is an error reply); a +failing or timed-out hook replies with a short error and never falls back +to the backend. Unknown slash commands still reach the backend as regular +messages. Hook output is delivered like any gateway reply and is recorded +in canonical history. `/help` lists configured hook commands under +"Custom commands" (sorted). + +Hooks run in the thread's queue like any other message: when the backend is +mid-reply, a hook command waits for that turn to finish and runs afterward — +replies stay in order and never interleave. `/stop` is the exception: it acts +on the in-flight request immediately. + +Hooks receive the message context as environment variables: `PUSH_THREAD`, +`PUSH_BACKEND`, `PUSH_ROW_ID`, and `PUSH_SESSION_ID` (only set when the thread +already has a backend session) — enough for info commands to report on the +current conversation. + +Never put secrets in hook commands: the config is read at startup and hook +commands run with the gateway's permissions **and environment** (including +any tokens set in the service environment). diff --git a/src/config.rs b/src/config.rs index 0d075da..442c98a 100644 --- a/src/config.rs +++ b/src/config.rs @@ -1,6 +1,6 @@ //! Gateway configuration loaded from a TOML file. -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; use std::path::{Component, Path, PathBuf}; use std::time::Duration; @@ -76,6 +76,10 @@ pub struct Config { pub agent: String, #[serde(default)] pub routes: Vec, + /// Gateway-level slash commands: `/name [args...]` runs the mapped shell + /// command and relays its stdout verbatim, without an agent turn. + #[serde(default)] + pub command_hooks: HashMap, /// Canonical root of the single user-owned assistant repository. #[serde(default)] pub assistant_root: String, @@ -243,6 +247,7 @@ impl Config { ], )?; let mut c: Config = value.try_into().context("parse TOML config")?; + validate_command_hooks(&mut c.command_hooks)?; let config_path = std::fs::canonicalize(&expanded_path) .with_context(|| format!("resolve config {expanded_path}"))?; c.db_path = expand_home(&c.db_path); @@ -864,6 +869,43 @@ impl AgentBackend { } } +/// Names of the built-in gateway commands; hooks may not shadow them. +const BUILTIN_COMMANDS: &[&str] = &["clear", "new", "reset", "help", "stop"]; + +/// Normalizes hook names to lowercase and rejects names the dispatcher can +/// never reach: empty, whitespace or path separators, or shadowing a built-in. +/// Dispatch lowercases the incoming command, so an uppercase config key would +/// be accepted but unreachable, and `/help` would advertise it. +fn validate_command_hooks(hooks: &mut HashMap) -> Result<()> { + let mut normalized = HashMap::with_capacity(hooks.len()); + for (name, command) in std::mem::take(hooks) { + if command.trim().is_empty() { + bail!("command_hooks.{name}: command must not be empty"); + } + let lowered = name.to_lowercase(); + if lowered.is_empty() + || !lowered + .bytes() + .all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_') + { + bail!( + "command_hooks.{name}: names must be ASCII letters, digits, '-' or '_' (no whitespace or slashes)" + ); + } + if BUILTIN_COMMANDS.contains(&lowered.as_str()) { + bail!("command_hooks.{name}: \"/{lowered}\" is a built-in command and cannot be overridden"); + } + if normalized.contains_key(&lowered) { + bail!( + "command_hooks.{name}: duplicate after case-normalization — \"/{lowered}\" is already defined" + ); + } + normalized.insert(lowered, command); + } + *hooks = normalized; + Ok(()) +} + fn default_db_path() -> String { "~/Library/Messages/chat.db".to_string() } @@ -921,6 +963,7 @@ mod tests { voice_name: DEFAULT_VOICE_NAME.to_string(), agent: "codex".to_string(), routes: Vec::new(), + command_hooks: HashMap::new(), assistant_root: root.to_string_lossy().to_string(), jobs_dir: root.join("jobs").to_string_lossy().to_string(), jobs_agent: None, @@ -1048,4 +1091,33 @@ mod tests { let _ = std::fs::remove_dir_all(assistant); let _ = std::fs::remove_dir_all(outside); } + + #[test] + fn command_hooks_normalize_names_and_reject_collisions() { + let mut hooks = HashMap::new(); + hooks.insert("Status".to_string(), "echo ok".to_string()); + validate_command_hooks(&mut hooks).unwrap(); + assert!(hooks.contains_key("status")); + assert!(!hooks.contains_key("Status")); + + let mut hooks = HashMap::new(); + hooks.insert("Report".to_string(), "echo a".to_string()); + hooks.insert("report".to_string(), "echo b".to_string()); + let error = validate_command_hooks(&mut hooks).unwrap_err(); + assert!(error + .to_string() + .contains("duplicate after case-normalization")); + } + + #[test] + fn command_hooks_reject_whitespace_names_and_builtins() { + let mut hooks = HashMap::new(); + hooks.insert("bad name".to_string(), "echo hi".to_string()); + assert!(validate_command_hooks(&mut hooks).is_err()); + + let mut hooks = HashMap::new(); + hooks.insert("clear".to_string(), "echo hi".to_string()); + let error = validate_command_hooks(&mut hooks).unwrap_err(); + assert!(error.to_string().contains("built-in")); + } } diff --git a/src/gateway/tests.rs b/src/gateway/tests.rs index c7a0bd0..6e2541c 100644 --- a/src/gateway/tests.rs +++ b/src/gateway/tests.rs @@ -1600,6 +1600,302 @@ async fn telegram_filters_before_agent_and_replies_to_originating_chat() { let _ = std::fs::remove_dir_all(assistant_dir); } +#[tokio::test(flavor = "current_thread")] +async fn command_hook_runs_and_relays_stdout_without_agent() { + let state_path = temp_state_path(); + let sessions_dir = temp_path("command-hook-sessions"); + let assistant_dir = temp_path("command-hook-assistant"); + std::fs::create_dir_all(&assistant_dir).unwrap(); + let calls = Arc::new(Mutex::new(Vec::new())); + let hook_path = temp_path("command-hook-script"); + let script = "#!/bin/sh\necho \"status: ok args=[$*] thread=$PUSH_THREAD backend=$PUSH_BACKEND row=$PUSH_ROW_ID session=${PUSH_SESSION_ID:-none}\"\n"; + std::fs::write(&hook_path, script).unwrap(); + let mut cfg = test_config( + &state_path, + sessions_dir.to_str().unwrap(), + assistant_dir.to_str().unwrap(), + ); + cfg.channel = "telegram".to_string(); + cfg.self_handles.clear(); + cfg.allow_from.clear(); + cfg.telegram_bot_token = Some("secret".to_string()); + cfg.telegram_allow_user_ids = vec![7]; + cfg.command_hooks.insert( + "status".to_string(), + format!("sh {}", hook_path.to_str().unwrap()), + ); + let mut gateway = Gateway::new(cfg).unwrap(); + gateway.ctx.runners = Arc::new(fake_runners(calls.clone())); + + run_messages( + &mut gateway, + vec![ + telegram_message(1, 7, 7, false, "/status"), + // A mapped command owns all its forms: args reach the hook. + telegram_message(2, 7, 7, false, "/status foo bar"), + telegram_message(3, 7, 7, false, "/help"), + ], + ) + .await; + + assert_eq!(calls.lock().unwrap().len(), 0); + let replies = gateway.ctx.sent_replies.lock().unwrap(); + assert_eq!( + &replies[0..2], + [ + ( + "7".to_string(), + "status: ok args=[] thread=telegram:dm:7 backend=codex row=1 session=none" + .to_string() + ), + ( + "7".to_string(), + "status: ok args=[foo bar] thread=telegram:dm:7 backend=codex row=2 session=none" + .to_string() + ) + ] + ); + assert_eq!( + replies[2].1, + "Commands:\n/clear - start a fresh conversation\n/stop - stop the active request\n/help - this message\n\nCustom commands: /status" + ); + + let _ = std::fs::remove_file(&state_path); + let _ = std::fs::remove_file(format!("{state_path}.audit.jsonl")); + let _ = std::fs::remove_file(&hook_path); + let _ = std::fs::remove_dir_all(sessions_dir); + let _ = std::fs::remove_dir_all(assistant_dir); +} + +#[tokio::test(flavor = "current_thread")] +async fn unknown_slash_command_falls_through_to_agent() { + let state_path = temp_state_path(); + let sessions_dir = temp_path("slash-fallthrough-sessions"); + let assistant_dir = temp_path("slash-fallthrough-assistant"); + std::fs::create_dir_all(&assistant_dir).unwrap(); + let calls = Arc::new(Mutex::new(Vec::new())); + let mut cfg = test_config( + &state_path, + sessions_dir.to_str().unwrap(), + assistant_dir.to_str().unwrap(), + ); + cfg.channel = "telegram".to_string(); + cfg.self_handles.clear(); + cfg.allow_from.clear(); + cfg.telegram_bot_token = Some("secret".to_string()); + cfg.telegram_allow_user_ids = vec![7]; + cfg.command_hooks + .insert("status".to_string(), "true".to_string()); + let mut gateway = Gateway::new(cfg).unwrap(); + gateway.ctx.runners = Arc::new(fake_runners(calls.clone())); + + run_messages( + &mut gateway, + vec![telegram_message(1, 7, 7, false, "/bogus details")], + ) + .await; + + let calls = calls.lock().unwrap(); + assert_eq!(calls.len(), 1); + assert_eq!( + crate::prompt::current_message(&calls[0].prompt).as_deref(), + Some("/bogus details") + ); + + let _ = std::fs::remove_file(&state_path); + let _ = std::fs::remove_file(format!("{state_path}.audit.jsonl")); + let _ = std::fs::remove_dir_all(sessions_dir); + let _ = std::fs::remove_dir_all(assistant_dir); +} + +#[tokio::test(flavor = "current_thread")] +async fn command_hook_matches_multiline_command_word() { + let state_path = temp_state_path(); + let sessions_dir = temp_path("command-hook-multiline-sessions"); + let assistant_dir = temp_path("command-hook-multiline-assistant"); + std::fs::create_dir_all(&assistant_dir).unwrap(); + let calls = Arc::new(Mutex::new(Vec::new())); + let hook_path = temp_path("command-hook-multiline-script"); + std::fs::write(&hook_path, "#!/bin/sh\necho \"got: $*\"\n").unwrap(); + let mut cfg = test_config( + &state_path, + sessions_dir.to_str().unwrap(), + assistant_dir.to_str().unwrap(), + ); + cfg.channel = "telegram".to_string(); + cfg.self_handles.clear(); + cfg.allow_from.clear(); + cfg.telegram_bot_token = Some("secret".to_string()); + cfg.telegram_allow_user_ids = vec![7]; + cfg.command_hooks.insert( + "report".to_string(), + format!("sh {}", hook_path.to_str().unwrap()), + ); + let mut gateway = Gateway::new(cfg).unwrap(); + gateway.ctx.runners = Arc::new(fake_runners(calls.clone())); + + // Newline between command word and args must not break hook routing. + run_messages( + &mut gateway, + vec![telegram_message(1, 7, 7, false, "/report\nagents")], + ) + .await; + + assert_eq!(calls.lock().unwrap().len(), 0); + assert_eq!( + gateway.ctx.sent_replies.lock().unwrap().as_slice(), + [("7".to_string(), "got: agents".to_string())] + ); + + let _ = std::fs::remove_file(&state_path); + let _ = std::fs::remove_file(format!("{state_path}.audit.jsonl")); + let _ = std::fs::remove_file(&hook_path); + let _ = std::fs::remove_dir_all(sessions_dir); + let _ = std::fs::remove_dir_all(assistant_dir); +} + +#[tokio::test(flavor = "current_thread")] +async fn builtin_with_arguments_reaches_backend_unchanged() { + let state_path = temp_state_path(); + let sessions_dir = temp_path("builtin-args-sessions"); + let assistant_dir = temp_path("builtin-args-assistant"); + std::fs::create_dir_all(&assistant_dir).unwrap(); + let calls = Arc::new(Mutex::new(Vec::new())); + let mut cfg = test_config( + &state_path, + sessions_dir.to_str().unwrap(), + assistant_dir.to_str().unwrap(), + ); + cfg.channel = "telegram".to_string(); + cfg.self_handles.clear(); + cfg.allow_from.clear(); + cfg.telegram_bot_token = Some("secret".to_string()); + cfg.telegram_allow_user_ids = vec![7]; + let mut gateway = Gateway::new(cfg).unwrap(); + gateway.ctx.runners = Arc::new(fake_runners(calls.clone())); + + run_messages( + &mut gateway, + vec![telegram_message(1, 7, 7, false, "/clear typo")], + ) + .await; + + // Pre-hook behavior: built-ins with trailing arguments are prompt content. + let calls = calls.lock().unwrap(); + assert_eq!(calls.len(), 1); + assert_eq!( + crate::prompt::current_message(&calls[0].prompt).as_deref(), + Some("/clear typo") + ); + + let _ = std::fs::remove_file(&state_path); + let _ = std::fs::remove_file(format!("{state_path}.audit.jsonl")); + let _ = std::fs::remove_dir_all(sessions_dir); + let _ = std::fs::remove_dir_all(assistant_dir); +} + +#[tokio::test(flavor = "current_thread")] +async fn command_hook_output_is_capped_without_panicking() { + let state_path = temp_state_path(); + let sessions_dir = temp_path("command-hook-cap-sessions"); + let assistant_dir = temp_path("command-hook-cap-assistant"); + std::fs::create_dir_all(&assistant_dir).unwrap(); + let calls = Arc::new(Mutex::new(Vec::new())); + let mut cfg = test_config( + &state_path, + sessions_dir.to_str().unwrap(), + assistant_dir.to_str().unwrap(), + ); + cfg.channel = "telegram".to_string(); + cfg.self_handles.clear(); + cfg.allow_from.clear(); + cfg.telegram_bot_token = Some("secret".to_string()); + cfg.telegram_allow_user_ids = vec![7]; + // 40k multibyte characters = 80k bytes: crosses the 64 KiB cap mid-character. + cfg.command_hooks.insert( + "flood".to_string(), + "python3 -c 'print(\"\u{00e9}\" * 40000)'".to_string(), + ); + let mut gateway = Gateway::new(cfg).unwrap(); + gateway.ctx.runners = Arc::new(fake_runners(calls.clone())); + + run_messages( + &mut gateway, + vec![telegram_message(1, 7, 7, false, "/flood")], + ) + .await; + + assert_eq!(calls.lock().unwrap().len(), 0); + let replies = gateway.ctx.sent_replies.lock().unwrap(); + assert_eq!(replies.len(), 1); + assert_eq!(replies[0].1, "Command hook output exceeded 64 KiB."); + + let _ = std::fs::remove_file(&state_path); + let _ = std::fs::remove_file(format!("{state_path}.audit.jsonl")); + let _ = std::fs::remove_dir_all(sessions_dir); + let _ = std::fs::remove_dir_all(assistant_dir); +} + +#[tokio::test(flavor = "current_thread")] +async fn command_hook_session_id_is_backend_scoped() { + let state_path = temp_state_path(); + let sessions_dir = temp_path("command-hook-session-sessions"); + let assistant_dir = temp_path("command-hook-session-assistant"); + std::fs::create_dir_all(&assistant_dir).unwrap(); + let calls = Arc::new(Mutex::new(Vec::new())); + let hook_path = temp_path("command-hook-session-script"); + std::fs::write( + &hook_path, + "#!/bin/sh\necho \"session=${PUSH_SESSION_ID:-none}\"\n", + ) + .unwrap(); + let mut cfg = test_config( + &state_path, + sessions_dir.to_str().unwrap(), + assistant_dir.to_str().unwrap(), + ); + cfg.channel = "telegram".to_string(); + cfg.self_handles.clear(); + cfg.allow_from.clear(); + cfg.telegram_bot_token = Some("secret".to_string()); + cfg.telegram_allow_user_ids = vec![7]; + cfg.command_hooks.insert( + "session".to_string(), + format!("sh {}", hook_path.to_str().unwrap()), + ); + let mut gateway = Gateway::new(cfg).unwrap(); + gateway.ctx.runners = Arc::new(fake_runners(calls.clone())); + + // A session stored for a DIFFERENT backend must not leak to this hook. + // The fake runner routes as codex; the stored session is claude's. + gateway + .ctx + .store + .lock() + .unwrap() + .session_for("telegram:dm:7", "claude", "claude-old-session".to_string()) + .unwrap(); + + run_messages( + &mut gateway, + vec![telegram_message(1, 7, 7, false, "/session")], + ) + .await; + + assert_eq!(calls.lock().unwrap().len(), 0); + let replies = gateway.ctx.sent_replies.lock().unwrap(); + assert_eq!( + replies[0].1, "session=none", + "claude session must not leak into a codex-routed hook" + ); + + let _ = std::fs::remove_file(&state_path); + let _ = std::fs::remove_file(format!("{state_path}.audit.jsonl")); + let _ = std::fs::remove_file(&hook_path); + let _ = std::fs::remove_dir_all(sessions_dir); + let _ = std::fs::remove_dir_all(assistant_dir); +} + #[tokio::test(flavor = "current_thread")] async fn telegram_topic_gets_own_thread_and_reply_targets_the_topic() { let state_path = temp_state_path(); @@ -3974,6 +4270,7 @@ fn test_config(state_path: &str, _sessions_dir: &str, assistant_dir: &str) -> Co voice_name: crate::config::DEFAULT_VOICE_NAME.to_string(), agent: "codex".to_string(), routes: Vec::new(), + command_hooks: HashMap::new(), assistant_root: assistant_dir.to_string(), jobs_dir: format!("{state_path}.jobs"), jobs_agent: None, diff --git a/src/gateway/worker.rs b/src/gateway/worker.rs index ecba8df..98a6de3 100644 --- a/src/gateway/worker.rs +++ b/src/gateway/worker.rs @@ -10,6 +10,8 @@ use anyhow::{Context, Result}; use tokio::sync::{mpsc, watch}; use tracing::{error, info, warn}; +use std::process::Stdio; + use crate::agent::{Request, RunError}; use crate::history::{DeliveryStatus, OutboundMessage, OutboundOrigin}; use crate::image::{PreparedImages, MAX_IMAGE_BYTES, MAX_IMAGE_COUNT}; @@ -103,7 +105,7 @@ where } if job.image_attachments.is_empty() { - if let Some(reply) = command(ctx, &job) { + if let Some(reply) = command(ctx, &job).await { let delivery = record_and_deliver(ctx, &job, OutboundOrigin::Gateway, &reply).await; if delivery.is_ok() { info!( @@ -814,25 +816,262 @@ fn complete_job(ctx: &Ctx, job: &Job, reason: &str) { complete_row(&ctx.store, &ctx.ack, ctx.channel.id(), job.row_id); } +/// Splits `/word args...` input into `(word, args)` for command routing. +/// The word is split on any whitespace, so `/report\nagents` routes to the +/// `report` hook like `/report agents` does. Returns `None` for anything +/// that is not a slash command with a word. +fn slash_command(text: &str) -> Option<(&str, &str)> { + let rest = text.trim().strip_prefix('/')?; + let rest = rest.trim_start(); + if rest.is_empty() { + return None; + } + Some(match rest.find(char::is_whitespace) { + Some(index) => (&rest[..index], rest[index..].trim()), + None => (rest, ""), + }) +} + /// Handles gateway-level slash commands before anything reaches the agent. -fn command(ctx: &Ctx, job: &Job) -> Option { - match job.text.trim().to_lowercase().as_str() { - "/clear" | "/new" | "/reset" => match ctx.store.lock().unwrap().rotate( - &job.thread, - job.backend.as_str(), - ctx.runners - .get(&job.backend) - .map(|r| r.initial_session_id()) - .unwrap_or_default(), - ) { - Ok(()) => Some("Started a fresh conversation.".to_string()), - Err(_) => Some("Couldn't reset the conversation.".to_string()), - }, - "/help" => Some( - "Commands:\n/clear - start a fresh conversation\n/stop - stop the active request\n/help - this message" - .to_string(), - ), - _ => None, +/// Built-ins are hardcoded; `[command_hooks]` maps `/name` to a shell command +/// whose stdout is relayed verbatim (deterministic, no agent turn). Unknown +/// slash commands fall through to the agent as before. +async fn command(ctx: &Ctx, job: &Job) -> Option { + let (word, args) = slash_command(&job.text)?; + let lowered = word.to_lowercase(); + // Built-ins match bare commands only: `/clear typo` reaches the backend + // unchanged, exactly as it did before hooks existed. + if args.is_empty() { + match lowered.as_str() { + "clear" | "new" | "reset" => match ctx.store.lock().unwrap().rotate( + &job.thread, + job.backend.as_str(), + ctx.runners + .get(&job.backend) + .map(|r| r.initial_session_id()) + .unwrap_or_default(), + ) { + Ok(()) => return Some("Started a fresh conversation.".to_string()), + Err(_) => return Some("Couldn't reset the conversation.".to_string()), + }, + "help" => { + let mut text = "Commands:\n/clear - start a fresh conversation\n/stop - stop the active request\n/help - this message".to_string(); + let mut names: Vec<&String> = ctx.cfg.command_hooks.keys().collect(); + names.sort_unstable(); + if !names.is_empty() { + let list = names + .iter() + .map(|n| format!("/{n}")) + .collect::>() + .join(", "); + text.push_str(&format!("\n\nCustom commands: {list}")); + } + return Some(text); + } + _ => {} + } + } + // A mapped command owns all its forms: args are appended to the + // command line as trailing positional parameters (argv-safe, no + // re-parsing). Command-shaped input stays deterministic by + // default — it never becomes prompt content. + let hook = ctx.cfg.command_hooks.get(lowered.as_str())?; + // Hook context env: scripts like /session need to know which + // thread/session/backend asked. + let session_id = ctx + .store + .lock() + .unwrap() + .peek_session_id(&job.thread, job.backend.as_str()); + let env = CommandHookEnv { + thread: job.thread.clone(), + backend: job.backend.as_str().to_string(), + row_id: job.row_id, + session_id, + }; + // Keep the typing indicator alive for the whole hook run — + // status-report-class scripts take 10s+, Telegram's typing + // action expires in ~5s. Same refresh loop the agent run uses. + match ctx.channel.typing_refresh() { + Some(refresh) => { + let channel = ctx.channel.clone(); + let target = job.target.clone(); + let thread = job.thread.clone(); + let log_thread = thread.clone(); + let env2 = env.clone(); + Some( + run_with_periodic_activity( + run_command_hook(hook, args, env2), + refresh, + move || { + let channel = channel.clone(); + let target = target.clone(); + let thread = log_thread.clone(); + async move { + if let Err(e) = channel.send_typing(&target).await { + warn!("[{thread}] typing update failed: {e}"); + } + } + }, + ) + .await, + ) + } + None => Some(run_command_hook(hook, args, env).await), + } +} + +#[derive(Clone)] +struct CommandHookEnv { + thread: String, + backend: String, + row_id: i64, + session_id: Option, +} + +impl CommandHookEnv { + fn envs(&self) -> Vec<(String, String)> { + let mut env = vec![ + ("PUSH_THREAD".to_string(), self.thread.clone()), + ("PUSH_BACKEND".to_string(), self.backend.clone()), + ("PUSH_ROW_ID".to_string(), self.row_id.to_string()), + ]; + if let Some(session_id) = &self.session_id { + env.push(("PUSH_SESSION_ID".to_string(), session_id.clone())); + } + env + } +} + +/// Runs a command hook through `/bin/sh`. The trimmed stdout is the reply; +/// failures reply with a short deterministic error instead of falling back to +/// the agent, so a broken hook never turns into a surprise LLM turn. +/// The hook runs in its own process group and every timeout/cap path signals +/// the whole group, so backgrounded descendants die with the shell instead +/// of leaking. stdout/stderr are read with byte caps — a runaway hook is +/// bounded memory and a deterministic reply, never an OOM. +async fn run_command_hook(hook: &str, args: &str, env: CommandHookEnv) -> String { + // `sh -c "{hook} \"$@\"" sh `: the hook's own fixed args + // come first, message args are appended to the same command line (argv, + // never re-parsed by the shell). The bare `sh` is $0 for the -c string. + let child = tokio::process::Command::new("/bin/sh") + .arg("-c") + .arg(format!("{hook} \"$@\"")) + .arg("sh") + .args(args.split_whitespace()) + .envs(env.envs()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .kill_on_drop(true) + .process_group(0) + .spawn(); + let mut child = match child { + Ok(child) => child, + Err(error) => return format!("Command hook failed to run: {error}"), + }; + // Leader pid == process group id (process_group(0)); captured before the + // timeout scope so the expiry path can still signal the group after the + // future (and the Child) is dropped. + let pgid = child.id().expect("hook child alive before wait") as libc::pid_t; + let mut stdout_pipe = child.stdout.take().expect("hook stdout piped"); + let mut stderr_pipe = child.stderr.take().expect("hook stderr piped"); + let collect = async { + let (stdout, stderr) = + tokio::join!(read_capped(&mut stdout_pipe), read_capped(&mut stderr_pipe),); + let status = child.wait().await; + (stdout, stderr, status) + }; + let result = + tokio::time::timeout(Duration::from_secs(COMMAND_HOOK_TIMEOUT_SECS), collect).await; + match result { + Err(_) => { + // kill_on_drop reaped the leader when the inner future dropped; + // the group signal takes care of any descendants it spawned. + kill_hook_group(pgid); + "Command hook timed out.".to_string() + } + Ok(((stdout, out_capped, out_error), (stderr, err_capped, err_error), status)) => { + if out_capped || err_capped { + // Stop the hook (and its descendants) so the pipes close; + // partial output beyond the cap is not delivered. + kill_hook_group(pgid); + return "Command hook output exceeded 64 KiB.".to_string(); + } + if let Some(error) = out_error.or(err_error) { + warn!("[{}] command hook pipe read failed: {error}", env.thread); + return "Command hook failed: output could not be read.".to_string(); + } + match status { + Err(error) => format!("Command hook failed to run: {error}"), + Ok(status) if status.success() => { + let stdout = String::from_utf8_lossy(&stdout).trim().to_string(); + if stdout.is_empty() { + "Command hook produced no output.".to_string() + } else { + stdout + } + } + Ok(status) => { + let stderr = String::from_utf8_lossy(&stderr).trim().to_string(); + let stderr = truncate_chars(&stderr, 500); + if stderr.is_empty() { + format!("Command hook exited with {status}") + } else { + format!("Command hook exited with {status}: {stderr}") + } + } + } + } + } +} + +const COMMAND_HOOK_TIMEOUT_SECS: u64 = 15; +const COMMAND_HOOK_MAX_STDOUT: usize = 64 * 1024; // same cap the timeout hook applies + +/// SIGKILLs the hook's whole process group (negative pid). Covers the shell +/// leader and any backgrounded descendants — kill_on_drop only ever signals +/// the leader. Errors are ignored: the group may already be gone. +fn kill_hook_group(pgid: libc::pid_t) { + // Safety: signal syscall with an integer pid; no pointers involved. + unsafe { + libc::kill(-pgid, libc::SIGKILL); + } +} + +/// Reads a pipe with a byte cap, returning the bytes, whether the cap was +/// exceeded, and the first read error (if any). A read error is not EOF: +/// partial output from a failing pipe is a hook failure. +async fn read_capped( + reader: &mut R, +) -> (Vec, bool, Option) { + use tokio::io::AsyncReadExt; + let mut buf = Vec::new(); + let mut chunk = [0u8; 8192]; + let mut capped = false; + let mut read_error = None; + loop { + match reader.read(&mut chunk).await { + Ok(0) => break, + Ok(n) => { + if buf.len() + n > COMMAND_HOOK_MAX_STDOUT { + capped = true; + break; + } + buf.extend_from_slice(&chunk[..n]); + } + Err(error) => { + read_error = Some(error); + break; + } + } + } + (buf, capped, read_error) +} + +fn truncate_chars(text: &str, max: usize) -> &str { + match text.char_indices().nth(max) { + Some((index, _)) => &text[..index], + None => text, } } diff --git a/src/store.rs b/src/store.rs index 18ee7bf..9629f64 100644 --- a/src/store.rs +++ b/src/store.rs @@ -326,6 +326,24 @@ impl Store { Ok(()) } + /// Read-only session lookup for info commands: the stored session id for + /// a thread on the given backend, or None when the thread has no session + /// for that backend yet. Unlike `session_for`, never creates a row. + /// Backend-scoped on purpose: after a route change, the stored session + /// belongs to the previous backend and must not leak to hooks. + pub fn peek_session_id(&self, thread: &str, backend: &str) -> Option { + let (channel, thread_key) = split_thread(thread).ok()?; + self.conn + .query_row( + "SELECT session_id FROM backend_sessions + WHERE channel = ?1 AND thread_key = ?2 AND backend = ?3", + params![channel, thread_key, backend], + |row| row.get::<_, String>(0), + ) + .ok() + .filter(|id| !id.trim().is_empty()) + } + /// Returns the agent session id for a thread, creating one if needed. The /// second value is true when the backend has not started that session yet. pub fn session_for( diff --git a/src/test_support.rs b/src/test_support.rs index 9d8d601..89e61c5 100644 --- a/src/test_support.rs +++ b/src/test_support.rs @@ -1,3 +1,4 @@ +use std::collections::HashMap; use std::path::{Path, PathBuf}; use std::time::Duration; use uuid::Uuid; @@ -61,6 +62,7 @@ pub fn test_config() -> crate::config::Config { voice_name: crate::config::DEFAULT_VOICE_NAME.to_string(), agent: "codex".to_string(), routes: Vec::new(), + command_hooks: HashMap::new(), assistant_root: "/fake/assistant".to_string(), jobs_dir: "/fake/jobs".to_string(), jobs_agent: None,