fix: harden emoji reactions + feat: configurable ACP event subscriptions (#59)

This commit is contained in:
tlongwell-block
2026-03-13 21:25:50 -04:00
committed by GitHub
parent f4b9aadade
commit c85a2226a5
20 changed files with 2467 additions and 410 deletions

127
Cargo.lock generated
View File

@@ -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",

View File

@@ -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"] }

View File

@@ -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<u64>,
}
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<StopReason, AcpError> {
// 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(())
}

View File

@@ -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<String>,
/// 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<String>,
#[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<String>,
// --- 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<String>,
#[arg(
long,
env = "SPROUT_ACP_SYSTEM_PROMPT_FILE",
conflicts_with = "system_prompt"
)]
pub system_prompt_file: Option<PathBuf>,
#[arg(long, env = "SPROUT_ACP_INITIAL_MESSAGE")]
pub initial_message: Option<String>,
#[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<Vec<u32>>,
#[arg(long, env = "SPROUT_ACP_CHANNELS", value_delimiter = ',')]
pub channels: Option<Vec<String>>,
#[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<Keys, ConfigError> {
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<Vec<u32>>,
/// 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<String>,
pub relay_url: String,
pub agent_command: String,
pub agent_args: Vec<String>,
pub mcp_command: String,
pub turn_timeout_secs: u64,
pub system_prompt: Option<String>,
pub initial_message: Option<String>,
pub subscribe_mode: SubscribeMode,
pub dedup_mode: DedupMode,
pub ignore_self: bool,
pub kinds_override: Option<Vec<u32>>,
pub channels_override: Option<Vec<String>>,
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<Self, ConfigError> {
let keys = parse_keys_var("SPROUT_PRIVATE_KEY")
.or_else(|_| parse_keys_var("SPROUT_ACP_PRIVATE_KEY"))?;
pub fn from_cli() -> Result<Self, ConfigError> {
// 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::<u64>().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<SubscriptionRule>,
}
pub fn load_rules(path: &std::path::Path) -> Result<Vec<SubscriptionRule>, 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<Uuid, ChannelFilter> {
use sprout_core::kind::{
KIND_STREAM_MESSAGE, KIND_STREAM_REMINDER, KIND_WORKFLOW_APPROVAL_REQUESTED,
};
let target_channels: Vec<Uuid> = if let Some(ref overrides) = config.channels_override {
overrides
.iter()
.filter_map(|s| s.parse::<Uuid>().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<Vec<u32>> = 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::<Uuid>().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<u32>,
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();
}
}

View File

@@ -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<String>),
}
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<u32>,
/// 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<String>,
/// Tag passed to the prompt template. Falls back to `name` if absent.
#[serde(default)]
pub prompt_tag: Option<String>,
}
// ── 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<bool, FilterError> {
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<evalexpr::HashMapContext, String> {
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<MatchedRule> {
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");
}
}

View File

@@ -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<SubscriptionRule> = 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<Uuid, String> = 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<AcpClient> {
// ── 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<Uuid, String>,
channel_id: Uuid,
acp: &mut AcpClient,
mcp_servers: &[McpServer],
config: &Config,
) -> Result<String, AcpError> {
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<McpServer> {
},
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 {

View File

@@ -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<Event>,
pub events: Vec<BatchEvent>,
}
// ── EventQueue ────────────────────────────────────────────────────────────────
/// Per-channel event queue with global one-in-flight enforcement.
///
/// # State Machine
///
/// ```text
/// State:
/// queues: Map<channel_id, VecDeque<QueuedEvent>>
/// prompt_in_flight: bool
/// queues: Map<channel_id, VecDeque<QueuedEvent>>
/// in_flight_channel: Option<Uuid>
/// 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<FlushBatch>:
/// 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<Uuid, VecDeque<QueuedEvent>>,
prompt_in_flight: bool,
in_flight_channel: Option<Uuid>,
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<FlushBatch> {
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<Event> = queue.into_iter().map(|qe| qe.event).collect();
let events: Vec<BatchEvent> = 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: <prompt_tag>]
/// Channel: ...
/// Kind: ...
/// From: ...
/// Time: ...
/// Content: ...
/// ```
///
/// Batch format (N > 1):
/// ```text
/// [Sprout events — N events]
///
/// --- Event 1 (<prompt_tag>) ---
/// 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);
}
}

View File

@@ -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-<uuid>`.
pub async fn subscribe_channel(&mut self, channel_id: Uuid) -> Result<(), RelayError> {
/// NIP-01 `REQ` built from `filter`. Subscription ID is `ch-<uuid>`.
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<Uuid, u64>,
/// Set of event IDs seen, for deduplication.
seen_ids: HashSet<String>,
/// Per-channel filter used on subscribe (for resubscribe after reconnect).
active_filters: HashMap<Uuid, ChannelFilter>,
}
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<RelayCommand>,
@@ -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<u64>,
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)"
}
);
}

View File

@@ -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<Utc>,
pubkey: &[u8],
emoji: &str,
reaction_event_id: &[u8],
) -> Result<bool> {
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,

View File

@@ -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<Utc>,
pubkey: &[u8],
emoji: &str,
reaction_event_id: &[u8],
) -> Result<bool> {
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.

View File

@@ -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 }

View File

@@ -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) ────

View File

@@ -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 {

View File

@@ -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<String, String>,
external_to_internal_event_ids: DashMap<String, String>,
internal_event_channels: DashMap<String, Uuid>,
/// 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<String, String>,
external_to_internal_event_ids: Cache<String, String>,
internal_event_channels: Cache<String, Uuid>,
}
impl Translator {
@@ -55,17 +59,32 @@ impl Translator {
channel_map: Arc<ChannelMap>,
api_base: impl Into<String>,
api_token: impl Into<String>,
relay_pubkey: impl Into<String>,
) -> 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<String> {
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<String> {
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<Uuid> {
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<Option<Event>, ProxyError> {
self.translate_outbound_tag_targeted(event, allowed_channels, Kind::Reaction)
.await
}
async fn translate_outbound_deletion(
&self,
event: &Event,
allowed_channels: &[Uuid],
) -> Result<Option<Event>, 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<Option<Event>, 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<Tag> =
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<Event, ProxyError> {
// Collect all hex #e tag values.
let e_tag_values: Vec<String> = 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<Uuid> = 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<Tag> =
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.

View File

@@ -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,
})))
}

View File

@@ -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<reqwest::Client> = 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<JsonValue, WorkflowError> {
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<JsonValue, Wo
/// POST `{"emoji": emoji}` to `POST /api/messages/{message_id}/reactions`.
#[cfg(feature = "reqwest")]
async fn add_reaction_impl(message_id: &str, emoji: &str) -> Result<JsonValue, WorkflowError> {
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)

View File

@@ -201,16 +201,30 @@ export function ChannelManagementSheet({
const [inviteRole, setInviteRole] =
React.useState<Exclude<ChannelMember["role"], "owner">>("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<string | null>(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;

View File

@@ -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,
};
}

View File

@@ -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;

View File

@@ -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);