diff --git a/Cargo.lock b/Cargo.lock index df2de6541..214965503 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -55,9 +55,9 @@ dependencies = [ [[package]] name = "anstream" -version = "0.6.21" +version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "43d5b281e737544384e969a5ccad3f1cdd24b48086a0fc1b2a5262a26b8f4f4a" +checksum = "824a212faf96e9acacdbd09febd34438f8f711fb84e09a8916013cd7815ca28d" dependencies = [ "anstyle", "anstyle-parse", @@ -70,15 +70,15 @@ dependencies = [ [[package]] name = "anstyle" -version = "1.0.13" +version = "1.0.14" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5192cca8006f1fd4f7237516f40fa183bb07f8fbdfedaa0036de5ea9b0b45e78" +checksum = "940b3a0ca603d1eade50a4846a2afffd5ef57a9feac2c0e2ec2e14f9ead76000" [[package]] name = "anstyle-parse" -version = "0.2.7" +version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4e7644824f0aa2c7b9384579234ef10eb7efb6a0deb83f9630a49594dd9c15c2" +checksum = "52ce7f38b242319f7cabaa6813055467063ecdc9d355bbb4ce0c68908cd8130e" dependencies = [ "utf8parse", ] @@ -464,9 +464,9 @@ dependencies = [ [[package]] name = "clap" -version = "4.5.60" +version = "4.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2797f34da339ce31042b27d23607e051786132987f595b02ba4f6a6dffb7030a" +checksum = "b193af5b67834b676abd72466a96c1024e6a6ad978a1f484bd90b85c94041351" dependencies = [ "clap_builder", "clap_derive", @@ -474,9 +474,9 @@ dependencies = [ [[package]] name = "clap_builder" -version = "4.5.60" +version = "4.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "24a241312cea5059b13574bb9b3861cabf758b879c15190b37b6d6fd63ab6876" +checksum = "714a53001bf66416adb0e2ef5ac857140e7dc3a0c48fb28b2f10762fc4b5069f" dependencies = [ "anstream", "anstyle", @@ -486,9 +486,9 @@ dependencies = [ [[package]] name = "clap_derive" -version = "4.5.55" +version = "4.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a92793da1a46a5f2a02a6f4c46c6496b28c43638adea8306fcb0caa1634f24e5" +checksum = "1110bd8a634a1ab8cb04345d8d878267d57c3cf1b38d91b71af6686408bbca6a" dependencies = [ "heck", "proc-macro2", @@ -498,15 +498,15 @@ dependencies = [ [[package]] name = "clap_lex" -version = "1.0.0" +version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3a822ea5bc7590f9d40f1ba12c0dc3c2760f3482c6984db1573ad11031420831" +checksum = "c8d4a3bb8b1e0c1050499d1815f5ab16d04f0959b233085fb31653fbfc9d98f9" [[package]] name = "colorchoice" -version = "1.0.4" +version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b05b61dc5112cbb17e4b6cd61790d9845d13888356391624cbe7e41efeac1e75" +checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" [[package]] name = "combine" @@ -1488,9 +1488,9 @@ checksum = "09edd9e8b54e49e587e4f6295a7d29c3ea94d469cb40ab8ca70b288248a81db2" [[package]] name = "libc" -version = "0.2.182" +version = "0.2.183" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6800badb6cb2082ffd7b6a67e6125bb39f18782f793520caee8cb8846be06112" +checksum = "b5b646652bf6661599e1da8901b3b9522896f01e736bad5f723fe7a3a27f899d" [[package]] name = "libm" @@ -1765,9 +1765,9 @@ dependencies = [ [[package]] name = "once_cell" -version = "1.21.3" +version = "1.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "42f5e15c9953c5e4ccceeb2e7382a716482c34515315f7b03532b8b4e8393d2d" +checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" [[package]] name = "once_cell_polyfill" @@ -1989,9 +1989,9 @@ dependencies = [ [[package]] name = "quinn-proto" -version = "0.11.13" +version = "0.11.14" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f1906b49b0c3bc04b5fe5d86a77925ae6524a19b816ae38ce1e426255f1d8a31" +checksum = "434b42fec591c96ef50e21e886936e66d3cc3f737104fdb9b737c40ffb94c098" dependencies = [ "bytes", "getrandom 0.3.4", @@ -2237,9 +2237,9 @@ dependencies = [ [[package]] name = "rmcp" -version = "1.1.0" +version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d2cb14cb9278a12eae884c9f3c0cfeca2cc28f361211206424a1d7abed95f090" +checksum = "ba6b9d2f0efe2258b23767f1f9e0054cfbcac9c2d6f81a031214143096d7864f" dependencies = [ "async-trait", "base64", @@ -2259,9 +2259,9 @@ dependencies = [ [[package]] name = "rmcp-macros" -version = "1.1.0" +version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6a02ea81d9482b07e1fe156ac7cf98b6823d51fb84531936a5e1cbb4eec31ad5" +checksum = "ab9d95d7ed26ad8306352b0d5f05b593222b272790564589790d210aa15caa9e" dependencies = [ "darling", "proc-macro2", @@ -2489,6 +2489,15 @@ dependencies = [ "serde_core", ] +[[package]] +name = "serde_spanned" +version = "0.6.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bf41e0cfaf7226dca15e8197172c295a782857fcb97fad1808a166870dee75a3" +dependencies = [ + "serde", +] + [[package]] name = "serde_urlencoded" version = "0.7.1" @@ -2655,6 +2664,8 @@ version = "0.1.0" dependencies = [ "anyhow", "chrono", + "clap", + "evalexpr", "futures-util", "nostr", "reqwest", @@ -2664,6 +2675,7 @@ dependencies = [ "thiserror", "tokio", "tokio-tungstenite 0.26.2", + "toml", "tracing", "tracing-subscriber", "url", @@ -2806,6 +2818,7 @@ dependencies = [ "dashmap", "futures-util", "hmac", + "moka", "nostr", "rand 0.8.5", "reqwest", @@ -3381,6 +3394,47 @@ dependencies = [ "tokio", ] +[[package]] +name = "toml" +version = "0.8.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc1beb996b9d83529a9e75c17a1686767d148d70663143c7854d8b4a09ced362" +dependencies = [ + "serde", + "serde_spanned", + "toml_datetime", + "toml_edit", +] + +[[package]] +name = "toml_datetime" +version = "0.6.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "22cddaf88f4fbc13c51aebbf5f8eceb5c7c5a9da2ac40a13519eb5b0a0e8f11c" +dependencies = [ + "serde", +] + +[[package]] +name = "toml_edit" +version = "0.22.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "41fe8c660ae4257887cf66394862d21dbca4a6ddd26f04a3560410406a2f819a" +dependencies = [ + "indexmap", + "serde", + "serde_spanned", + "toml_datetime", + "toml_write", + "winnow", +] + +[[package]] +name = "toml_write" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5d99f8c9a7727884afe522e9bd5edbfc91a3312b36a77b5fb8926e4c31a41801" + [[package]] name = "tower" version = "0.5.3" @@ -3489,9 +3543,9 @@ dependencies = [ [[package]] name = "tracing-subscriber" -version = "0.3.22" +version = "0.3.23" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2f30143827ddab0d256fd843b7a66d164e9f271cfa0dde49142c5ca0ca291f1e" +checksum = "cb7f578e5945fb242538965c2d0b04418d38ec25c79d160cd279bf0731c8d319" dependencies = [ "matchers", "nu-ansi-term", @@ -4133,6 +4187,15 @@ version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d6bbff5f0aada427a1e5a6da5f1f98158182f26556f345ac9e04d36d0ebed650" +[[package]] +name = "winnow" +version = "0.7.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df79d97927682d2fd8adb29682d1140b343be4ac0f08fd68b7765d9c059d3945" +dependencies = [ + "memchr", +] + [[package]] name = "wit-bindgen" version = "0.51.0" @@ -4252,18 +4315,18 @@ dependencies = [ [[package]] name = "zerocopy" -version = "0.8.40" +version = "0.8.42" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a789c6e490b576db9f7e6b6d661bcc9799f7c0ac8352f56ea20193b2681532e5" +checksum = "f2578b716f8a7a858b7f02d5bd870c14bf4ddbbcf3a4c05414ba6503640505e3" dependencies = [ "zerocopy-derive", ] [[package]] name = "zerocopy-derive" -version = "0.8.40" +version = "0.8.42" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f65c489a7071a749c849713807783f70672b28094011623e200cb86dcb835953" +checksum = "7e6cc098ea4d3bd6246687de65af3f920c430e236bee1e3bf2e441463f08a02f" dependencies = [ "proc-macro2", "quote", diff --git a/crates/sprout-acp/Cargo.toml b/crates/sprout-acp/Cargo.toml index 1cb33e21a..92b6ab962 100644 --- a/crates/sprout-acp/Cargo.toml +++ b/crates/sprout-acp/Cargo.toml @@ -47,5 +47,14 @@ tracing-subscriber = { workspace = true } thiserror = { workspace = true } anyhow = { workspace = true } +# CLI +clap = { version = "4", features = ["derive", "env"] } + +# Config file +toml = "0.8" + +# Filter expressions +evalexpr = { workspace = true } + [dev-dependencies] tokio = { workspace = true, features = ["test-util"] } diff --git a/crates/sprout-acp/src/acp.rs b/crates/sprout-acp/src/acp.rs index 049d19953..0da61d3cc 100644 --- a/crates/sprout-acp/src/acp.rs +++ b/crates/sprout-acp/src/acp.rs @@ -76,6 +76,9 @@ pub enum AcpError { #[error("Agent process exited unexpectedly")] AgentExited, + #[error("Turn timed out")] + Timeout, + #[error("Protocol error: {0}")] Protocol(String), } @@ -106,6 +109,10 @@ pub struct AcpClient { /// Guards against double-response if a timeout fires after the allow_once /// response was written but before `pending_permission_id` was cleared. permission_responded: bool, + /// The JSON-RPC id of the most recently sent `session/prompt` request. + /// Used by [`cancel_with_cleanup`] to drain the correct response. + /// Set in [`session_prompt`]; consumed in [`cancel_with_cleanup`]. + last_prompt_id: Option, } impl AcpClient { @@ -142,6 +149,7 @@ impl AcpClient { next_id: 0, pending_permission_id: None, permission_responded: false, + last_prompt_id: None, }) } @@ -199,7 +207,11 @@ impl AcpClient { { "type": "text", "text": prompt_text } ] }); + // Record the prompt request ID before send_request increments next_id. + // Used by cancel_with_cleanup to drain the correct response. + self.last_prompt_id = Some(self.next_id); let result = self.send_request("session/prompt", params).await?; + self.last_prompt_id = None; // Clear after normal completion. self.parse_stop_reason(&result) } @@ -228,6 +240,13 @@ impl AcpClient { /// /// Returns the final [`StopReason`] (almost always [`StopReason::Cancelled`]). pub async fn cancel_with_cleanup(&mut self, session_id: &str) -> Result { + // Validate precondition before any side effects — fail fast if there's + // no in-flight prompt (prevents writing permission responses or cancel + // notifications to the agent when no prompt is active). + let prompt_id = self.last_prompt_id.take().ok_or_else(|| { + AcpError::Protocol("cancel_with_cleanup called with no in-flight prompt".into()) + })?; + // Step 1: respond to any pending permission request with "cancelled", // but only if we haven't already responded (guards against double-response race). if let Some(perm_id) = self.pending_permission_id.clone() { @@ -246,10 +265,6 @@ impl AcpClient { // Step 2: send session/cancel notification (no id) self.session_cancel(session_id).await?; tracing::info!(target: "acp::cancel", "sent session/cancel for {session_id}"); - - // Step 3: drain until we get the session/prompt response. - // The prompt id is next_id - 1 (the most recently sent request was session/prompt). - let prompt_id = self.next_id - 1; let result = self.read_until_response(prompt_id).await?; self.parse_stop_reason(&result) } @@ -494,9 +509,11 @@ impl AcpClient { } }; - self.write_ndjson(&response).await?; - // Mark as responded and clear pending now that we've successfully written the response. + // Mark as responded BEFORE the write — if the future is dropped by a + // timeout between write and flag-set, cancel_with_cleanup would otherwise + // see permission_responded=false and send a second response (protocol violation). self.permission_responded = true; + self.write_ndjson(&response).await?; self.pending_permission_id = None; Ok(()) } diff --git a/crates/sprout-acp/src/config.rs b/crates/sprout-acp/src/config.rs index 8933df061..96ebfe454 100644 --- a/crates/sprout-acp/src/config.rs +++ b/crates/sprout-acp/src/config.rs @@ -1,107 +1,845 @@ +//! Configuration for the sprout-acp harness. +//! +//! CLI-first: every option is a CLI flag with env var fallback. +//! Config file (TOML) for complex subscription rules. + +use std::collections::HashMap; +use std::path::PathBuf; + +use clap::Parser; use nostr::Keys; use thiserror::Error; +use uuid::Uuid; + +use crate::filter::SubscriptionRule; + +// ── Errors ──────────────────────────────────────────────────────────────────── -/// Errors that can occur when loading configuration from environment variables. #[derive(Debug, Error)] pub enum ConfigError { - #[error("required environment variable {0} is not set")] - MissingVar(&'static str), + #[error("failed to parse nostr keys: {0}")] + KeyParse(#[from] nostr::key::Error), - #[error("failed to parse nostr keys from {var}: {source}")] - KeyParse { - var: &'static str, - #[source] - source: nostr::key::Error, - }, + #[error("failed to read file: {0}")] + Io(#[from] std::io::Error), + + #[error("config file error: {0}")] + ConfigFile(String), } -/// Configuration for the sprout-acp harness. -#[derive(Debug)] -pub struct Config { - /// Agent's nostr keypair — used for both relay auth and agent identity. - /// - /// Parsed from `SPROUT_PRIVATE_KEY` (preferred) or the legacy - /// `SPROUT_ACP_PRIVATE_KEY` + `SPROUT_AGENT_PRIVATE_KEY` pair. - pub keys: Keys, - /// API token, optional. Required if the relay enforces token auth. - pub api_token: Option, - /// Relay WebSocket URL (`SPROUT_RELAY_URL`). Default: `ws://localhost:3000`. +// ── Enums ───────────────────────────────────────────────────────────────────── + +#[derive(Debug, Clone, clap::ValueEnum)] +pub enum SubscribeMode { + Mentions, + All, + Config, +} + +#[derive(Debug, Clone, Copy, clap::ValueEnum)] +pub enum DedupMode { + Drop, + Queue, +} + +// ── CLI ─────────────────────────────────────────────────────────────────────── + +#[derive(Debug, Parser)] +#[command( + name = "sprout-acp", + about = "ACP harness that bridges Sprout events to AI agents" +)] +pub struct CliArgs { + #[arg(long, env = "SPROUT_RELAY_URL", default_value = "ws://localhost:3000")] pub relay_url: String, - // --- Agent binary --- - /// Agent command (`SPROUT_ACP_AGENT_COMMAND`). Default: `goose`. + #[arg(long, env = "SPROUT_PRIVATE_KEY")] + pub private_key: String, + + #[arg(long, env = "SPROUT_API_TOKEN")] + pub api_token: Option, + + #[arg(long, env = "SPROUT_ACP_AGENT_COMMAND", default_value = "goose")] pub agent_command: String, - /// Agent arguments (`SPROUT_ACP_AGENT_ARGS`, comma-separated). Default: `["acp"]`. + + #[arg( + long, + env = "SPROUT_ACP_AGENT_ARGS", + default_value = "acp", + value_delimiter = ',' + )] pub agent_args: Vec, - // --- MCP server --- - /// MCP server binary path (`SPROUT_ACP_MCP_COMMAND`). Default: `sprout-mcp-server`. + #[arg( + long, + env = "SPROUT_ACP_MCP_COMMAND", + default_value = "sprout-mcp-server" + )] pub mcp_command: String, - // --- Tuning --- - /// Maximum turn duration in seconds (`SPROUT_ACP_TURN_TIMEOUT`). Default: 300. - pub turn_timeout_secs: u64, + #[arg(long, env = "SPROUT_ACP_TURN_TIMEOUT", default_value = "300")] + pub turn_timeout: u64, + + #[arg( + long, + env = "SPROUT_ACP_SYSTEM_PROMPT", + conflicts_with = "system_prompt_file" + )] + pub system_prompt: Option, + + #[arg( + long, + env = "SPROUT_ACP_SYSTEM_PROMPT_FILE", + conflicts_with = "system_prompt" + )] + pub system_prompt_file: Option, + + #[arg(long, env = "SPROUT_ACP_INITIAL_MESSAGE")] + pub initial_message: Option, + + #[arg( + long, + env = "SPROUT_ACP_SUBSCRIBE", + default_value = "mentions", + value_enum + )] + pub subscribe: SubscribeMode, + + #[arg(long, env = "SPROUT_ACP_KINDS", value_delimiter = ',')] + pub kinds: Option>, + + #[arg(long, env = "SPROUT_ACP_CHANNELS", value_delimiter = ',')] + pub channels: Option>, + + #[arg(long, env = "SPROUT_ACP_NO_MENTION_FILTER")] + pub no_mention_filter: bool, + + #[arg(long, env = "SPROUT_ACP_CONFIG", default_value = "./sprout-acp.toml")] + pub config: PathBuf, + + #[arg(long, env = "SPROUT_ACP_DEDUP", default_value = "drop", value_enum)] + pub dedup: DedupMode, + + #[arg(long, env = "SPROUT_ACP_NO_IGNORE_SELF")] + pub no_ignore_self: bool, } -/// Parse a nostr `Keys` from the named environment variable. -fn parse_keys_var(var: &'static str) -> Result { - let nsec = std::env::var(var).map_err(|_| ConfigError::MissingVar(var))?; - Keys::parse(&nsec).map_err(|e| ConfigError::KeyParse { var, source: e }) +// ── Merged NIP-01 filter ────────────────────────────────────────────────────── + +/// Merged NIP-01 subscription filter for a single channel. +#[derive(Debug, Clone)] +pub struct ChannelFilter { + /// Event kinds to subscribe to. None = wildcard (all kinds). + pub kinds: Option>, + /// Whether to include `#p` tag filter for agent pubkey. + pub require_mention: bool, +} + +// ── Resolved config ─────────────────────────────────────────────────────────── + +#[derive(Debug)] +pub struct Config { + pub keys: Keys, + pub api_token: Option, + pub relay_url: String, + pub agent_command: String, + pub agent_args: Vec, + pub mcp_command: String, + pub turn_timeout_secs: u64, + pub system_prompt: Option, + pub initial_message: Option, + pub subscribe_mode: SubscribeMode, + pub dedup_mode: DedupMode, + pub ignore_self: bool, + pub kinds_override: Option>, + pub channels_override: Option>, + pub no_mention_filter: bool, + pub config_path: PathBuf, } impl Config { - /// Load configuration from environment variables. - /// - /// Key resolution order: - /// 1. `SPROUT_PRIVATE_KEY` — single key for everything (preferred) - /// 2. `SPROUT_ACP_PRIVATE_KEY` — legacy harness key (fallback) - pub fn from_env() -> Result { - let keys = parse_keys_var("SPROUT_PRIVATE_KEY") - .or_else(|_| parse_keys_var("SPROUT_ACP_PRIVATE_KEY"))?; + pub fn from_cli() -> Result { + // Propagate legacy env vars so clap sees them under the canonical names. + // Only set if the canonical var is absent — clap's `env` reads these. + // NOTE: std::env::set_var is safe in edition 2021 but will require + // `unsafe` in edition 2024. This runs inside #[tokio::main] so worker + // threads are already alive. When migrating to edition 2024, move this + // to a non-async fn main() wrapper that runs before the runtime starts. + for (legacy, canonical) in [ + ("SPROUT_ACP_PRIVATE_KEY", "SPROUT_PRIVATE_KEY"), + ("SPROUT_ACP_API_TOKEN", "SPROUT_API_TOKEN"), + ] { + if std::env::var(canonical).is_err() { + if let Ok(val) = std::env::var(legacy) { + std::env::set_var(canonical, &val); + } + } + } - let api_token = std::env::var("SPROUT_API_TOKEN") - .or_else(|_| std::env::var("SPROUT_ACP_API_TOKEN")) - .ok(); + let args = CliArgs::parse(); + let keys = Keys::parse(&args.private_key)?; - let relay_url = - std::env::var("SPROUT_RELAY_URL").unwrap_or_else(|_| "ws://localhost:3000".to_string()); + let system_prompt = if let Some(text) = args.system_prompt { + Some(text) + } else if let Some(ref path) = args.system_prompt_file { + Some(std::fs::read_to_string(path)?) + } else { + None + }; - let agent_command = - std::env::var("SPROUT_ACP_AGENT_COMMAND").unwrap_or_else(|_| "goose".to_string()); - - let agent_args = std::env::var("SPROUT_ACP_AGENT_ARGS") - .map(|s| s.split(',').map(|a| a.trim().to_string()).collect()) - .unwrap_or_else(|_| vec!["acp".to_string()]); - - let mcp_command = std::env::var("SPROUT_ACP_MCP_COMMAND") - .unwrap_or_else(|_| "sprout-mcp-server".to_string()); - - let turn_timeout_secs = std::env::var("SPROUT_ACP_TURN_TIMEOUT") - .ok() - .and_then(|s| s.parse::().ok()) - .unwrap_or(300); + if matches!(args.subscribe, SubscribeMode::Config) { + if args.kinds.is_some() { + tracing::warn!("--kinds is ignored in config mode"); + } + if args.channels.is_some() { + tracing::warn!("--channels is ignored in config mode"); + } + if args.no_mention_filter { + tracing::warn!("--no-mention-filter is ignored in config mode"); + } + } Ok(Config { keys, - api_token, - relay_url, - agent_command, - agent_args, - mcp_command, - turn_timeout_secs, + api_token: args.api_token, + relay_url: args.relay_url, + agent_command: args.agent_command, + agent_args: args.agent_args, + mcp_command: args.mcp_command, + turn_timeout_secs: args.turn_timeout, + system_prompt, + initial_message: args.initial_message, + subscribe_mode: args.subscribe, + dedup_mode: args.dedup, + ignore_self: !args.no_ignore_self, + kinds_override: args.kinds, + channels_override: args.channels, + no_mention_filter: args.no_mention_filter, + config_path: args.config, }) } - /// Return a human-readable summary (no secrets). + /// Human-readable summary (no secrets). pub fn summary(&self) -> String { format!( - "relay={} pubkey={} agent_cmd={} {} mcp_cmd={} turn_timeout={}s", + "relay={} pubkey={} agent_cmd={} {} mcp_cmd={} timeout={}s subscribe={:?} dedup={:?} ignore_self={}", self.relay_url, self.keys.public_key().to_hex(), self.agent_command, self.agent_args.join(" "), self.mcp_command, self.turn_timeout_secs, + self.subscribe_mode, + self.dedup_mode, + self.ignore_self, ) } } + +// ── TOML config file ────────────────────────────────────────────────────────── + +#[derive(Debug, serde::Deserialize)] +struct TomlConfig { + #[serde(default)] + rules: Vec, +} + +pub fn load_rules(path: &std::path::Path) -> Result, ConfigError> { + let content = std::fs::read_to_string(path)?; + let config: TomlConfig = + toml::from_str(&content).map_err(|e| ConfigError::ConfigFile(e.to_string()))?; + + if config.rules.len() > 100 { + return Err(ConfigError::ConfigFile(format!( + "too many rules ({}, max 100)", + config.rules.len() + ))); + } + + let mut seen_names = std::collections::HashSet::new(); + for rule in &config.rules { + if rule.name.trim().is_empty() { + return Err(ConfigError::ConfigFile( + "rule name must not be empty".into(), + )); + } + if !seen_names.insert(&rule.name) { + return Err(ConfigError::ConfigFile(format!( + "duplicate rule name: {}", + rule.name + ))); + } + if let Some(ref expr) = rule.filter { + if expr.len() > 4096 { + return Err(ConfigError::ConfigFile(format!( + "rule '{}': filter too long ({} bytes, max 4096)", + rule.name, + expr.len() + ))); + } + // Fail fast: parse the expression at load time so typos don't + // silently produce dead rules at runtime. + if let Err(e) = evalexpr::build_operator_tree(expr) { + return Err(ConfigError::ConfigFile(format!( + "rule '{}': invalid filter expression: {e}", + rule.name, + ))); + } + } + // Validate channel scope — catch typos like "ALL" or "All" early. + if let crate::filter::ChannelScope::All(ref s) = rule.channels { + if s != "all" { + return Err(ConfigError::ConfigFile(format!( + "rule '{}': channels must be \"all\" or a list, got {:?}", + rule.name, s, + ))); + } + } + } + + Ok(config.rules) +} + +// ── Subscription resolution ─────────────────────────────────────────────────── + +/// Resolve per-channel NIP-01 filters from config + discovered channels. +pub fn resolve_channel_filters( + config: &Config, + discovered_channels: &[Uuid], + rules: &[SubscriptionRule], +) -> HashMap { + use sprout_core::kind::{ + KIND_STREAM_MESSAGE, KIND_STREAM_REMINDER, KIND_WORKFLOW_APPROVAL_REQUESTED, + }; + + let target_channels: Vec = if let Some(ref overrides) = config.channels_override { + overrides + .iter() + .filter_map(|s| s.parse::().ok()) + .filter(|id| discovered_channels.contains(id)) + .collect() + } else { + discovered_channels.to_vec() + }; + + let mut result = HashMap::new(); + + match config.subscribe_mode { + SubscribeMode::Mentions => { + let kinds = config.kinds_override.clone().unwrap_or_else(|| { + vec![ + KIND_STREAM_MESSAGE, + KIND_WORKFLOW_APPROVAL_REQUESTED, + KIND_STREAM_REMINDER, + ] + }); + let require_mention = !config.no_mention_filter; + for ch in &target_channels { + result.insert( + *ch, + ChannelFilter { + kinds: Some(kinds.clone()), + require_mention, + }, + ); + } + } + SubscribeMode::All => { + for ch in &target_channels { + result.insert( + *ch, + ChannelFilter { + kinds: config.kinds_override.clone(), + require_mention: false, + }, + ); + } + } + SubscribeMode::Config => { + for ch in discovered_channels { + let mut merged_kinds: Option> = Some(vec![]); + let mut require_mention = true; + let mut has_rule = false; + + for rule in rules { + if !rule_applies_to_channel(rule, *ch) { + continue; + } + has_rule = true; + if rule.kinds.is_empty() { + merged_kinds = None; + } else if let Some(ref mut kinds) = merged_kinds { + for k in &rule.kinds { + if !kinds.contains(k) { + kinds.push(*k); + } + } + } + if !rule.require_mention { + require_mention = false; + } + } + + if has_rule { + result.insert( + *ch, + ChannelFilter { + kinds: merged_kinds, + require_mention, + }, + ); + } + } + } + } + + result +} + +fn rule_applies_to_channel(rule: &SubscriptionRule, channel_id: Uuid) -> bool { + use crate::filter::ChannelScope; + match &rule.channels { + ChannelScope::All(s) if s == "all" => true, + ChannelScope::List(ids) => ids + .iter() + .any(|id| id.parse::().ok() == Some(channel_id)), + _ => false, + } +} + +// ── Tests ───────────────────────────────────────────────────────────────────── + +#[cfg(test)] +mod tests { + use super::*; + use crate::filter::{ChannelScope, SubscriptionRule}; + + /// Build a minimal Config for testing without CLI parsing. + fn test_config(mode: SubscribeMode) -> Config { + Config { + keys: nostr::Keys::generate(), + api_token: None, + relay_url: "ws://localhost:3000".into(), + agent_command: "goose".into(), + agent_args: vec!["acp".into()], + mcp_command: "sprout-mcp-server".into(), + turn_timeout_secs: 300, + system_prompt: None, + initial_message: None, + subscribe_mode: mode, + dedup_mode: DedupMode::Drop, + ignore_self: true, + kinds_override: None, + channels_override: None, + no_mention_filter: false, + config_path: PathBuf::from("./sprout-acp.toml"), + } + } + + fn make_rule( + name: &str, + channels: ChannelScope, + kinds: Vec, + mention: bool, + ) -> SubscriptionRule { + SubscriptionRule { + name: name.into(), + channels, + kinds, + require_mention: mention, + filter: None, + prompt_tag: None, + } + } + + // ── resolve_channel_filters: Mentions mode ─────────────────────────────── + + #[test] + fn test_mentions_mode_default_kinds() { + let config = test_config(SubscribeMode::Mentions); + let channels = vec![Uuid::new_v4(), Uuid::new_v4()]; + let result = resolve_channel_filters(&config, &channels, &[]); + + assert_eq!(result.len(), 2); + for ch in &channels { + let f = result.get(ch).expect("channel should be present"); + assert!(f.require_mention, "mentions mode requires mention"); + let kinds = f.kinds.as_ref().expect("should have kinds"); + assert!(kinds.contains(&sprout_core::kind::KIND_STREAM_MESSAGE)); + assert!(kinds.contains(&sprout_core::kind::KIND_WORKFLOW_APPROVAL_REQUESTED)); + assert!(kinds.contains(&sprout_core::kind::KIND_STREAM_REMINDER)); + } + } + + #[test] + fn test_mentions_mode_custom_kinds() { + let mut config = test_config(SubscribeMode::Mentions); + config.kinds_override = Some(vec![1, 7]); + let channels = vec![Uuid::new_v4()]; + let result = resolve_channel_filters(&config, &channels, &[]); + + let f = result.get(&channels[0]).unwrap(); + assert_eq!(f.kinds.as_ref().unwrap(), &[1, 7]); + } + + #[test] + fn test_mentions_mode_no_mention_filter() { + let mut config = test_config(SubscribeMode::Mentions); + config.no_mention_filter = true; + let channels = vec![Uuid::new_v4()]; + let result = resolve_channel_filters(&config, &channels, &[]); + + let f = result.get(&channels[0]).unwrap(); + assert!(!f.require_mention); + } + + // ── resolve_channel_filters: All mode ──────────────────────────────────── + + #[test] + fn test_all_mode_wildcard() { + let config = test_config(SubscribeMode::All); + let channels = vec![Uuid::new_v4(), Uuid::new_v4(), Uuid::new_v4()]; + let result = resolve_channel_filters(&config, &channels, &[]); + + assert_eq!(result.len(), 3); + for ch in &channels { + let f = result.get(ch).unwrap(); + assert!( + f.kinds.is_none(), + "all mode with no override = wildcard kinds" + ); + assert!(!f.require_mention); + } + } + + #[test] + fn test_all_mode_with_kinds_override() { + let mut config = test_config(SubscribeMode::All); + config.kinds_override = Some(vec![40001, 7]); + let channels = vec![Uuid::new_v4()]; + let result = resolve_channel_filters(&config, &channels, &[]); + + let f = result.get(&channels[0]).unwrap(); + assert_eq!(f.kinds.as_ref().unwrap(), &[40001, 7]); + } + + // ── resolve_channel_filters: channels_override ─────────────────────────── + + #[test] + fn test_channels_override_filters_to_discovered() { + let mut config = test_config(SubscribeMode::All); + let ch_a = Uuid::new_v4(); + let ch_b = Uuid::new_v4(); + let ch_unknown = Uuid::new_v4(); + // Override includes ch_a and an unknown channel. + config.channels_override = Some(vec![ch_a.to_string(), ch_unknown.to_string()]); + + let discovered = vec![ch_a, ch_b]; + let result = resolve_channel_filters(&config, &discovered, &[]); + + // Only ch_a should be present (intersection of override and discovered). + assert_eq!(result.len(), 1); + assert!(result.contains_key(&ch_a)); + assert!(!result.contains_key(&ch_b)); + assert!(!result.contains_key(&ch_unknown)); + } + + // ── resolve_channel_filters: Config mode ───────────────────────────────── + + #[test] + fn test_config_mode_single_rule_all_channels() { + let config = test_config(SubscribeMode::Config); + let ch = Uuid::new_v4(); + let rules = vec![make_rule( + "catch-all", + ChannelScope::All("all".into()), + vec![40001], + false, + )]; + + let result = resolve_channel_filters(&config, &[ch], &rules); + assert_eq!(result.len(), 1); + let f = result.get(&ch).unwrap(); + assert_eq!(f.kinds.as_ref().unwrap(), &[40001]); + assert!(!f.require_mention); + } + + #[test] + fn test_config_mode_rule_targets_specific_channel() { + let config = test_config(SubscribeMode::Config); + let ch_a = Uuid::new_v4(); + let ch_b = Uuid::new_v4(); + let rules = vec![make_rule( + "only-a", + ChannelScope::List(vec![ch_a.to_string()]), + vec![40001], + false, + )]; + + let result = resolve_channel_filters(&config, &[ch_a, ch_b], &rules); + assert_eq!(result.len(), 1); + assert!(result.contains_key(&ch_a)); + assert!(!result.contains_key(&ch_b)); + } + + #[test] + fn test_config_mode_merge_overlapping_rules() { + let config = test_config(SubscribeMode::Config); + let ch = Uuid::new_v4(); + // Two rules both match the same channel with different kinds. + let rules = vec![ + make_rule( + "messages", + ChannelScope::All("all".into()), + vec![40001], + true, + ), + make_rule("reactions", ChannelScope::All("all".into()), vec![7], false), + ]; + + let result = resolve_channel_filters(&config, &[ch], &rules); + let f = result.get(&ch).unwrap(); + // Kinds should be the union: [40001, 7]. + let kinds = f.kinds.as_ref().expect("should have merged kinds"); + assert!(kinds.contains(&40001)); + assert!(kinds.contains(&7)); + // require_mention should be false (most permissive wins). + assert!(!f.require_mention); + } + + #[test] + fn test_config_mode_wildcard_kinds_propagates() { + let config = test_config(SubscribeMode::Config); + let ch = Uuid::new_v4(); + // First rule has specific kinds, second has empty (wildcard). + let rules = vec![ + make_rule( + "narrow", + ChannelScope::All("all".into()), + vec![40001], + false, + ), + make_rule("broad", ChannelScope::All("all".into()), vec![], false), + ]; + + let result = resolve_channel_filters(&config, &[ch], &rules); + let f = result.get(&ch).unwrap(); + // Once any rule has empty kinds (wildcard), merged result is None (wildcard). + assert!(f.kinds.is_none(), "wildcard should propagate"); + } + + #[test] + fn test_config_mode_no_matching_rules_empty_result() { + let config = test_config(SubscribeMode::Config); + let ch = Uuid::new_v4(); + let other_ch = Uuid::new_v4(); + // Rule only targets other_ch. + let rules = vec![make_rule( + "other", + ChannelScope::List(vec![other_ch.to_string()]), + vec![40001], + false, + )]; + + let result = resolve_channel_filters(&config, &[ch], &rules); + assert!(result.is_empty()); + } + + #[test] + fn test_config_mode_require_mention_most_permissive() { + let config = test_config(SubscribeMode::Config); + let ch = Uuid::new_v4(); + // First rule requires mention, second doesn't. + let rules = vec![ + make_rule("strict", ChannelScope::All("all".into()), vec![40001], true), + make_rule("lax", ChannelScope::All("all".into()), vec![7], false), + ]; + + let result = resolve_channel_filters(&config, &[ch], &rules); + let f = result.get(&ch).unwrap(); + assert!(!f.require_mention, "most permissive (false) should win"); + } + + // ── rule_applies_to_channel ────────────────────────────────────────────── + + #[test] + fn test_rule_applies_all() { + let rule = make_rule("test", ChannelScope::All("all".into()), vec![], false); + assert!(rule_applies_to_channel(&rule, Uuid::new_v4())); + } + + #[test] + fn test_rule_applies_all_invalid_string() { + let rule = make_rule("test", ChannelScope::All("ALL".into()), vec![], false); + assert!(!rule_applies_to_channel(&rule, Uuid::new_v4())); + } + + #[test] + fn test_rule_applies_list_match() { + let ch = Uuid::new_v4(); + let rule = make_rule( + "test", + ChannelScope::List(vec![ch.to_string()]), + vec![], + false, + ); + assert!(rule_applies_to_channel(&rule, ch)); + } + + #[test] + fn test_rule_applies_list_no_match() { + let rule = make_rule( + "test", + ChannelScope::List(vec![Uuid::new_v4().to_string()]), + vec![], + false, + ); + assert!(!rule_applies_to_channel(&rule, Uuid::new_v4())); + } + + // ── load_rules validation ──────────────────────────────────────────────── + + #[test] + fn test_load_rules_valid_toml() { + let dir = std::env::temp_dir().join("sprout-acp-test-valid"); + let path = dir.join("rules.toml"); + std::fs::create_dir_all(&dir).unwrap(); + std::fs::write( + &path, + r#" +[[rules]] +name = "catch-all" +channels = "all" +kinds = [40001] +require_mention = false +"#, + ) + .unwrap(); + + let rules = load_rules(&path).unwrap(); + assert_eq!(rules.len(), 1); + assert_eq!(rules[0].name, "catch-all"); + std::fs::remove_dir_all(&dir).ok(); + } + + #[test] + fn test_load_rules_empty_name_rejected() { + let dir = std::env::temp_dir().join("sprout-acp-test-empty-name"); + let path = dir.join("rules.toml"); + std::fs::create_dir_all(&dir).unwrap(); + std::fs::write( + &path, + r#" +[[rules]] +name = " " +channels = "all" +"#, + ) + .unwrap(); + + let err = load_rules(&path).unwrap_err(); + assert!(err.to_string().contains("name must not be empty")); + std::fs::remove_dir_all(&dir).ok(); + } + + #[test] + fn test_load_rules_duplicate_name_rejected() { + let dir = std::env::temp_dir().join("sprout-acp-test-dup-name"); + let path = dir.join("rules.toml"); + std::fs::create_dir_all(&dir).unwrap(); + std::fs::write( + &path, + r#" +[[rules]] +name = "dup" +channels = "all" + +[[rules]] +name = "dup" +channels = "all" +"#, + ) + .unwrap(); + + let err = load_rules(&path).unwrap_err(); + assert!(err.to_string().contains("duplicate rule name")); + std::fs::remove_dir_all(&dir).ok(); + } + + #[test] + fn test_load_rules_invalid_filter_rejected() { + let dir = std::env::temp_dir().join("sprout-acp-test-bad-filter"); + let path = dir.join("rules.toml"); + std::fs::create_dir_all(&dir).unwrap(); + // evalexpr rejects unbalanced parens at parse time. + std::fs::write( + &path, + r#" +[[rules]] +name = "bad" +channels = "all" +filter = "(((" +"#, + ) + .unwrap(); + + let err = load_rules(&path).unwrap_err(); + assert!(err.to_string().contains("invalid filter expression")); + std::fs::remove_dir_all(&dir).ok(); + } + + #[test] + fn test_load_rules_channel_scope_typo_rejected() { + let dir = std::env::temp_dir().join("sprout-acp-test-scope-typo"); + let path = dir.join("rules.toml"); + std::fs::create_dir_all(&dir).unwrap(); + std::fs::write( + &path, + r#" +[[rules]] +name = "typo" +channels = "ALL" +"#, + ) + .unwrap(); + + let err = load_rules(&path).unwrap_err(); + assert!(err.to_string().contains("must be \"all\" or a list")); + std::fs::remove_dir_all(&dir).ok(); + } + + #[test] + fn test_load_rules_too_many_rules_rejected() { + let dir = std::env::temp_dir().join("sprout-acp-test-too-many"); + let path = dir.join("rules.toml"); + std::fs::create_dir_all(&dir).unwrap(); + let mut toml = String::new(); + for i in 0..101 { + toml.push_str(&format!( + "[[rules]]\nname = \"rule-{i}\"\nchannels = \"all\"\n\n" + )); + } + std::fs::write(&path, &toml).unwrap(); + + let err = load_rules(&path).unwrap_err(); + assert!(err.to_string().contains("too many rules")); + std::fs::remove_dir_all(&dir).ok(); + } + + #[test] + fn test_load_rules_filter_too_long_rejected() { + let dir = std::env::temp_dir().join("sprout-acp-test-long-filter"); + let path = dir.join("rules.toml"); + std::fs::create_dir_all(&dir).unwrap(); + let long_expr = format!("\"{}\"", "a".repeat(4097)); + std::fs::write( + &path, + format!("[[rules]]\nname = \"long\"\nchannels = \"all\"\nfilter = {long_expr}\n"), + ) + .unwrap(); + + let err = load_rules(&path).unwrap_err(); + assert!(err.to_string().contains("filter too long")); + std::fs::remove_dir_all(&dir).ok(); + } +} diff --git a/crates/sprout-acp/src/filter.rs b/crates/sprout-acp/src/filter.rs new file mode 100644 index 000000000..8cb93d841 --- /dev/null +++ b/crates/sprout-acp/src/filter.rs @@ -0,0 +1,557 @@ +//! Content filtering and subscription rule matching. +//! +//! Responsibilities: +//! - Building an evalexpr context from a Nostr event +//! - Evaluating boolean filter expressions with a hard timeout +//! - Matching events against ordered subscription rules (first match wins) + +use std::time::Duration; + +use tracing::warn; + +// ── Errors ──────────────────────────────────────────────────────────────────── + +/// Errors that can occur during filter expression evaluation. +#[derive(Debug, thiserror::Error)] +pub enum FilterError { + #[error("expression too long ({len} bytes, max {max})")] + ExpressionTooLong { len: usize, max: usize }, + #[error("evaluation timed out")] + Timeout, + #[error("evaluation error: {0}")] + EvalError(String), +} + +// ── FilterContext ───────────────────────────────────────────────────────────── + +/// Variables extracted from a Nostr event for use in filter expressions. +#[derive(Debug, Clone)] +pub struct FilterContext { + /// Event content (message body). + pub content: String, + /// Event author pubkey as hex string. + pub author: String, + /// Nostr event kind number. + pub kind: u32, + /// Channel UUID as string. + pub channel_id: String, + /// Event `created_at` unix timestamp. + pub timestamp: u64, +} + +impl FilterContext { + /// Build a `FilterContext` from a Nostr event and its channel UUID. + pub fn from_event(event: &nostr::Event, channel_id: uuid::Uuid) -> Self { + Self { + content: event.content.clone(), + author: event.pubkey.to_hex(), + kind: event.kind.as_u16() as u32, + channel_id: channel_id.to_string(), + timestamp: event.created_at.as_u64(), + } + } +} + +// ── SubscriptionRule ────────────────────────────────────────────────────────── + +/// Scope of channels a subscription rule applies to. +#[derive(Debug, Clone, serde::Deserialize)] +#[serde(untagged)] +pub enum ChannelScope { + /// The literal string `"all"` — matches every channel. + All(String), + /// An explicit list of channel UUID strings. + List(Vec), +} + +impl ChannelScope { + /// Returns `true` if this scope covers the given channel UUID. + /// + /// `ChannelScope::All` only matches when the inner string is exactly `"all"`. + pub fn matches(&self, channel_id: &uuid::Uuid) -> bool { + match self { + ChannelScope::All(s) => s == "all", + ChannelScope::List(ids) => ids.iter().any(|id| id == &channel_id.to_string()), + } + } +} + +/// A single subscription rule from the agent config. +#[derive(Debug, Clone, serde::Deserialize)] +pub struct SubscriptionRule { + /// Human-readable rule name; used as fallback `prompt_tag`. + pub name: String, + /// Which channels this rule applies to. + pub channels: ChannelScope, + /// Nostr event kinds to match. Empty = wildcard (all kinds). + #[serde(default)] + pub kinds: Vec, + /// If `true`, the event must contain a `p` tag referencing the agent pubkey. + #[serde(default)] + pub require_mention: bool, + /// Optional evalexpr boolean expression for fine-grained filtering. + #[serde(default)] + pub filter: Option, + /// Tag passed to the prompt template. Falls back to `name` if absent. + #[serde(default)] + pub prompt_tag: Option, +} + +// ── MatchedRule ─────────────────────────────────────────────────────────────── + +/// The result of a successful rule match. +#[derive(Debug, Clone)] +pub struct MatchedRule { + /// Zero-based index of the matching rule in the rules slice. + #[cfg_attr(not(test), allow(dead_code))] + pub rule_index: usize, + /// Prompt tag to use (rule's `prompt_tag` or its `name`). + pub prompt_tag: String, +} + +// ── evaluate_filter ─────────────────────────────────────────────────────────── + +/// Maximum expression length accepted by `evaluate_filter`. +/// +/// Bounds worst-case O(2^n) evaluation paths. The spawn_blocking thread cannot +/// be cancelled after a timeout fires, so we cap length before dispatching. +const MAX_EXPR_LEN: usize = 4096; + +/// Maximum wall-clock time allowed for a single evalexpr evaluation. +const EVAL_TIMEOUT: Duration = Duration::from_millis(100); + +/// Evaluate a boolean filter expression against a `FilterContext`. +/// +/// - Caps expression length at [`MAX_EXPR_LEN`] bytes. +/// - Runs evaluation on a blocking thread with a [`EVAL_TIMEOUT`] hard timeout. +/// - Registers custom string helpers: `str_contains`, `str_starts_with`, +/// `str_ends_with`, `str_len` (duplicated intentionally from sprout-workflow). +pub async fn evaluate_filter(expr: &str, ctx: &FilterContext) -> Result { + if expr.len() > MAX_EXPR_LEN { + return Err(FilterError::ExpressionTooLong { + len: expr.len(), + max: MAX_EXPR_LEN, + }); + } + + let eval_ctx = build_eval_context(ctx).map_err(FilterError::EvalError)?; + let expr_owned = expr.to_owned(); + + let result = tokio::time::timeout( + EVAL_TIMEOUT, + tokio::task::spawn_blocking(move || { + evalexpr::eval_boolean_with_context(&expr_owned, &eval_ctx) + }), + ) + .await + .map_err(|_| FilterError::Timeout)? + .map_err(|e| FilterError::EvalError(format!("eval task panicked: {e}")))? + .map_err(|e| FilterError::EvalError(e.to_string()))?; + + Ok(result) +} + +/// Build an `evalexpr::HashMapContext` from a `FilterContext`. +/// +/// Variables exposed to expressions: +/// +/// | Name | Type | Source | +/// |--------------|--------|---------------------------| +/// | `content` | string | `event.content` | +/// | `author` | string | `event.pubkey` (hex) | +/// | `kind` | int | `event.kind` | +/// | `channel_id` | string | channel UUID | +/// | `timestamp` | int | `event.created_at` | +/// +/// Also registers `str_contains`, `str_starts_with`, `str_ends_with`, +/// `str_len` — duplicated from sprout-workflow intentionally so this crate +/// has no runtime dependency on sprout-workflow. +fn build_eval_context(ctx: &FilterContext) -> Result { + use evalexpr::*; + + let mut eval_ctx = HashMapContext::new(); + + // ── Custom string functions ─────────────────────────────────────────────── + // evalexpr v11 does not ship these helpers; register them manually. + + eval_ctx + .set_function( + "str_contains".into(), + Function::new(|args| { + let args = args.as_fixed_len_tuple(2)?; + let haystack = args[0].as_string()?; + let needle = args[1].as_string()?; + Ok(Value::Boolean(haystack.contains(needle.as_str()))) + }), + ) + .map_err(|e| e.to_string())?; + + eval_ctx + .set_function( + "str_starts_with".into(), + Function::new(|args| { + let args = args.as_fixed_len_tuple(2)?; + let s = args[0].as_string()?; + let prefix = args[1].as_string()?; + Ok(Value::Boolean(s.starts_with(prefix.as_str()))) + }), + ) + .map_err(|e| e.to_string())?; + + eval_ctx + .set_function( + "str_ends_with".into(), + Function::new(|args| { + let args = args.as_fixed_len_tuple(2)?; + let s = args[0].as_string()?; + let suffix = args[1].as_string()?; + Ok(Value::Boolean(s.ends_with(suffix.as_str()))) + }), + ) + .map_err(|e| e.to_string())?; + + eval_ctx + .set_function( + "str_len".into(), + Function::new(|arg| { + let s = arg.as_string()?; + Ok(Value::Int(s.len() as i64)) + }), + ) + .map_err(|e| e.to_string())?; + + // ── Event variables ─────────────────────────────────────────────────────── + + eval_ctx + .set_value("content".into(), Value::String(ctx.content.clone())) + .map_err(|e| e.to_string())?; + eval_ctx + .set_value("author".into(), Value::String(ctx.author.clone())) + .map_err(|e| e.to_string())?; + eval_ctx + .set_value("kind".into(), Value::Int(ctx.kind as i64)) + .map_err(|e| e.to_string())?; + eval_ctx + .set_value("channel_id".into(), Value::String(ctx.channel_id.clone())) + .map_err(|e| e.to_string())?; + eval_ctx + .set_value("timestamp".into(), Value::Int(ctx.timestamp as i64)) + .map_err(|e| e.to_string())?; + + Ok(eval_ctx) +} + +// ── match_event ─────────────────────────────────────────────────────────────── + +/// Match a Nostr event against an ordered list of subscription rules. +/// +/// Rules are evaluated in order; the first rule whose conditions all pass +/// wins. Returns `None` if no rule matches. +/// +/// # Matching logic (per rule) +/// +/// 1. **channels** — if not `"all"`, the event's channel UUID must be in the list. +/// 2. **kinds** — if non-empty, the event kind must be in the list. +/// 3. **require_mention** — if `true`, a `p` tag matching `agent_pubkey_hex` must exist. +/// 4. **filter** — if `Some`, the evalexpr expression must evaluate to `true`. +/// Evaluation errors are logged as warnings and treated as non-matching. +pub async fn match_event( + event: &nostr::Event, + channel_id: uuid::Uuid, + rules: &[SubscriptionRule], + agent_pubkey_hex: &str, +) -> Option { + let filter_ctx = FilterContext::from_event(event, channel_id); + + for (index, rule) in rules.iter().enumerate() { + // 1. Channel scope check. + if !rule.channels.matches(&channel_id) { + continue; + } + + // 2. Kind filter (empty = wildcard). + if !rule.kinds.is_empty() && !rule.kinds.contains(&(event.kind.as_u16() as u32)) { + continue; + } + + // 3. Mention check — look for a `p` tag whose value equals agent_pubkey_hex. + if rule.require_mention { + let mentioned = event.tags.iter().any(|tag| { + tag.kind().to_string() == "p" && (tag.content() == Some(agent_pubkey_hex)) + }); + if !mentioned { + continue; + } + } + + // 4. Optional evalexpr filter expression. + if let Some(expr) = &rule.filter { + match evaluate_filter(expr, &filter_ctx).await { + Ok(true) => {} + Ok(false) => continue, + Err(e) => { + warn!(rule = %rule.name, error = %e, "filter expression error; skipping rule"); + continue; + } + } + } + + // All checks passed — this rule wins. + let prompt_tag = rule.prompt_tag.clone().unwrap_or_else(|| rule.name.clone()); + + return Some(MatchedRule { + rule_index: index, + prompt_tag, + }); + } + + None +} + +// ── Tests ───────────────────────────────────────────────────────────────────── + +#[cfg(test)] +mod tests { + use super::*; + use nostr::{EventBuilder, Keys, Kind, Tag}; + use uuid::Uuid; + + /// Build a minimal test event with the given kind and content. + fn make_event(kind: u32, content: &str) -> nostr::Event { + let keys = Keys::generate(); + EventBuilder::new(Kind::Custom(kind as u16), content, []) + .sign_with_keys(&keys) + .unwrap() + } + + /// Build a test event with an explicit `p` tag. + fn make_event_with_p_tag(kind: u32, content: &str, p_hex: &str) -> nostr::Event { + let keys = Keys::generate(); + let p_tag = Tag::parse(&["p", p_hex]).expect("tag parse"); + EventBuilder::new(Kind::Custom(kind as u16), content, [p_tag]) + .sign_with_keys(&keys) + .unwrap() + } + + fn any_channel() -> Uuid { + Uuid::new_v4() + } + + // ── FilterContext ───────────────────────────────────────────────────────── + + #[test] + fn test_filter_context_from_event() { + let event = make_event(40001, "hello world"); + let channel_id = any_channel(); + let ctx = FilterContext::from_event(&event, channel_id); + + assert_eq!(ctx.content, "hello world"); + assert_eq!(ctx.author, event.pubkey.to_hex()); + assert_eq!(ctx.kind, 40001); + assert_eq!(ctx.channel_id, channel_id.to_string()); + assert_eq!(ctx.timestamp, event.created_at.as_u64()); + } + + // ── evaluate_filter ─────────────────────────────────────────────────────── + + #[tokio::test] + async fn test_evaluate_filter_str_contains() { + let event = make_event(40001, "P1 incident in production"); + let ctx = FilterContext::from_event(&event, any_channel()); + + let result = evaluate_filter(r#"str_contains(content, "P1")"#, &ctx) + .await + .unwrap(); + assert!(result); + + let result = evaluate_filter(r#"str_contains(content, "P2")"#, &ctx) + .await + .unwrap(); + assert!(!result); + } + + #[tokio::test] + async fn test_evaluate_filter_kind_check() { + let event = make_event(40001, "some content"); + let ctx = FilterContext::from_event(&event, any_channel()); + + let result = evaluate_filter("kind == 40001", &ctx).await.unwrap(); + assert!(result); + + let result = evaluate_filter("kind == 1", &ctx).await.unwrap(); + assert!(!result); + } + + #[tokio::test] + async fn test_evaluate_filter_too_long() { + let event = make_event(40001, "content"); + let ctx = FilterContext::from_event(&event, any_channel()); + + let long_expr = "a".repeat(MAX_EXPR_LEN + 1); + let err = evaluate_filter(&long_expr, &ctx).await.unwrap_err(); + + assert!(matches!( + err, + FilterError::ExpressionTooLong { len, max } + if len == MAX_EXPR_LEN + 1 && max == MAX_EXPR_LEN + )); + } + + // ── match_event ─────────────────────────────────────────────────────────── + + #[tokio::test] + async fn test_match_event_first_match_wins() { + let event = make_event(40001, "hello"); + let channel_id = any_channel(); + + let rules = vec![ + SubscriptionRule { + name: "first".into(), + channels: ChannelScope::All("all".into()), + kinds: vec![], + require_mention: false, + filter: None, + prompt_tag: Some("tag-first".into()), + }, + SubscriptionRule { + name: "second".into(), + channels: ChannelScope::All("all".into()), + kinds: vec![], + require_mention: false, + filter: None, + prompt_tag: Some("tag-second".into()), + }, + ]; + + let matched = match_event(&event, channel_id, &rules, "").await.unwrap(); + assert_eq!(matched.rule_index, 0); + assert_eq!(matched.prompt_tag, "tag-first"); + } + + #[tokio::test] + async fn test_match_event_kind_filter() { + let event = make_event(40001, "hello"); + let channel_id = any_channel(); + + let rules = vec![ + SubscriptionRule { + name: "wrong-kind".into(), + channels: ChannelScope::All("all".into()), + kinds: vec![1], + require_mention: false, + filter: None, + prompt_tag: None, + }, + SubscriptionRule { + name: "right-kind".into(), + channels: ChannelScope::All("all".into()), + kinds: vec![40001], + require_mention: false, + filter: None, + prompt_tag: Some("matched".into()), + }, + ]; + + let matched = match_event(&event, channel_id, &rules, "").await.unwrap(); + assert_eq!(matched.rule_index, 1); + assert_eq!(matched.prompt_tag, "matched"); + } + + #[tokio::test] + async fn test_match_event_require_mention() { + let agent_pubkey = "deadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeef"; + + let event_no_mention = make_event(40001, "hello"); + let event_with_mention = make_event_with_p_tag(40001, "hello", agent_pubkey); + let channel_id = any_channel(); + + let rules = vec![SubscriptionRule { + name: "mention-only".into(), + channels: ChannelScope::All("all".into()), + kinds: vec![], + require_mention: true, + filter: None, + prompt_tag: Some("mentioned".into()), + }]; + + // Without mention — no match. + let result = match_event(&event_no_mention, channel_id, &rules, agent_pubkey).await; + assert!(result.is_none()); + + // With mention — matches. + let matched = match_event(&event_with_mention, channel_id, &rules, agent_pubkey) + .await + .unwrap(); + assert_eq!(matched.prompt_tag, "mentioned"); + } + + #[tokio::test] + async fn test_match_event_no_match() { + let event = make_event(1, "hello"); + let channel_id = any_channel(); + + let rules = vec![SubscriptionRule { + name: "kind-40001-only".into(), + channels: ChannelScope::All("all".into()), + kinds: vec![40001], + require_mention: false, + filter: None, + prompt_tag: None, + }]; + + let result = match_event(&event, channel_id, &rules, "").await; + assert!(result.is_none()); + } + + // ── ChannelScope ────────────────────────────────────────────────────────── + + #[test] + fn test_channel_scope_all() { + let scope = ChannelScope::All("all".into()); + assert!(scope.matches(&Uuid::new_v4())); + assert!(scope.matches(&Uuid::new_v4())); + } + + #[test] + fn test_channel_scope_all_invalid_string() { + // Only the literal "all" should match; other strings must not. + let scope = ChannelScope::All("ALL".into()); + assert!(!scope.matches(&Uuid::new_v4())); + + let scope = ChannelScope::All("".into()); + assert!(!scope.matches(&Uuid::new_v4())); + } + + #[test] + fn test_channel_scope_list() { + let id_a = Uuid::new_v4(); + let id_b = Uuid::new_v4(); + let id_c = Uuid::new_v4(); + + let scope = ChannelScope::List(vec![id_a.to_string(), id_b.to_string()]); + + assert!(scope.matches(&id_a)); + assert!(scope.matches(&id_b)); + assert!(!scope.matches(&id_c)); + } + + // ── prompt_tag fallback ─────────────────────────────────────────────────── + + #[tokio::test] + async fn test_prompt_tag_falls_back_to_name() { + let event = make_event(40001, "hello"); + let channel_id = any_channel(); + + let rules = vec![SubscriptionRule { + name: "my-rule".into(), + channels: ChannelScope::All("all".into()), + kinds: vec![], + require_mention: false, + filter: None, + prompt_tag: None, // no explicit tag + }]; + + let matched = match_event(&event, channel_id, &rules, "").await.unwrap(); + assert_eq!(matched.prompt_tag, "my-rule"); + } +} diff --git a/crates/sprout-acp/src/main.rs b/crates/sprout-acp/src/main.rs index 0907c0889..6d5b4eac0 100644 --- a/crates/sprout-acp/src/main.rs +++ b/crates/sprout-acp/src/main.rs @@ -2,6 +2,7 @@ mod acp; mod config; +mod filter; mod queue; mod relay; @@ -15,9 +16,13 @@ use tracing_subscriber::EnvFilter; use uuid::Uuid; use acp::{AcpClient, AcpError, EnvVar, McpServer, StopReason}; -use config::Config; +use config::{Config, DedupMode, SubscribeMode}; +use filter::SubscriptionRule; use queue::{EventQueue, QueuedEvent}; use relay::HarnessRelay; +use sprout_core::kind::{ + KIND_STREAM_MESSAGE, KIND_STREAM_REMINDER, KIND_WORKFLOW_APPROVAL_REQUESTED, +}; #[tokio::main] async fn main() -> Result<()> { @@ -28,7 +33,7 @@ async fn main() -> Result<()> { .compact() .init(); - let config = Config::from_env().map_err(|e| anyhow::anyhow!("configuration error: {e}"))?; + let config = Config::from_cli().map_err(|e| anyhow::anyhow!("configuration error: {e}"))?; tracing::info!("sprout-acp starting: {}", config.summary()); // ── Step 1: Spawn ACP agent subprocess and initialize ───────────────────── @@ -47,7 +52,7 @@ async fn main() -> Result<()> { tracing::info!("connected to relay at {}", config.relay_url); - // ── Step 3: Discover channels and subscribe ─────────────────────────────── + // ── Step 3: Discover channels and build subscription rules ──────────────── let channels = relay .discover_channels() .await @@ -55,17 +60,63 @@ async fn main() -> Result<()> { tracing::info!("discovered {} channel(s)", channels.len()); - for channel_id in &channels { - if let Err(e) = relay.subscribe_channel(*channel_id).await { + // Build subscription rules from the configured mode. + let rules: Vec = match config.subscribe_mode { + SubscribeMode::Mentions => { + vec![SubscriptionRule { + name: "mentions".into(), + channels: filter::ChannelScope::All("all".into()), + kinds: config.kinds_override.clone().unwrap_or_else(|| { + vec![ + KIND_STREAM_MESSAGE, + KIND_WORKFLOW_APPROVAL_REQUESTED, + KIND_STREAM_REMINDER, + ] + }), + require_mention: !config.no_mention_filter, + filter: None, + prompt_tag: Some("@mention".into()), + }] + } + SubscribeMode::All => { + vec![SubscriptionRule { + name: "all".into(), + channels: filter::ChannelScope::All("all".into()), + kinds: config.kinds_override.clone().unwrap_or_default(), + require_mention: false, + filter: None, + prompt_tag: Some("all".into()), + }] + } + SubscribeMode::Config => { + let loaded = config::load_rules(&config.config_path)?; + if loaded.is_empty() { + tracing::warn!( + "config file {} contains zero rules — agent will receive no events", + config.config_path.display() + ); + } + loaded + } + }; + + // ── Step 4: Subscribe to channels ──────────────────────────────────────── + let channel_filters = config::resolve_channel_filters(&config, &channels, &rules); + if channel_filters.is_empty() { + tracing::warn!("no channel subscriptions resolved — agent will sit idle"); + } + for (channel_id, filter) in &channel_filters { + if let Err(e) = relay.subscribe_channel(*channel_id, filter.clone()).await { tracing::warn!("failed to subscribe to channel {channel_id}: {e}"); } else { tracing::info!("subscribed to channel {channel_id}"); } } - // ── Step 4: Main orchestration loop ────────────────────────────────────── + // ── Step 5: Main orchestration loop ────────────────────────────────────── let mut sessions: HashMap = HashMap::new(); - let mut queue = EventQueue::new(); + let dedup_mode = config.dedup_mode; + let mut queue = EventQueue::new(dedup_mode); let mcp_servers = build_mcp_servers(&config); let turn_timeout = Duration::from_secs(config.turn_timeout_secs); @@ -81,11 +132,42 @@ async fn main() -> Result<()> { match sprout_event { Some(sprout_event) => { + // Self-event filter: drop events authored by this agent. + if config.ignore_self && sprout_event.event.pubkey.to_hex() == pubkey_hex { + tracing::debug!( + channel_id = %sprout_event.channel_id, + "dropping self-authored event" + ); + continue; + } + + // Rule match: find first matching rule. + let matched = filter::match_event( + &sprout_event.event, + sprout_event.channel_id, + &rules, + &pubkey_hex, + ) + .await; + + let prompt_tag = match matched { + Some(m) => m.prompt_tag, + None => { + tracing::debug!( + channel_id = %sprout_event.channel_id, + kind = sprout_event.event.kind.as_u16(), + "event matched no rule — dropping" + ); + continue; + } + }; + // Push event into the queue. queue.push(QueuedEvent { channel_id: sprout_event.channel_id, event: sprout_event.event, received_at: std::time::Instant::now(), + prompt_tag, }); // Try to flush and process batches. @@ -97,7 +179,7 @@ async fn main() -> Result<()> { let channel_id = batch.channel_id; // Format prompt before potentially requeuing (borrows batch by ref). - let prompt_text = queue::format_prompt(&batch); + let prompt_text = queue::format_prompt(&batch, config.system_prompt.as_deref()); // Get or create session for this channel. let session_id = match get_or_create_session( @@ -105,15 +187,30 @@ async fn main() -> Result<()> { channel_id, &mut acp, &mcp_servers, + &config, ) .await { Ok(id) => id, + Err(AcpError::AgentExited) => { + tracing::error!("agent exited during session setup — respawning"); + sessions.clear(); + match dedup_mode { + DedupMode::Queue => queue.requeue(batch), + DedupMode::Drop => { /* discard */ } + } + queue.mark_complete(); + acp = respawn_agent(&config).await?; + break; + } Err(e) => { tracing::error!( "failed to create session for channel {channel_id}: {e}" ); - queue.requeue(batch); + match dedup_mode { + DedupMode::Queue => queue.requeue(batch), + DedupMode::Drop => { /* discard */ } + } queue.mark_complete(); break; } @@ -136,14 +233,20 @@ async fn main() -> Result<()> { Ok(Err(AcpError::AgentExited)) => { tracing::error!("agent process exited — respawning"); sessions.clear(); - queue.requeue(batch); + match dedup_mode { + DedupMode::Queue => queue.requeue(batch), + DedupMode::Drop => { /* discard */ } + } queue.mark_complete(); acp = respawn_agent(&config).await?; break; } Ok(Err(e)) => { tracing::error!("session_prompt error for channel {channel_id}: {e}"); - queue.requeue(batch); + match dedup_mode { + DedupMode::Queue => queue.requeue(batch), + DedupMode::Drop => { /* discard */ } + } queue.mark_complete(); sessions.remove(&channel_id); break; @@ -163,7 +266,10 @@ async fn main() -> Result<()> { acp = respawn_agent(&config).await?; } Err(e) => { - tracing::error!("cancel_with_cleanup error: {e}"); + tracing::error!( + "cancel_with_cleanup error for channel {channel_id}: {e} — invalidating session" + ); + sessions.remove(&channel_id); } } queue.mark_complete(); @@ -173,19 +279,16 @@ async fn main() -> Result<()> { } } None => { - // Relay connection lost — reconnect. - tracing::warn!("relay connection lost — reconnecting"); - loop { - match relay.reconnect().await { - Ok(()) => { - tracing::info!("relay reconnected"); - break; - } - Err(e) => { - tracing::error!("relay reconnect failed: {e} — retrying in 5s"); - tokio::time::sleep(Duration::from_secs(5)).await; - } - } + // Relay event stream ended — request background reconnect. + // The background task handles the actual reconnection and + // resubscription asynchronously; we just resume waiting for events. + tracing::warn!("relay event stream ended — requesting reconnect"); + if let Err(e) = relay.reconnect().await { + // Background task is dead (cmd_tx closed). Sleep to prevent + // a CPU-burning spin loop, then exit — no recovery possible. + tracing::error!("relay background task is gone: {e} — exiting"); + tokio::time::sleep(Duration::from_secs(1)).await; + break; } } } @@ -228,11 +331,19 @@ async fn spawn_and_init(config: &Config) -> Result { // ── Helper: get or create session for a channel ─────────────────────────────── +/// Get or create an ACP session for the given channel. +/// +/// If a session already exists, returns it immediately. Otherwise creates a new +/// session and — if `config.initial_message` is set — sends it as the first +/// prompt before returning. The initial-message turn counts against +/// `in_flight_channel`, so in `drop` mode events for this channel are dropped +/// while it runs. async fn get_or_create_session( sessions: &mut HashMap, channel_id: Uuid, acp: &mut AcpClient, mcp_servers: &[McpServer], + config: &Config, ) -> Result { if let Some(session_id) = sessions.get(&channel_id) { return Ok(session_id.clone()); @@ -246,6 +357,56 @@ async fn get_or_create_session( let session_id = acp.session_new(&cwd, mcp_servers.to_vec()).await?; tracing::info!("created session {session_id} for channel {channel_id}"); sessions.insert(channel_id, session_id.clone()); + + // Send initial message if configured. + if let Some(ref initial_message) = config.initial_message { + tracing::info!("sending initial message to session {session_id} for channel {channel_id}"); + let turn_timeout = Duration::from_secs(config.turn_timeout_secs); + let result = timeout( + turn_timeout, + acp.session_prompt(&session_id, initial_message), + ) + .await; + match result { + Ok(Ok(stop_reason)) => { + tracing::info!( + "initial message complete for channel {channel_id}: {stop_reason:?}" + ); + } + Ok(Err(e)) => { + tracing::error!( + "initial message failed for channel {channel_id}: {e} — invalidating session" + ); + sessions.remove(&channel_id); + return Err(e); + } + Err(_elapsed) => { + tracing::warn!( + "initial message timed out for channel {channel_id} — cancelling and invalidating session" + ); + // Cancel the in-flight prompt to keep the NDJSON stream in sync, + // matching the cleanup path used by the main event loop. + match acp.cancel_with_cleanup(&session_id).await { + Ok(_) => { + // Agent is still alive — just this session timed out. + sessions.remove(&channel_id); + return Err(AcpError::Timeout); + } + Err(AcpError::AgentExited) => { + // Agent actually died during cancel. + sessions.remove(&channel_id); + return Err(AcpError::AgentExited); + } + Err(cancel_err) => { + tracing::error!("cancel_with_cleanup failed during initial message timeout: {cancel_err}"); + sessions.remove(&channel_id); + return Err(AcpError::Timeout); + } + } + } + } + } + Ok(session_id) } @@ -264,7 +425,11 @@ fn build_mcp_servers(config: &Config) -> Vec { }, EnvVar { name: "SPROUT_PRIVATE_KEY".into(), - value: config.keys.secret_key().to_bech32().unwrap_or_default(), + value: config + .keys + .secret_key() + .to_bech32() + .expect("secret key bech32 encoding should never fail"), }, ]; if let Some(ref token) = config.api_token { diff --git a/crates/sprout-acp/src/queue.rs b/crates/sprout-acp/src/queue.rs index c97e06ff8..732dc800c 100644 --- a/crates/sprout-acp/src/queue.rs +++ b/crates/sprout-acp/src/queue.rs @@ -4,68 +4,107 @@ //! When the harness is ready to prompt the agent, it flushes the channel with //! the oldest pending event, draining ALL events for that channel into a single //! batch. Only one `session/prompt` is in flight at a time across all channels. +//! +//! ## Dedup modes +//! +//! - **Drop** (default) — while a prompt is in-flight for channel C, new events +//! for channel C are silently dropped (debug-logged). Events for other channels +//! still queue normally. +//! - **Queue** — all events accumulate; batched on the next flush cycle. use nostr::{Event, ToBech32}; use std::collections::{HashMap, VecDeque}; use std::time::Instant; use uuid::Uuid; +use crate::config::DedupMode; + +// ── Types ───────────────────────────────────────────────────────────────────── + /// An event waiting in the queue. #[derive(Debug, Clone)] pub struct QueuedEvent { pub channel_id: Uuid, pub event: Event, pub received_at: Instant, + /// Tag identifying which rule (or mode) matched this event. + pub prompt_tag: String, +} + +/// A single event inside a [`FlushBatch`]. +#[derive(Debug)] +pub struct BatchEvent { + pub event: Event, + pub prompt_tag: String, } /// A batch of events to prompt the agent with. #[derive(Debug)] pub struct FlushBatch { pub channel_id: Uuid, - pub events: Vec, + pub events: Vec, } +// ── EventQueue ──────────────────────────────────────────────────────────────── + /// Per-channel event queue with global one-in-flight enforcement. /// /// # State Machine /// /// ```text /// State: -/// queues: Map> -/// prompt_in_flight: bool +/// queues: Map> +/// in_flight_channel: Option +/// dedup_mode: DedupMode /// /// Transitions: /// push(event): -/// queues[event.channel_id].push_back(event) +/// if dedup_mode == Drop AND in_flight_channel == Some(event.channel_id): +/// debug log + discard +/// else: +/// queues[event.channel_id].push_back(event) /// /// flush_next() → Option: -/// if prompt_in_flight: return None +/// if in_flight_channel.is_some(): return None /// if all queues empty: return None /// channel = pick channel with oldest head event (min received_at) /// events = drain queues[channel] -/// remove queues[channel] if now empty -/// prompt_in_flight = true +/// in_flight_channel = Some(channel) /// return Some(FlushBatch { channel, events }) /// /// mark_complete(): -/// prompt_in_flight = false +/// in_flight_channel = None /// ``` pub struct EventQueue { queues: HashMap>, - prompt_in_flight: bool, + in_flight_channel: Option, + dedup_mode: DedupMode, } impl EventQueue { - /// Create a new empty event queue. - pub fn new() -> Self { + /// Create a new empty event queue with the given dedup mode. + pub fn new(dedup_mode: DedupMode) -> Self { Self { queues: HashMap::new(), - prompt_in_flight: false, + in_flight_channel: None, + dedup_mode, } } /// Push an event into the queue for its channel. + /// + /// In [`DedupMode::Drop`], events for the currently in-flight channel are + /// silently discarded (debug-logged). pub fn push(&mut self, event: QueuedEvent) { + if matches!(self.dedup_mode, DedupMode::Drop) + && self.in_flight_channel == Some(event.channel_id) + { + tracing::debug!( + channel_id = %event.channel_id, + "dropping event for in-flight channel (drop mode)" + ); + return; + } self.queues .entry(event.channel_id) .or_default() @@ -77,9 +116,9 @@ impl EventQueue { /// Returns `None` if a prompt is already in flight or if all queues are /// empty. Otherwise picks the channel with the oldest pending event (FIFO /// fairness across channels), drains ALL events for that channel into a - /// single batch, sets `prompt_in_flight = true`, and returns the batch. + /// single batch, sets `in_flight_channel`, and returns the batch. pub fn flush_next(&mut self) -> Option { - if self.prompt_in_flight { + if self.in_flight_channel.is_some() { return None; } @@ -93,16 +132,22 @@ impl EventQueue { // Drain ALL events for that channel. let queue = self.queues.remove(&channel_id)?; - let events: Vec = queue.into_iter().map(|qe| qe.event).collect(); + let events: Vec = queue + .into_iter() + .map(|qe| BatchEvent { + event: qe.event, + prompt_tag: qe.prompt_tag, + }) + .collect(); - self.prompt_in_flight = true; + self.in_flight_channel = Some(channel_id); Some(FlushBatch { channel_id, events }) } - /// Mark the current prompt as complete. Clears `prompt_in_flight`. + /// Mark the current prompt as complete. Clears `in_flight_channel`. pub fn mark_complete(&mut self) { - self.prompt_in_flight = false; + self.in_flight_channel = None; } /// Re-queue a batch of events that failed to process. @@ -117,10 +162,11 @@ impl EventQueue { pub fn requeue(&mut self, batch: FlushBatch) { let queue = self.queues.entry(batch.channel_id).or_default(); // Push to front in reverse order so original order is preserved. - for event in batch.events.into_iter().rev() { + for be in batch.events.into_iter().rev() { queue.push_front(QueuedEvent { channel_id: batch.channel_id, - event, + event: be.event, + prompt_tag: be.prompt_tag, received_at: Instant::now(), }); } @@ -129,7 +175,7 @@ impl EventQueue { /// Whether a prompt is currently in flight. #[allow(dead_code)] pub fn is_in_flight(&self) -> bool { - self.prompt_in_flight + self.in_flight_channel.is_some() } /// Total number of pending events across all channels. @@ -147,43 +193,97 @@ impl EventQueue { impl Default for EventQueue { fn default() -> Self { - Self::new() + Self::new(DedupMode::Drop) } } -/// Format the Channel/From/Time/Message lines for a single event. -fn format_event_lines(channel_id: Uuid, event: &Event) -> String { - format!( - "Channel: {}\nFrom: {}\nTime: {}\nMessage: {}", - channel_id, - event - .pubkey - .to_bech32() - .unwrap_or_else(|_| event.pubkey.to_hex()), - chrono::DateTime::from_timestamp(event.created_at.as_u64() as i64, 0) - .map(|dt| dt.to_rfc3339()) - .unwrap_or_else(|| event.created_at.as_u64().to_string()), - event.content, - ) +// ── Prompt formatting ───────────────────────────────────────────────────────── + +/// Stream-message kinds — these get the compact format (no raw Tags line). +const STREAM_MESSAGE_KINDS: &[u32] = &[ + sprout_core::kind::KIND_STREAM_MESSAGE, + sprout_core::kind::KIND_STREAM_MESSAGE_V2, +]; + +/// Format the per-event block lines for a single [`BatchEvent`]. +/// +/// Non-stream-message kinds (anything not in `[40001, 40002]`) include a +/// `Tags:` line with the raw Nostr tags serialised as a JSON array-of-arrays. +fn format_event_block(channel_id: Uuid, be: &BatchEvent) -> String { + let npub = be + .event + .pubkey + .to_bech32() + .unwrap_or_else(|_| be.event.pubkey.to_hex()); + + let time = chrono::DateTime::from_timestamp(be.event.created_at.as_u64() as i64, 0) + .map(|dt| dt.to_rfc3339()) + .unwrap_or_else(|| be.event.created_at.as_u64().to_string()); + + let kind = be.event.kind.as_u16() as u32; + + let mut block = format!( + "Channel: {channel_id}\nKind: {kind}\nFrom: {npub}\nTime: {time}\nContent: {}", + be.event.content, + ); + + // Include raw tags for non-stream-message kinds. + if !STREAM_MESSAGE_KINDS.contains(&kind) { + let tags_json: Vec<&[String]> = be.event.tags.iter().map(|t| t.as_slice()).collect(); + if let Ok(tags_str) = serde_json::to_string(&tags_json) { + block.push_str(&format!("\nTags: {tags_str}")); + } + } + + block } -/// Format a batch of events into a prompt string for the agent. -pub fn format_prompt(batch: &FlushBatch) -> String { - if batch.events.len() == 1 { +/// Format a [`FlushBatch`] into a prompt string for the agent. +/// +/// If `system_prompt` is `Some`, it is prepended as a `[System]` block. +/// +/// Single-event format: +/// ```text +/// [Sprout event: ] +/// Channel: ... +/// Kind: ... +/// From: ... +/// Time: ... +/// Content: ... +/// ``` +/// +/// Batch format (N > 1): +/// ```text +/// [Sprout events — N events] +/// +/// --- Event 1 () --- +/// Channel: ... +/// ... +/// ``` +pub fn format_prompt(batch: &FlushBatch, system_prompt: Option<&str>) -> String { + let body = if batch.events.len() == 1 { + let be = &batch.events[0]; format!( - "[Sprout @mention]\n{}", - format_event_lines(batch.channel_id, &batch.events[0]) + "[Sprout event: {}]\n{}", + be.prompt_tag, + format_event_block(batch.channel_id, be) ) } else { - let mut prompt = format!("[Sprout @mention — {} events]\n", batch.events.len()); - for (i, event) in batch.events.iter().enumerate() { - prompt.push_str(&format!( - "\n--- Event {} ---\n{}\n", + let mut s = format!("[Sprout events — {} events]", batch.events.len()); + for (i, be) in batch.events.iter().enumerate() { + s.push_str(&format!( + "\n\n--- Event {} ({}) ---\n{}", i + 1, - format_event_lines(batch.channel_id, event) + be.prompt_tag, + format_event_block(batch.channel_id, be) )); } - prompt + s + }; + + match system_prompt { + Some(sp) => format!("[System]\n{sp}\n\n{body}"), + None => body, } } @@ -195,7 +295,7 @@ mod tests { use nostr::{EventBuilder, Keys, Kind}; use std::time::Duration; - /// Build a test event with the given content. + /// Build a test event with the given content and kind. fn make_event(content: &str) -> Event { let keys = Keys::generate(); EventBuilder::new(Kind::Custom(40001), content, []) @@ -209,6 +309,7 @@ mod tests { channel_id, event: make_event(content), received_at: Instant::now(), + prompt_tag: "test".into(), } } @@ -218,6 +319,7 @@ mod tests { channel_id, event: make_event(content), received_at: Instant::now() - age, + prompt_tag: "test".into(), } } @@ -225,7 +327,7 @@ mod tests { #[test] fn test_push_flush_basic() { - let mut q = EventQueue::new(); + let mut q = EventQueue::new(DedupMode::Queue); let ch = Uuid::new_v4(); q.push(make_queued(ch, "hello")); @@ -233,7 +335,7 @@ mod tests { let batch = q.flush_next().expect("should return a batch"); assert_eq!(batch.channel_id, ch); assert_eq!(batch.events.len(), 1); - assert_eq!(batch.events[0].content, "hello"); + assert_eq!(batch.events[0].event.content, "hello"); // Queue should be empty now. assert_eq!(q.pending_count(), 0); @@ -244,7 +346,7 @@ mod tests { #[test] fn test_in_flight_blocks_flush() { - let mut q = EventQueue::new(); + let mut q = EventQueue::new(DedupMode::Queue); let ch = Uuid::new_v4(); q.push(make_queued(ch, "first")); @@ -262,7 +364,7 @@ mod tests { #[test] fn test_mark_complete_enables_flush() { - let mut q = EventQueue::new(); + let mut q = EventQueue::new(DedupMode::Queue); let ch = Uuid::new_v4(); q.push(make_queued(ch, "first")); @@ -280,14 +382,14 @@ mod tests { let batch = q.flush_next().expect("should flush after mark_complete"); assert_eq!(batch.channel_id, ch); assert_eq!(batch.events.len(), 1); - assert_eq!(batch.events[0].content, "second"); + assert_eq!(batch.events[0].event.content, "second"); } // ── Test 4: batch drain ─────────────────────────────────────────────────── #[test] fn test_batch_drain_all_events() { - let mut q = EventQueue::new(); + let mut q = EventQueue::new(DedupMode::Queue); let ch = Uuid::new_v4(); q.push(make_queued(ch, "msg1")); @@ -299,9 +401,9 @@ mod tests { let batch = q.flush_next().expect("should return batch"); assert_eq!(batch.channel_id, ch); assert_eq!(batch.events.len(), 3); - assert_eq!(batch.events[0].content, "msg1"); - assert_eq!(batch.events[1].content, "msg2"); - assert_eq!(batch.events[2].content, "msg3"); + assert_eq!(batch.events[0].event.content, "msg1"); + assert_eq!(batch.events[1].event.content, "msg2"); + assert_eq!(batch.events[2].event.content, "msg3"); // All drained. assert_eq!(q.pending_count(), 0); @@ -312,7 +414,7 @@ mod tests { #[test] fn test_fifo_fairness_picks_oldest_channel() { - let mut q = EventQueue::new(); + let mut q = EventQueue::new(DedupMode::Queue); let ch_a = Uuid::new_v4(); let ch_b = Uuid::new_v4(); @@ -323,14 +425,14 @@ mod tests { let batch = q.flush_next().expect("should return batch"); // A is older, so it should be picked first. assert_eq!(batch.channel_id, ch_a); - assert_eq!(batch.events[0].content, "from A"); + assert_eq!(batch.events[0].event.content, "from A"); } // ── Test 6: multi-channel interleave ───────────────────────────────────── #[test] fn test_multi_channel_interleave() { - let mut q = EventQueue::new(); + let mut q = EventQueue::new(DedupMode::Queue); let ch_a = Uuid::new_v4(); let ch_b = Uuid::new_v4(); @@ -352,7 +454,7 @@ mod tests { // Second flush picks B. let batch_b = q.flush_next().expect("second flush"); assert_eq!(batch_b.channel_id, ch_b); - assert_eq!(batch_b.events[0].content, "B-event"); + assert_eq!(batch_b.events[0].event.content, "B-event"); assert_eq!(q.pending_count(), 0); } @@ -361,7 +463,7 @@ mod tests { #[test] fn test_empty_queue_returns_none() { - let mut q = EventQueue::new(); + let mut q = EventQueue::new(DedupMode::Queue); assert!(q.flush_next().is_none()); } @@ -369,7 +471,7 @@ mod tests { #[test] fn test_pending_count() { - let mut q = EventQueue::new(); + let mut q = EventQueue::new(DedupMode::Queue); let ch_a = Uuid::new_v4(); let ch_b = Uuid::new_v4(); @@ -409,15 +511,18 @@ mod tests { let batch = FlushBatch { channel_id: ch, - events: vec![event], + events: vec![BatchEvent { + event, + prompt_tag: "@mention".into(), + }], }; - let prompt = format_prompt(&batch); + let prompt = format_prompt(&batch, None); - assert!(prompt.starts_with("[Sprout @mention]\n")); + assert!(prompt.starts_with("[Sprout event: @mention]\n")); assert!(prompt.contains(&format!("Channel: {}", ch))); assert!(prompt.contains(&format!("From: {}", npub))); - assert!(prompt.contains("Message: Hello @agent")); + assert!(prompt.contains("Content: Hello @agent")); // Should NOT contain "--- Event 1 ---" (that's the multi-event format). assert!(!prompt.contains("--- Event 1 ---")); } @@ -426,7 +531,7 @@ mod tests { #[test] fn test_requeue_preserves_events() { - let mut queue = EventQueue::new(); + let mut queue = EventQueue::new(DedupMode::Queue); let ch = Uuid::new_v4(); queue.push(make_queued(ch, "msg1")); queue.push(make_queued(ch, "msg2")); @@ -442,13 +547,13 @@ mod tests { // Should be able to flush again and get the same events in order. let batch2 = queue.flush_next().unwrap(); assert_eq!(batch2.events.len(), 2); - assert_eq!(batch2.events[0].content, "msg1"); - assert_eq!(batch2.events[1].content, "msg2"); + assert_eq!(batch2.events[0].event.content, "msg1"); + assert_eq!(batch2.events[1].event.content, "msg2"); } #[test] fn test_requeue_interleaves_with_other_channels() { - let mut queue = EventQueue::new(); + let mut queue = EventQueue::new(DedupMode::Queue); let ch_a = Uuid::new_v4(); let ch_b = Uuid::new_v4(); @@ -480,18 +585,31 @@ mod tests { let batch = FlushBatch { channel_id: ch, - events: vec![e1, e2, e3], + events: vec![ + BatchEvent { + event: e1, + prompt_tag: "tag-a".into(), + }, + BatchEvent { + event: e2, + prompt_tag: "tag-b".into(), + }, + BatchEvent { + event: e3, + prompt_tag: "tag-c".into(), + }, + ], }; - let prompt = format_prompt(&batch); + let prompt = format_prompt(&batch, None); - assert!(prompt.starts_with("[Sprout @mention — 3 events]\n")); - assert!(prompt.contains("--- Event 1 ---")); - assert!(prompt.contains("--- Event 2 ---")); - assert!(prompt.contains("--- Event 3 ---")); - assert!(prompt.contains("Message: first message")); - assert!(prompt.contains("Message: second message")); - assert!(prompt.contains("Message: third message")); + assert!(prompt.starts_with("[Sprout events — 3 events]")); + assert!(prompt.contains("--- Event 1 (tag-a) ---")); + assert!(prompt.contains("--- Event 2 (tag-b) ---")); + assert!(prompt.contains("--- Event 3 (tag-c) ---")); + assert!(prompt.contains("Content: first message")); + assert!(prompt.contains("Content: second message")); + assert!(prompt.contains("Content: third message")); // All events reference the same channel. assert_eq!( prompt.matches(&format!("Channel: {}", ch)).count(), @@ -499,4 +617,64 @@ mod tests { "each event block should include the channel id" ); } + + // ── Test 11: system prompt prepended ───────────────────────────────────── + + #[test] + fn test_format_prompt_with_system_prompt() { + let ch = Uuid::new_v4(); + let event = make_event("hello"); + + let batch = FlushBatch { + channel_id: ch, + events: vec![BatchEvent { + event, + prompt_tag: "test".into(), + }], + }; + + let prompt = format_prompt(&batch, Some("You are a triage bot.")); + assert!(prompt.starts_with("[System]\nYou are a triage bot.\n\n[Sprout event: test]\n")); + } + + // ── Test 12: drop mode discards in-flight channel events ───────────────── + + #[test] + fn test_drop_mode_discards_in_flight_events() { + let mut q = EventQueue::new(DedupMode::Drop); + let ch = Uuid::new_v4(); + + q.push(make_queued(ch, "first")); + let _batch = q.flush_next().expect("first flush"); + assert!(q.is_in_flight()); + + // In drop mode, pushing to the in-flight channel should be discarded. + q.push(make_queued(ch, "dropped")); + assert_eq!(q.pending_count(), 0, "event should be dropped"); + + q.mark_complete(); + // Nothing to flush. + assert!(q.flush_next().is_none()); + } + + // ── Test 13: drop mode still queues other channels ──────────────────────── + + #[test] + fn test_drop_mode_queues_other_channels() { + let mut q = EventQueue::new(DedupMode::Drop); + let ch_a = Uuid::new_v4(); + let ch_b = Uuid::new_v4(); + + q.push(make_queued(ch_a, "A-first")); + let _batch = q.flush_next().expect("flush A"); + assert!(q.is_in_flight()); + + // Events for ch_b should still queue. + q.push(make_queued(ch_b, "B-event")); + assert_eq!(q.pending_count(), 1); + + q.mark_complete(); + let batch_b = q.flush_next().expect("flush B"); + assert_eq!(batch_b.channel_id, ch_b); + } } diff --git a/crates/sprout-acp/src/relay.rs b/crates/sprout-acp/src/relay.rs index e5a6a6ba4..15ee9ead6 100644 --- a/crates/sprout-acp/src/relay.rs +++ b/crates/sprout-acp/src/relay.rs @@ -44,9 +44,7 @@ use tokio_tungstenite::{connect_async, tungstenite::Message, MaybeTlsStream, Web use tracing::{debug, info, warn}; use uuid::Uuid; -use sprout_core::kind::{ - KIND_STREAM_MESSAGE, KIND_STREAM_REMINDER, KIND_WORKFLOW_APPROVAL_REQUESTED, -}; +use crate::config::ChannelFilter; // ── Types ───────────────────────────────────────────────────────────────────── @@ -126,8 +124,11 @@ enum RelayMessage { /// Commands sent from `HarnessRelay` to the background WebSocket task. enum RelayCommand { - /// Subscribe to a channel (sends a NIP-01 REQ). - Subscribe { channel_id: Uuid }, + /// Subscribe to a channel (sends a NIP-01 REQ) with the given filter. + Subscribe { + channel_id: Uuid, + filter: ChannelFilter, + }, /// Unsubscribe from a channel (sends a NIP-01 CLOSE). #[allow(dead_code)] Unsubscribe { channel_id: Uuid }, @@ -207,12 +208,11 @@ impl HarnessRelay { Ok(Self { event_rx, cmd_tx, - // SAFETY: default builder with only timeout config cannot fail http: reqwest::Client::builder() .timeout(std::time::Duration::from_secs(10)) .connect_timeout(std::time::Duration::from_secs(5)) .build() - .expect("SAFETY: default builder with only timeout config cannot fail"), + .map_err(|e| RelayError::Http(format!("failed to build HTTP client: {e}")))?, relay_url: relay_url.to_string(), api_token: api_token.map(|t| t.to_string()), keys: keys.clone(), @@ -266,14 +266,17 @@ impl HarnessRelay { Ok(ids) } - /// Subscribe to events in a channel filtered by agent pubkey. + /// Subscribe to events in a channel using the given filter. /// /// Sends a `Subscribe` command to the background task, which issues the - /// NIP-01 `REQ` for kinds 40001, 46010, 40007 tagged with the channel UUID - /// and agent pubkey. Subscription ID is `ch-`. - pub async fn subscribe_channel(&mut self, channel_id: Uuid) -> Result<(), RelayError> { + /// NIP-01 `REQ` built from `filter`. Subscription ID is `ch-`. + pub async fn subscribe_channel( + &mut self, + channel_id: Uuid, + filter: ChannelFilter, + ) -> Result<(), RelayError> { self.cmd_tx - .send(RelayCommand::Subscribe { channel_id }) + .send(RelayCommand::Subscribe { channel_id, filter }) .await .map_err(|_| RelayError::ConnectionClosed)?; debug!("queued subscribe for channel {channel_id}"); @@ -330,6 +333,8 @@ struct BgState { last_seen: HashMap, /// Set of event IDs seen, for deduplication. seen_ids: HashSet, + /// Per-channel filter used on subscribe (for resubscribe after reconnect). + active_filters: HashMap, } impl BgState { @@ -338,6 +343,7 @@ impl BgState { active_subscriptions: HashMap::new(), last_seen: HashMap::new(), seen_ids: HashSet::new(), + active_filters: HashMap::new(), } } @@ -414,6 +420,7 @@ async fn run_background_task( &relay_url, api_token.as_deref(), &agent_pubkey_hex, + false, // skip_drain: wait for Reconnect command ) .await; } @@ -429,6 +436,7 @@ async fn run_background_task( &relay_url, api_token.as_deref(), &agent_pubkey_hex, + false, ) .await; } @@ -443,6 +451,7 @@ async fn run_background_task( &relay_url, api_token.as_deref(), &agent_pubkey_hex, + false, ) .await; } @@ -452,9 +461,10 @@ async fn run_background_task( // ── Command from HarnessRelay ───────────────────────────────────── cmd = cmd_rx.recv() => { match cmd { - Some(RelayCommand::Subscribe { channel_id }) => { - send_subscribe(&mut ws, &state, channel_id, &agent_pubkey_hex, None).await; + Some(RelayCommand::Subscribe { channel_id, filter }) => { + send_subscribe(&mut ws, &state, channel_id, &agent_pubkey_hex, None, &filter).await; state.active_subscriptions.insert(channel_id, channel_sub_id(channel_id)); + state.active_filters.insert(channel_id, filter); } Some(RelayCommand::Unsubscribe { channel_id }) => { if let Some(sub_id) = state.active_subscriptions.remove(&channel_id) { @@ -466,7 +476,7 @@ async fn run_background_task( } } Some(RelayCommand::Reconnect) => { - // Caller already got None from next_event(); drive reconnect now. + // Reconnect command already consumed — skip the drain loop. wait_for_reconnect( &mut ws, &mut cmd_rx, @@ -475,6 +485,7 @@ async fn run_background_task( &relay_url, api_token.as_deref(), &agent_pubkey_hex, + true, // skip_drain: command already consumed ) .await; } @@ -587,9 +598,15 @@ async fn handle_ws_message( } } -/// Wait for a `Reconnect` command from the caller, then attempt reconnection -/// with exponential backoff. Resubscribes all active channels with `since` -/// filters on success. +/// Attempt reconnection with exponential backoff. Resubscribes all active +/// channels with `since` filters on success. +/// +/// If `skip_drain` is `false`, drains the command channel until a `Reconnect` +/// command arrives (used when called from the WS-error path where the caller +/// hasn't sent Reconnect yet). If `true`, skips the drain and reconnects +/// immediately (used when called from the `RelayCommand::Reconnect` arm where +/// the command was already consumed). +#[allow(clippy::too_many_arguments)] async fn wait_for_reconnect( ws: &mut WsStream, cmd_rx: &mut mpsc::Receiver, @@ -598,15 +615,27 @@ async fn wait_for_reconnect( relay_url: &str, api_token: Option<&str>, agent_pubkey_hex: &str, + skip_drain: bool, ) { - // Drain commands until we get Reconnect (or Shutdown). - loop { - match cmd_rx.recv().await { - Some(RelayCommand::Reconnect) => break, - Some(RelayCommand::Shutdown) | None => return, - // Ignore Subscribe/Unsubscribe while disconnected — they'll be - // resubmitted after reconnect via active_subscriptions. - _ => {} + if !skip_drain { + // Drain commands until we get Reconnect (or Shutdown). + loop { + match cmd_rx.recv().await { + Some(RelayCommand::Reconnect) => break, + Some(RelayCommand::Shutdown) | None => return, + // Apply Subscribe/Unsubscribe to state so reconnect reflects + // the latest caller intent (not just pre-disconnect state). + Some(RelayCommand::Subscribe { channel_id, filter }) => { + state + .active_subscriptions + .insert(channel_id, channel_sub_id(channel_id)); + state.active_filters.insert(channel_id, filter); + } + Some(RelayCommand::Unsubscribe { channel_id }) => { + state.active_subscriptions.remove(&channel_id); + state.active_filters.remove(&channel_id); + } + } } } @@ -625,7 +654,14 @@ async fn wait_for_reconnect( info!("resubscribing to {} channel(s)", channels.len()); for channel_id in channels { let since = state.last_seen.get(&channel_id).copied(); - send_subscribe(ws, state, channel_id, agent_pubkey_hex, since).await; + let filter = state.active_filters.get(&channel_id).cloned().unwrap_or( + ChannelFilter { + kinds: None, + require_mention: false, + }, + ); + send_subscribe(ws, state, channel_id, agent_pubkey_hex, since, &filter) + .await; } } @@ -643,32 +679,50 @@ async fn wait_for_reconnect( } } -/// Send a NIP-01 REQ for a channel, optionally with a `since` filter. +/// Send a NIP-01 REQ for a channel, built from a [`ChannelFilter`]. +/// +/// - `kinds` is included only when `filter.kinds` is `Some`; `None` = wildcard. +/// - `#p` is included only when `filter.require_mention` is `true`. +/// - `#h` is always included (channel-scoped subscription). +/// - On first subscribe (`since` is `None`) adds `since=now` to avoid replaying +/// history. On reconnect (`since` is `Some`) subtracts [`SINCE_SKEW_SECS`]. async fn send_subscribe( ws: &mut WsStream, _state: &BgState, channel_id: Uuid, agent_pubkey_hex: &str, since: Option, + filter: &ChannelFilter, ) { let sub_id = channel_sub_id(channel_id); - let mut filter = json!({ - "kinds": [ - KIND_STREAM_MESSAGE, - KIND_WORKFLOW_APPROVAL_REQUESTED, - KIND_STREAM_REMINDER, - ], - "#h": [channel_id.to_string()], - "#p": [agent_pubkey_hex], - }); + let mut req_filter = serde_json::Map::new(); - // Fix 2: Add `since` filter on reconnect (with 5-second skew buffer). - if let Some(ts) = since { - filter["since"] = json!(ts.saturating_sub(SINCE_SKEW_SECS)); + // kinds — omit entirely for wildcard subscriptions. + if let Some(ref kinds) = filter.kinds { + req_filter.insert("kinds".into(), json!(kinds)); } - let req = json!(["REQ", sub_id, filter]); + // #h — always present (channel scope). + req_filter.insert("#h".into(), json!([channel_id.to_string()])); + + // #p — only when require_mention is true. + if filter.require_mention { + req_filter.insert("#p".into(), json!([agent_pubkey_hex])); + } + + // since — on first subscribe use current time to skip history; on reconnect + // subtract skew buffer to catch events missed during the disconnect window. + let since_ts = match since { + Some(ts) => ts.saturating_sub(SINCE_SKEW_SECS), + None => std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs(), + }; + req_filter.insert("since".into(), json!(since_ts)); + + let req = json!(["REQ", sub_id, Value::Object(req_filter)]); match serde_json::to_string(&req) { Ok(text) => { @@ -680,7 +734,7 @@ async fn send_subscribe( if since.is_some() { " (with since filter)" } else { - "" + " (since=now)" } ); } diff --git a/crates/sprout-db/src/lib.rs b/crates/sprout-db/src/lib.rs index f431101e1..b36b52d45 100644 --- a/crates/sprout-db/src/lib.rs +++ b/crates/sprout-db/src/lib.rs @@ -620,6 +620,26 @@ impl Db { reaction::remove_reaction_by_source_event_id(&self.pool, reaction_event_id).await } + /// Backfill the source event ID on an active reaction row. + pub async fn set_reaction_event_id( + &self, + event_id: &[u8], + event_created_at: DateTime, + pubkey: &[u8], + emoji: &str, + reaction_event_id: &[u8], + ) -> Result { + reaction::set_reaction_event_id( + &self.pool, + event_id, + event_created_at, + pubkey, + emoji, + reaction_event_id, + ) + .await + } + /// Look up the active reaction row for one actor + emoji + target tuple. pub async fn get_active_reaction_record( &self, diff --git a/crates/sprout-db/src/reaction.rs b/crates/sprout-db/src/reaction.rs index 2401f493c..e46e2b2ff 100644 --- a/crates/sprout-db/src/reaction.rs +++ b/crates/sprout-db/src/reaction.rs @@ -84,10 +84,11 @@ pub async fn add_reaction( ON DUPLICATE KEY UPDATE created_at = IF(removed_at IS NOT NULL, NOW(6), created_at), removed_at = NULL, - reaction_event_id = CASE - WHEN removed_at IS NOT NULL THEN VALUES(reaction_event_id) - ELSE COALESCE(reaction_event_id, VALUES(reaction_event_id)) - END + -- Keep existing non-NULL event ID if the new value is NULL. + -- REST calls add_reaction(None) first, then backfills via + -- set_reaction_event_id. Without COALESCE, a re-add would + -- clobber the existing event ID with NULL. + reaction_event_id = COALESCE(VALUES(reaction_event_id), reaction_event_id) "#, ) .bind(event_created_at) @@ -191,6 +192,40 @@ pub async fn get_active_reaction_record( .transpose() } +/// Backfill the source event ID on an active reaction row. +/// +/// Called after the kind:7 event is created and stored, to link the +/// reaction row to its source event. Returns `true` if the row was updated. +pub async fn set_reaction_event_id( + pool: &MySqlPool, + event_id: &[u8], + event_created_at: DateTime, + pubkey: &[u8], + emoji: &str, + reaction_event_id: &[u8], +) -> Result { + let result = sqlx::query( + r#" + UPDATE reactions + SET reaction_event_id = ? + WHERE event_created_at = ? + AND event_id = ? + AND pubkey = ? + AND emoji = ? + AND removed_at IS NULL + "#, + ) + .bind(reaction_event_id) + .bind(event_created_at) + .bind(event_id) + .bind(pubkey) + .bind(emoji) + .execute(pool) + .await?; + + Ok(result.rows_affected() > 0) +} + // ── Read operations ─────────────────────────────────────────────────────────── /// Get all active reactions for an event, grouped by emoji. diff --git a/crates/sprout-proxy/Cargo.toml b/crates/sprout-proxy/Cargo.toml index f49dc0149..26d1ef8e5 100644 --- a/crates/sprout-proxy/Cargo.toml +++ b/crates/sprout-proxy/Cargo.toml @@ -20,6 +20,7 @@ thiserror = { workspace = true } sha2 = { workspace = true } hmac = { workspace = true } dashmap = { workspace = true } +moka = { workspace = true } reqwest = { workspace = true } tokio-tungstenite = { workspace = true } futures-util = { workspace = true } diff --git a/crates/sprout-proxy/src/main.rs b/crates/sprout-proxy/src/main.rs index 2f828d560..a293eef0f 100644 --- a/crates/sprout-proxy/src/main.rs +++ b/crates/sprout-proxy/src/main.rs @@ -46,6 +46,14 @@ async fn main() { let server_key_hex = env_required("SPROUT_PROXY_SERVER_KEY"); let salt_hex = env_required("SPROUT_PROXY_SALT"); let api_token = env_required("SPROUT_PROXY_API_TOKEN"); + let relay_pubkey = env_required("SPROUT_RELAY_PUBKEY").to_lowercase(); + // Validate relay pubkey is well-formed 64-char hex at startup. + // Input is lowercased above, so mixed-case is accepted. + if relay_pubkey.len() != 64 || !relay_pubkey.chars().all(|c| c.is_ascii_hexdigit()) { + eprintln!("error: SPROUT_RELAY_PUBKEY must be a 64-character hex string (32 bytes)"); + std::process::exit(1); + } + info!(relay_pubkey = %relay_pubkey, "relay pubkey configured for attribution trust"); // ── Parse server keypair ────────────────────────────────────────────────── @@ -96,6 +104,7 @@ async fn main() { channel_map.clone(), api_base.clone(), api_token.clone(), + relay_pubkey, )); // ── Init guest store (empty — guests registered via POST /admin/guests) ──── diff --git a/crates/sprout-proxy/src/server.rs b/crates/sprout-proxy/src/server.rs index 39da43d1d..cb2ca696f 100644 --- a/crates/sprout-proxy/src/server.rs +++ b/crates/sprout-proxy/src/server.rs @@ -1057,6 +1057,7 @@ mod tests { channel_map.clone(), "http://localhost:3000", "sprout_test", + "deadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeef", )); let upstream = Arc::new(UpstreamClient::new("ws://localhost:3000", "sprout_test")); ProxyState { diff --git a/crates/sprout-proxy/src/translate.rs b/crates/sprout-proxy/src/translate.rs index b303af25d..b90b9fe9c 100644 --- a/crates/sprout-proxy/src/translate.rs +++ b/crates/sprout-proxy/src/translate.rs @@ -15,11 +15,12 @@ //! each external user maps to a consistent shadow pubkey across sessions. use std::sync::Arc; +use std::time::Duration; -use dashmap::DashMap; +use moka::sync::Cache; use nostr::prelude::*; use sprout_core::kind::{ - event_kind_u32, KIND_REACTION, KIND_STREAM_MESSAGE, KIND_STREAM_MESSAGE_EDIT, + event_kind_u32, KIND_DELETION, KIND_REACTION, KIND_STREAM_MESSAGE, KIND_STREAM_MESSAGE_EDIT, KIND_STREAM_MESSAGE_V2, }; use uuid::Uuid; @@ -43,9 +44,12 @@ pub struct Translator { http_client: reqwest::Client, api_base: String, api_token: String, - internal_to_external_event_ids: DashMap, - external_to_internal_event_ids: DashMap, - internal_event_channels: DashMap, + /// Hex-encoded public key of the relay. Only events signed by this key + /// may carry trusted `actor`/`p` attribution tags. + relay_pubkey: String, + internal_to_external_event_ids: Cache, + external_to_internal_event_ids: Cache, + internal_event_channels: Cache, } impl Translator { @@ -55,17 +59,32 @@ impl Translator { channel_map: Arc, api_base: impl Into, api_token: impl Into, + relay_pubkey: impl Into, ) -> Self { + let cache_ttl = Duration::from_secs(3600); Self { kind_translator: KindTranslator::new(), shadow_keys, channel_map, - http_client: reqwest::Client::new(), + http_client: reqwest::Client::builder() + .timeout(Duration::from_secs(5)) + .build() + .expect("HTTP client build must succeed"), api_base: api_base.into(), api_token: api_token.into(), - internal_to_external_event_ids: DashMap::new(), - external_to_internal_event_ids: DashMap::new(), - internal_event_channels: DashMap::new(), + relay_pubkey: relay_pubkey.into(), + internal_to_external_event_ids: Cache::builder() + .max_capacity(100_000) + .time_to_idle(cache_ttl) + .build(), + external_to_internal_event_ids: Cache::builder() + .max_capacity(100_000) + .time_to_idle(cache_ttl) + .build(), + internal_event_channels: Cache::builder() + .max_capacity(100_000) + .time_to_idle(cache_ttl) + .build(), } } } @@ -81,15 +100,11 @@ impl Translator { } fn lookup_external_event_id(&self, internal_event_id: &str) -> Option { - self.internal_to_external_event_ids - .get(internal_event_id) - .map(|value| value.value().clone()) + self.internal_to_external_event_ids.get(internal_event_id) } fn lookup_internal_event_id(&self, external_event_id: &str) -> Option { - self.external_to_internal_event_ids - .get(external_event_id) - .map(|value| value.value().clone()) + self.external_to_internal_event_ids.get(external_event_id) } fn cache_internal_channel(&self, internal_event_id: &str, channel_uuid: Uuid) { @@ -98,12 +113,28 @@ impl Translator { } fn lookup_internal_channel(&self, internal_event_id: &str) -> Option { - self.internal_event_channels - .get(internal_event_id) - .map(|value| *value.value()) + self.internal_event_channels.get(internal_event_id) } fn resolve_shadow_author_hex(&self, event: &Event) -> String { + // Only trust actor/p tags on relay-signed events (API-originated). + // User-signed events could carry spoofed actor/p tags — if the event + // pubkey doesn't match the relay's keypair, fall back to the event + // pubkey directly. + if event.pubkey.to_hex() == self.relay_pubkey { + for tag_name in &["actor", "p"] { + for tag in event.tags.iter() { + let slice = tag.as_slice(); + if slice.len() >= 2 && slice[0] == *tag_name { + let hex = &slice[1]; + if hex.len() == 64 && hex.chars().all(|c| c.is_ascii_hexdigit()) { + // Normalize to lowercase for consistent shadow-key derivation. + return hex.to_lowercase(); + } + } + } + } + } event.pubkey.to_hex() } @@ -247,6 +278,30 @@ impl Translator { &self, event: &Event, allowed_channels: &[Uuid], + ) -> Result, ProxyError> { + self.translate_outbound_tag_targeted(event, allowed_channels, Kind::Reaction) + .await + } + + async fn translate_outbound_deletion( + &self, + event: &Event, + allowed_channels: &[Uuid], + ) -> Result, ProxyError> { + self.translate_outbound_tag_targeted(event, allowed_channels, Kind::EventDeletion) + .await + } + + /// Shared outbound translation for tag-targeted events (reactions, deletions). + /// + /// Resolves the channel via `#h`, translates all internal `#e` event IDs to + /// their external counterparts, strips `#h` and `actor` tags, and re-signs + /// with the shadow key for the original author. + async fn translate_outbound_tag_targeted( + &self, + event: &Event, + allowed_channels: &[Uuid], + kind: Kind, ) -> Result, ProxyError> { let Some((_uuid_str, channel_info)) = self.resolve_channel_info(event, allowed_channels)? else { @@ -280,7 +335,7 @@ impl Translator { parts.extend(slice.iter().skip(2).cloned()); let part_refs: Vec<&str> = parts.iter().map(String::as_str).collect(); new_tags.push(Tag::parse(&part_refs).map_err(|e| { - ProxyError::Upstream(format!("reaction tag build failed: {e}")) + ProxyError::Upstream(format!("outbound tag build failed: {e}")) })?); saw_target = true; continue; @@ -296,7 +351,7 @@ impl Translator { .shadow_keys .get_or_create(&self.resolve_shadow_author_hex(event))?; - let translated = EventBuilder::new(Kind::Reaction, event.content.clone(), new_tags) + let translated = EventBuilder::new(kind, event.content.clone(), new_tags) .custom_created_at(event.created_at) .sign_with_keys(&shadow_keys) .map_err(|e| ProxyError::Upstream(format!("outbound signing failed: {e}")))?; @@ -330,6 +385,11 @@ impl Translator { .translate_outbound_reaction(event, allowed_channels) .await; } + if kind_u32 == KIND_DELETION { + return self + .translate_outbound_deletion(event, allowed_channels) + .await; + } self.translate_outbound_message_like(event, allowed_channels) } @@ -359,7 +419,12 @@ impl Translator { let kind_u32 = event.kind.as_u16() as u32; // Accept kind:42 (channel message), kind:41 (channel metadata edit), - // kind:1 (text note), and kind:7 (reaction). Everything else is rejected. + // kind:1 (text note), and kind:7 (reaction). + // kind:5 (deletion) is intentionally NOT accepted inbound. + // Inbound kind:5 blocked: relay deletion handler lacks author-match authorization. + // External clients could delete any user's messages. Re-enable after adding + // author validation to handle_standard_deletion_event. + // Everything else is rejected. if kind_u32 != 42 && kind_u32 != 41 && kind_u32 != 1 && kind_u32 != 7 { return Err(ProxyError::PermissionDenied(format!( "kind {} not accepted by proxy (expected 1, 7, 41, or 42)", @@ -368,95 +433,13 @@ impl Translator { } if kind_u32 == 7 { - let target_external_id = event - .tags - .iter() - .rev() - .find_map(|tag| { - let slice = tag.as_slice(); - if slice.len() >= 2 - && slice[0] == "e" - && slice[1].len() == 64 - && slice[1].chars().all(|ch| ch.is_ascii_hexdigit()) - { - Some(slice[1].clone()) - } else { - None - } - }) - .ok_or_else(|| { - ProxyError::PermissionDenied( - "kind:7 event must reference a target message via #e".into(), - ) - })?; - - let target_internal_id = self - .lookup_internal_event_id(&target_external_id) - .ok_or_else(|| { - ProxyError::PermissionDenied( - "reaction target is unknown to the proxy; fetch the message first".into(), - ) - })?; - - let Some(channel_uuid) = self.lookup_internal_channel(&target_internal_id) else { - return Err(ProxyError::PermissionDenied( - "reaction target is unknown to the proxy; fetch the message first".into(), - )); - }; - - if !allowed_channels.contains(&channel_uuid) { - return Err(ProxyError::PermissionDenied( - "reaction target channel not in invite scope".into(), - )); - } - - let mut new_tags: Vec = - vec![Tag::parse(&["h", &channel_uuid.to_string()]).expect("h tag is always valid")]; - let mut saw_target = false; - - for tag in event.tags.iter() { - let slice = tag.as_slice(); - if slice.first().map(|value| value.as_str()) == Some("h") { - continue; - } - if slice.len() >= 2 - && slice[0] == "e" - && slice[1].len() == 64 - && slice[1].chars().all(|ch| ch.is_ascii_hexdigit()) - { - let Some(internal_id) = self.lookup_internal_event_id(&slice[1]) else { - return Err(ProxyError::PermissionDenied( - "reaction target is unknown to the proxy; fetch the message first" - .into(), - )); - }; - let mut parts = vec!["e".to_string(), internal_id]; - parts.extend(slice.iter().skip(2).cloned()); - let part_refs: Vec<&str> = parts.iter().map(String::as_str).collect(); - new_tags.push(Tag::parse(&part_refs).map_err(|e| { - ProxyError::Upstream(format!("reaction tag build failed: {e}")) - })?); - saw_target = true; - continue; - } - new_tags.push(tag.clone()); - } - - if !saw_target { - return Err(ProxyError::PermissionDenied( - "kind:7 event must reference a target message via #e".into(), - )); - } - - let shadow_keys = self.shadow_keys.get_or_create(external_pubkey)?; - let translated = EventBuilder::new(Kind::Reaction, &event.content, new_tags) - .custom_created_at(event.created_at) - .sign_with_keys(&shadow_keys) - .map_err(|e| ProxyError::Upstream(format!("inbound signing failed: {e}")))?; - - self.cache_event_mapping(&translated.id.to_hex(), &event.id.to_hex()); - self.cache_internal_channel(&translated.id.to_hex(), channel_uuid); - return Ok(translated); + return self.translate_inbound_tag_targeted( + event, + external_pubkey, + allowed_channels, + Kind::Reaction, + "reaction", + ); } // Find the channel-reference `#e` tag by looking up each `#e` value @@ -536,6 +519,148 @@ impl Translator { self.cache_internal_channel(&translated.id.to_hex(), channel_info.uuid); Ok(translated) } + + /// Shared inbound translation for tag-targeted events (kind:5 deletion, kind:7 reaction). + /// + /// # Security + /// + /// Collects ALL hex `#e` tags and verifies every one resolves to the **same** + /// channel UUID that is in `allowed_channels`. A mixed-target event (targets + /// spanning multiple channels) is rejected outright — this prevents a client + /// from smuggling out-of-scope event IDs inside a single deletion or reaction. + fn translate_inbound_tag_targeted( + &self, + event: &Event, + external_pubkey: &str, + allowed_channels: &[Uuid], + kind: Kind, + label: &str, + ) -> Result { + // Collect all hex #e tag values. + let e_tag_values: Vec = event + .tags + .iter() + .filter_map(|tag| { + let slice = tag.as_slice(); + if slice.len() >= 2 + && slice[0] == "e" + && slice[1].len() == 64 + && slice[1].chars().all(|ch| ch.is_ascii_hexdigit()) + { + Some(slice[1].clone()) + } else { + None + } + }) + .collect(); + + if e_tag_values.is_empty() { + return Err(ProxyError::PermissionDenied(format!( + "kind:{} event must reference a target message via #e", + event.kind.as_u16() + ))); + } + + // Resolve every #e tag to an internal event ID and channel UUID. + // All targets must resolve to the same channel — cross-channel targeting + // is rejected to prevent authorization bypass. + let mut resolved_channel: Option = None; + // Resolve all external → internal ID mappings upfront. We store + // the resolved pairs so the tag-building loop below never re-reads + // the moka cache (which could evict entries between passes). + let mut resolved_ids: Vec<(String, String)> = Vec::with_capacity(e_tag_values.len()); + for external_id in &e_tag_values { + let internal_id = self.lookup_internal_event_id(external_id).ok_or_else(|| { + ProxyError::PermissionDenied(format!( + "{label} target is unknown to the proxy; fetch the message first" + )) + })?; + let channel_uuid = self.lookup_internal_channel(&internal_id).ok_or_else(|| { + ProxyError::PermissionDenied(format!( + "{label} target is unknown to the proxy; fetch the message first" + )) + })?; + match resolved_channel { + None => resolved_channel = Some(channel_uuid), + Some(existing) if existing != channel_uuid => { + return Err(ProxyError::PermissionDenied(format!( + "{label} targets span multiple channels — cross-channel targeting rejected" + ))); + } + Some(_) => {} + } + resolved_ids.push((external_id.clone(), internal_id)); + } + let id_map: std::collections::HashMap<&str, &str> = resolved_ids + .iter() + .map(|(ext, int)| (ext.as_str(), int.as_str())) + .collect(); + + let channel_uuid = resolved_channel.expect("e_tag_values non-empty guarantees Some"); + + if !allowed_channels.contains(&channel_uuid) { + return Err(ProxyError::PermissionDenied(format!( + "{label} target channel not in invite scope" + ))); + } + + // Build translated tag list: prepend #h, translate all #e values, drop + // any client-supplied #h tags (prevent unauthorized channel injection). + let mut new_tags: Vec = + vec![Tag::parse(&["h", &channel_uuid.to_string()]).expect("h tag is always valid")]; + let mut saw_target = false; + + for tag in event.tags.iter() { + let slice = tag.as_slice(); + // Strip client-supplied #h tags (authorized #h already added above). + if slice.first().map(|v| v.as_str()) == Some("h") { + continue; + } + // Strip actor tags — untrusted client data should not persist. + if slice.first().map(|v| v.as_str()) == Some("actor") { + continue; + } + if slice.len() >= 2 + && slice[0] == "e" + && slice[1].len() == 64 + && slice[1].chars().all(|ch| ch.is_ascii_hexdigit()) + { + let internal_id = *id_map.get(slice[1].as_str()).ok_or_else(|| { + ProxyError::PermissionDenied(format!( + "{label} target is unknown to the proxy; fetch the message first" + )) + })?; + let mut parts = vec!["e".to_string(), internal_id.to_string()]; + parts.extend(slice.iter().skip(2).cloned()); + let part_refs: Vec<&str> = parts.iter().map(String::as_str).collect(); + new_tags.push( + Tag::parse(&part_refs).map_err(|e| { + ProxyError::Upstream(format!("{label} tag build failed: {e}")) + })?, + ); + saw_target = true; + continue; + } + new_tags.push(tag.clone()); + } + + if !saw_target { + return Err(ProxyError::PermissionDenied(format!( + "kind:{} event must reference a target message via #e", + event.kind.as_u16() + ))); + } + + let shadow_keys = self.shadow_keys.get_or_create(external_pubkey)?; + let translated = EventBuilder::new(kind, &event.content, new_tags) + .custom_created_at(event.created_at) + .sign_with_keys(&shadow_keys) + .map_err(|e| ProxyError::Upstream(format!("inbound signing failed: {e}")))?; + + self.cache_event_mapping(&translated.id.to_hex(), &event.id.to_hex()); + self.cache_internal_channel(&translated.id.to_hex(), channel_uuid); + Ok(translated) + } } // ─── Filter translation ─────────────────────────────────────────────────────── @@ -701,6 +826,7 @@ mod tests { channel_map.clone(), "http://localhost:3000", "sprout_test", + "deadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeef", ); // Retrieve the kind:40 event ID for the test channel. diff --git a/crates/sprout-relay/src/api/reactions.rs b/crates/sprout-relay/src/api/reactions.rs index 09711f5c0..4ebee8af1 100644 --- a/crates/sprout-relay/src/api/reactions.rs +++ b/crates/sprout-relay/src/api/reactions.rs @@ -96,6 +96,12 @@ pub async fn add_reaction_handler( if emoji.is_empty() { return Err(api_error(StatusCode::BAD_REQUEST, "emoji is required")); } + if emoji.chars().count() > 64 { + return Err(api_error( + StatusCode::BAD_REQUEST, + "emoji too long (max 64 characters)", + )); + } let event_id_bytes = decode_event_id(&event_id_hex)?; @@ -128,16 +134,25 @@ pub async fn add_reaction_handler( .single() .unwrap_or_default(); - if state + // Atomically add the reaction. ON DUPLICATE KEY UPDATE returns + // rows_affected=0 if the row is already active — no TOCTOU race. + let added = state .db - .get_active_reaction_record(&event_id_bytes, event_created_at, &pubkey_bytes, &emoji) + .add_reaction( + &event_id_bytes, + event_created_at, + &pubkey_bytes, + &emoji, + None, + ) .await - .map_err(|e| internal_error(&format!("db error: {e}")))? - .is_some() - { + .map_err(|e| internal_error(&format!("db error: {e}")))?; + + if !added { return Err(api_error(StatusCode::CONFLICT, "reaction already exists")); } + // Reaction is now active in DB. Build and store the kind:7 source event. let user_pubkey_hex = nostr_hex::encode(&pubkey_bytes); let mut tags = build_actor_tags(&user_pubkey_hex)?; tags.push( @@ -155,31 +170,60 @@ pub async fn add_reaction_handler( .sign_with_keys(&state.relay_keypair) .map_err(|e| internal_error(&format!("event signing error: {e}")))?; - let event_id = event.id.to_hex(); - let (stored_event, was_inserted) = state - .db - .insert_event(&event, stored.channel_id) - .await - .map_err(|e| internal_error(&format!("db error: {e}")))?; + let reaction_event_id = event.id.to_hex(); + let insert_result = state.db.insert_event(&event, stored.channel_id).await; + + let (stored_event, was_inserted) = match insert_result { + Ok(pair) => pair, + Err(e) => { + // Compensate: deactivate the reaction row so the user can retry. + let _ = state + .db + .remove_reaction(&event_id_bytes, event_created_at, &pubkey_bytes, &emoji) + .await; + return Err(internal_error(&format!("db error: {e}"))); + } + }; if !was_inserted { - return Err(api_error( - StatusCode::CONFLICT, - "reaction event already exists", - )); + // Event ID collision (INSERT IGNORE no-op) — e.g. rapid remove/re-add + // within the same second produces a deterministic event ID that already + // exists as a soft-deleted row. Log and continue; the reaction row is + // still valid, it just won't have a backfilled event ID. + tracing::warn!( + reaction_event_id, + "reaction source event already existed (INSERT IGNORE no-op)" + ); } - crate::handlers::side_effects::handle_side_effects(7, &event, &state) + // Backfill the source event ID on the reaction row. + if let Err(e) = state + .db + .set_reaction_event_id( + &event_id_bytes, + event_created_at, + &pubkey_bytes, + &emoji, + event.id.as_bytes(), + ) .await - .map_err(|e| internal_error(&format!("reaction side effect failed: {e}")))?; + { + tracing::warn!( + reaction_event_id, + "failed to backfill reaction_event_id: {e}" + ); + } + // Dispatch to connected clients. Skip handle_side_effects(7, ...) since + // we already wrote the reaction row above — calling it again would be a + // redundant no-op (ON DUPLICATE KEY UPDATE with same values). let _ = dispatch_persistent_event(&state, &stored_event, 7, &user_pubkey_hex).await; Ok(( StatusCode::CREATED, Json(serde_json::json!({ "added": true, - "event_id": event_id, + "event_id": reaction_event_id, })), )) } @@ -246,8 +290,27 @@ pub async fn remove_reaction_handler( .db .get_event_by_id(&reaction_event_id) .await - .map_err(|e| internal_error(&format!("db error: {e}")))? - .ok_or_else(|| not_found("reaction event not found"))?; + .map_err(|e| internal_error(&format!("db error: {e}")))?; + + // If the source event is missing (e.g. soft-deleted from a prior + // remove/re-add cycle), skip kind:5 creation and fall through to + // direct row removal below. + let reaction_event = match reaction_event { + Some(ev) => ev, + None => { + tracing::warn!( + reaction_event_id = reaction_event_hex, + "reaction source event missing — falling back to direct row removal" + ); + // Direct row removal — no kind:5 event needed. + state + .db + .remove_reaction(&event_id_bytes, event_created_at, &pubkey_bytes, &emoji) + .await + .map_err(|e| internal_error(&format!("db error: {e}")))?; + return Ok(Json(serde_json::json!({ "removed": true }))); + } + }; let user_pubkey_hex = nostr_hex::encode(&pubkey_bytes); let mut tags = build_actor_tags(&user_pubkey_hex)?; @@ -305,7 +368,6 @@ pub async fn remove_reaction_handler( Ok(Json(serde_json::json!({ "removed": true, - "legacy": true, }))) } diff --git a/crates/sprout-workflow/src/executor.rs b/crates/sprout-workflow/src/executor.rs index 829881399..b27f65488 100644 --- a/crates/sprout-workflow/src/executor.rs +++ b/crates/sprout-workflow/src/executor.rs @@ -867,20 +867,30 @@ async fn call_webhook_impl( /// /// This is an internal call — the workflow engine runs inside the relay process, /// so `localhost:3000` is always reachable without SSRF concerns. +/// +/// Returns a shared `reqwest::Client` reused across all workflow HTTP calls. +/// Sharing a single client reuses the underlying connection pool. +#[cfg(feature = "reqwest")] +fn shared_http_client() -> &'static reqwest::Client { + use std::sync::LazyLock; + use std::time::Duration; + static CLIENT: LazyLock = LazyLock::new(|| { + reqwest::Client::builder() + .timeout(Duration::from_secs(10)) + .build() + .expect("HTTP client build must succeed") + }); + &CLIENT +} + #[cfg(feature = "reqwest")] async fn send_message_impl(channel_id: &str, text: &str) -> Result { - use reqwest::Client; - use std::time::Duration; - let base_url = std::env::var("SPROUT_RELAY_BASE_URL") .unwrap_or_else(|_| "http://localhost:3000".to_owned()); let url = format!("{base_url}/api/channels/{channel_id}/messages"); - let client = Client::builder() - .timeout(Duration::from_secs(10)) - .build() - .map_err(|e| WorkflowError::WebhookError(format!("failed to build HTTP client: {e}")))?; + let client = shared_http_client(); let mut req = client .post(&url) @@ -927,18 +937,12 @@ async fn send_message_impl(channel_id: &str, text: &str) -> Result Result { - use reqwest::Client; - use std::time::Duration; - let base_url = std::env::var("SPROUT_RELAY_BASE_URL") .unwrap_or_else(|_| "http://localhost:3000".to_owned()); let url = format!("{base_url}/api/messages/{message_id}/reactions"); - let client = Client::builder() - .timeout(Duration::from_secs(10)) - .build() - .map_err(|e| WorkflowError::WebhookError(format!("failed to build HTTP client: {e}")))?; + let client = shared_http_client(); let mut req = client .post(&url) diff --git a/desktop/src/features/channels/ui/ChannelManagementSheet.tsx b/desktop/src/features/channels/ui/ChannelManagementSheet.tsx index 52ac636cf..ce2402236 100644 --- a/desktop/src/features/channels/ui/ChannelManagementSheet.tsx +++ b/desktop/src/features/channels/ui/ChannelManagementSheet.tsx @@ -201,16 +201,30 @@ export function ChannelManagementSheet({ const [inviteRole, setInviteRole] = React.useState>("member"); + // Sync drafts from server only when the sheet opens or the channel changes — + // not on every background refetch, which would clobber in-flight edits. + const syncedForRef = React.useRef(null); React.useEffect(() => { + if (!open) { + // Reset on close so the next open re-syncs from server. + syncedForRef.current = null; + return; + } if (!detail) { return; } + const key = detail.id; + if (syncedForRef.current === key) { + return; + } + syncedForRef.current = key; + setNameDraft(detail.name); setDescriptionDraft(detail.description); setTopicDraft(detail.topic ?? ""); setPurposeDraft(detail.purpose ?? ""); - }, [detail]); + }, [detail, open]); if (!channel) { return null; diff --git a/desktop/src/shared/api/relayClient.ts b/desktop/src/shared/api/relayClient.ts index c0f06645c..c42ae324e 100644 --- a/desktop/src/shared/api/relayClient.ts +++ b/desktop/src/shared/api/relayClient.ts @@ -6,7 +6,7 @@ import { signRelayEvent, } from "@/shared/api/tauri"; import type { PresenceStatus, RelayEvent } from "@/shared/api/types"; -import { STREAM_MESSAGE_KINDS } from "@/shared/constants/kinds"; +import { CHANNEL_EVENT_KINDS } from "@/shared/constants/kinds"; import { getTextPayload, sortEvents, @@ -175,7 +175,7 @@ class RelayClient { limit: number, ): RelaySubscriptionFilter { return { - kinds: [...STREAM_MESSAGE_KINDS], + kinds: [...CHANNEL_EVENT_KINDS], "#h": [channelId], limit, }; @@ -183,7 +183,7 @@ class RelayClient { private buildGlobalStreamFilter(limit: number): RelaySubscriptionFilter { return { - kinds: [...STREAM_MESSAGE_KINDS], + kinds: [...CHANNEL_EVENT_KINDS], limit, }; } diff --git a/desktop/src/shared/constants/kinds.ts b/desktop/src/shared/constants/kinds.ts index 8cc354e96..012cf299b 100644 --- a/desktop/src/shared/constants/kinds.ts +++ b/desktop/src/shared/constants/kinds.ts @@ -5,9 +5,9 @@ export const KIND_STREAM_MESSAGE_V2 = 40002; export const KIND_STREAM_MESSAGE_EDIT = 40003; export const KIND_STREAM_MESSAGE_DIFF = 40008; -export const STREAM_MESSAGE_KINDS = [ - KIND_DELETION, // 5 — tombstones for messages/reactions +export const CHANNEL_EVENT_KINDS = [ + KIND_DELETION, // 5 — NIP-09 event deletions KIND_REACTION, // 7 — NIP-25 reactions - KIND_STREAM_MESSAGE, // 40001 — was in original filter - KIND_STREAM_MESSAGE_DIFF, // 40008 — new in this PR + KIND_STREAM_MESSAGE, // 40001 — channel messages + KIND_STREAM_MESSAGE_DIFF, // 40008 — message diffs ] as const; diff --git a/migrations/20260315000002_reaction_event_id_unique.sql b/migrations/20260315000002_reaction_event_id_unique.sql new file mode 100644 index 000000000..5d179305b --- /dev/null +++ b/migrations/20260315000002_reaction_event_id_unique.sql @@ -0,0 +1,4 @@ +-- Upgrade the reaction_event_id index from non-unique to unique. +-- Prevents duplicate kind:7 events from creating orphaned reaction rows. +DROP INDEX idx_reactions_source_event ON reactions; +CREATE UNIQUE INDEX idx_reactions_source_event ON reactions (reaction_event_id);