From 091ef0c60fc29311e21ddc5927dba603d8d11990 Mon Sep 17 00:00:00 2001 From: Gforce10_desighn Date: Sun, 20 Sep 2026 11:21:19 +0900 Subject: [PATCH] feat(acp): accept signed events through a local forward socket Signed-off-by: Gforce10_desighn --- Cargo.lock | 1 + crates/buzz-acp/Cargo.toml | 2 +- crates/buzz-acp/FORWARDING.md | 41 + crates/buzz-acp/src/config.rs | 314 +++++++ crates/buzz-acp/src/forward.rs | 1330 +++++++++++++++++++++++++++++ crates/buzz-acp/src/intake.rs | 22 + crates/buzz-acp/src/lib.rs | 705 ++++++++++----- crates/buzz-acp/src/relay.rs | 97 +++ crates/buzz-acp/src/setup_mode.rs | 19 + 9 files changed, 2303 insertions(+), 228 deletions(-) create mode 100644 crates/buzz-acp/FORWARDING.md create mode 100644 crates/buzz-acp/src/forward.rs create mode 100644 crates/buzz-acp/src/intake.rs diff --git a/Cargo.lock b/Cargo.lock index 09d8faa9887..237f0e9e904 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5910,6 +5910,7 @@ dependencies = [ "cfg-if 1.0.4", "cfg_aliases", "libc", + "memoffset", ] [[package]] diff --git a/crates/buzz-acp/Cargo.toml b/crates/buzz-acp/Cargo.toml index d047849806f..4102b93c143 100644 --- a/crates/buzz-acp/Cargo.toml +++ b/crates/buzz-acp/Cargo.toml @@ -74,7 +74,7 @@ evalexpr = { workspace = true } # Process-group kill (safe wrapper around killpg) — Unix-only; kill_process_group # has a #[cfg(not(unix))] fallback in acp.rs. [target.'cfg(unix)'.dependencies] -nix = { version = "0.31", default-features = false, features = ["signal"] } +nix = { version = "0.31", default-features = false, features = ["signal", "socket", "user"] } [dev-dependencies] tokio = { workspace = true, features = ["test-util"] } diff --git a/crates/buzz-acp/FORWARDING.md b/crates/buzz-acp/FORWARDING.md new file mode 100644 index 00000000000..1f40e5a0488 --- /dev/null +++ b/crates/buzz-acp/FORWARDING.md @@ -0,0 +1,41 @@ +# Local forward input + +On Unix, `--forward-socket PATH --forward-peer-uid UID --forward-state PATH` +enables a length-prefixed JSON input compatible with the Hermes Buzz gateway. +The state path is required explicitly; use a private persistent directory. +`--relay-input false` disables channel and observer input subscriptions while +retaining membership updates and outbound publication. It requires a forward +socket. Setup mode retains its existing relay input. + +The socket checks native peer UID, original Nostr event signature, channel tag, +canonical channel UUID, membership and the SHA-256 dedupe key. Membership and +observer-control events are not admitted over the socket. Normal events use +the same author gate, mention/subscription rules, self filter, thread scope, +queue and steering path as relay events. Forward events refresh relay signing +identity before delegated workflow attribution because they have no WebSocket +connection generation. Windows can continue using relay input; forward socket +configuration is rejected there. + +One connection sends a four-byte big-endian length, then a JSON object with +`event`, `direction: "inbound"`, `chat_type`, `channel_id`, and `dedupe_key`. +The key is SHA-256 over event ID hex + `inbound` + canonical channel UUID. +Frames are limited to 8 MiB, with read deadlines and 32 concurrent connections. +ACKs use the same length prefix and contain `ack` (the event ID), `status`, +and `reason` (a string or null). The socket uses mode 0660: the client must +also have filesystem access through ownership/group permissions; passing the +peer UID check alone does not grant that access. + +`accepted` means admission or intentional policy filtering plus a persisted +occupancy key. `duplicate` also covers the configured drop-while-busy policy. +`rejected` is terminal for the current Hermes producer, including membership +mismatch. Keep both sides' channel sets synchronized. Infrastructure failure +has no ACK and invites retry. + +The queue is in memory: ACK is **not** a completed turn or durable work queue. +A crash after admission/persist can lose queued work; a crash before persist +can permit replay. The append-only occupancy file reloads its most recent +50,000 lines; older IDs may replay after restart. The append file and runtime +reservation set grow during a process lifetime; budget storage and memory. A concurrent duplicate can +persist while the original admission is still pending. These bounds make this +an opt-in local adapter, not exactly-once delivery. Use one consumer process +per socket/state file and protect its parent directory from untrusted writers. diff --git a/crates/buzz-acp/src/config.rs b/crates/buzz-acp/src/config.rs index b4d27903c62..a11d3476e52 100644 --- a/crates/buzz-acp/src/config.rs +++ b/crates/buzz-acp/src/config.rs @@ -45,6 +45,12 @@ pub enum ConfigError { #[error("config file error: {0}")] ConfigFile(String), + + /// `--relay-input false` with no forward socket is a silent seat (§4). + #[error( + "RELAY_INPUT_OFF_WITHOUT_FORWARD_SOCKET: --relay-input false requires BUZZ_ACP_FORWARD_SOCKET" + )] + RelayInputOffWithoutForwardSocket, } #[derive(Debug, Clone, PartialEq, clap::ValueEnum)] @@ -525,6 +531,35 @@ pub struct CliArgs { /// ignored (the watermark stays at startup time). #[arg(long, env = "BUZZ_ACP_REPLAY_FLOOR")] pub replay_floor: Option, + /// Unix socket path for Hermes Buzz gateway forward frames. Unset = no listener. + #[arg(long, env = "BUZZ_ACP_FORWARD_SOCKET")] + pub forward_socket: Option, + + /// UID that may connect to the forward socket. Required when the socket is set. + #[arg(long, env = "BUZZ_ACP_FORWARD_PEER_UID")] + pub forward_peer_uid: Option, + + /// Append-only jsonl of occupied forward dedupe keys. + #[arg(long, env = "BUZZ_ACP_FORWARD_STATE", requires = "forward_socket")] + pub forward_state: Option, + + /// Inbound relay subscription switch. Takes an explicit value + /// (`--relay-input false`) instead of being a bare flag: this pin is set + /// from a container env var, and a flag-style bool reads + /// `BUZZ_ACP_RELAY_INPUT=false` as "present, therefore true". + /// + /// Conventions §4 (single input path): when a forward sidecar is present + /// the seat listens on that path only. Default `true` keeps the existing + /// dual-path behaviour. `false` without `BUZZ_ACP_FORWARD_SOCKET` is a + /// silent seat and is rejected at startup. + #[arg( + long, + env = "BUZZ_ACP_RELAY_INPUT", + action = clap::ArgAction::Set, + num_args = 1, + default_value_t = true + )] + pub relay_input: bool, } /// Merged NIP-01 subscription filter for a single channel. @@ -629,6 +664,16 @@ pub struct Config { /// `from_cli()`. `None` when using the compiled-in default or when /// `--no-base-prompt` is set. pub base_prompt_content: Option, + /// Forward-input unix socket. `None` keeps the harness on relay-only intake. + pub forward_socket: Option, + /// Peer UID allowlisted to write forward frames. Required with `forward_socket`. + pub forward_peer_uid: Option, + /// Durable occupancy file for forward `dedupe_key`s. + pub forward_state: Option, + /// When false, the harness does not open inbound relay subscriptions. + /// Outbound publish (presence, occupancy, signed events) stays on. + /// Conventions §4: forward sidecar is then the only input path. + pub relay_input: bool, } /// Maximum length, in characters, of a session title sent to the adapter. @@ -1150,6 +1195,27 @@ impl Config { validate_multiple_event_handling(args.multiple_event_handling, args.dedup)?; + if !args.relay_input && args.forward_socket.is_none() { + return Err(ConfigError::RelayInputOffWithoutForwardSocket); + } + if args.forward_socket.is_some() { + if !cfg!(unix) { + return Err(ConfigError::ConfigFile( + "forward input requires Unix".into(), + )); + } + if args.forward_peer_uid.is_none() { + return Err(ConfigError::ConfigFile( + "BUZZ_ACP_FORWARD_SOCKET requires BUZZ_ACP_FORWARD_PEER_UID".into(), + )); + } + if args.forward_state.is_none() { + return Err(ConfigError::ConfigFile( + "BUZZ_ACP_FORWARD_SOCKET requires BUZZ_ACP_FORWARD_STATE".into(), + )); + } + } + let config = Config { keys, relay_url: args.relay_url, @@ -1204,6 +1270,10 @@ impl Config { agent_owner: args.agent_owner.map(|s| s.trim().to_ascii_lowercase()), no_base_prompt: args.no_base_prompt, base_prompt_content, + forward_socket: args.forward_socket, + forward_peer_uid: args.forward_peer_uid, + forward_state: args.forward_state, + relay_input: args.relay_input, }; Ok(config) @@ -1253,6 +1323,20 @@ impl Config { } } +/// Startup log line for the two intake paths. Relay off still holds a +/// membership REQ (`off(membership-only)`) so new rooms can be learned. +pub(crate) fn input_paths_log_line(relay_input: bool, forward: bool) -> String { + format!( + "input_paths relay={} forward={}", + if relay_input { + "on" + } else { + "off(membership-only)" + }, + if forward { "on" } else { "off" }, + ) +} + #[derive(Debug, serde::Deserialize)] struct TomlConfig { #[serde(default)] @@ -1580,6 +1664,10 @@ mod tests { agent_owner: None, no_base_prompt: false, base_prompt_content: None, + forward_socket: None, + forward_peer_uid: None, + forward_state: None, + relay_input: true, } } @@ -2453,6 +2541,28 @@ channels = "ALL" assert_eq!(config.permission_mode, PermissionMode::BypassPermissions); } + #[cfg(unix)] + #[test] + fn test_forward_socket_without_peer_uid_is_config_error() { + let key = "0000000000000000000000000000000000000000000000000000000000000001"; + let args = CliArgs::try_parse_from([ + "buzz-acp", + "--private-key", + key, + "--forward-socket", + "/tmp/bz-test.sock", + ]) + .expect("cli parses"); + assert!(args.forward_socket.is_some()); + assert!(args.forward_peer_uid.is_none()); + let err = Config::from_args(args).expect_err("socket without peer uid must fail"); + let msg = err.to_string(); + assert!( + msg.contains("FORWARD_PEER_UID"), + "error should name the missing peer uid: {msg}" + ); + } + #[test] fn test_permission_mode_value_enum_kebab_case() { // clap::ValueEnum generates kebab-case by default from PascalCase variants. @@ -3189,4 +3299,208 @@ channels = "ALL" Add `hide_env_values = true` to each: {violations:?}" ); } + // --- Relay inbound switch (§4 single input path) ---------------------------- + + #[test] + fn relay_input_takes_an_explicit_value_not_bare_presence() { + let bare = CliArgs::try_parse_from([ + "buzz-acp", + "--private-key", + TEST_PRIVATE_KEY, + "--relay-input", + ]); + assert!(bare.is_err(), "a bare flag must not be accepted"); + + let off = CliArgs::try_parse_from([ + "buzz-acp", + "--private-key", + TEST_PRIVATE_KEY, + "--relay-input", + "false", + ]) + .expect("explicit false parses"); + assert!(!off.relay_input); + + let on = CliArgs::try_parse_from([ + "buzz-acp", + "--private-key", + TEST_PRIVATE_KEY, + "--relay-input", + "true", + ]) + .expect("explicit true parses"); + assert!(on.relay_input); + } + + #[test] + fn relay_input_defaults_true() { + let args = CliArgs::try_parse_from(["buzz-acp", "--private-key", TEST_PRIVATE_KEY]) + .expect("clap should parse args"); + assert!(args.relay_input); + let config = Config::from_args(args).expect("default config is valid"); + assert!(config.relay_input); + } + + #[test] + fn relay_input_env_false_is_actually_false() { + use clap::{CommandFactory, FromArgMatches}; + let cmd = CliArgs::command(); + let arg = cmd + .get_arguments() + .find(|a| a.get_id() == "relay_input") + .expect("relay_input arg exists"); + assert_eq!( + arg.get_env() + .map(|v| v.to_string_lossy().into_owned()) + .as_deref(), + Some("BUZZ_ACP_RELAY_INPUT") + ); + assert!( + matches!(arg.get_action(), clap::ArgAction::Set), + "must take a value so env false is not 'present therefore true'" + ); + + // clap's bool value parser is what the env fallback uses. `false` must + // stay false — this is the Set vs SetTrue distinction the launcher pin + // depends on. + let matches = CliArgs::command() + .try_get_matches_from([ + "buzz-acp", + "--private-key", + TEST_PRIVATE_KEY, + "--relay-input", + "false", + ]) + .expect("explicit false matches"); + let parsed = CliArgs::from_arg_matches(&matches).expect("from matches"); + assert!(!parsed.relay_input); + } + + #[test] + fn relay_input_off_without_forward_socket_is_config_error() { + let args = CliArgs::try_parse_from([ + "buzz-acp", + "--private-key", + TEST_PRIVATE_KEY, + "--relay-input", + "false", + ]) + .expect("cli parses"); + let err = Config::from_args(args).expect_err("silent seat must fail"); + let msg = err.to_string(); + assert!( + matches!(err, ConfigError::RelayInputOffWithoutForwardSocket), + "expected RelayInputOffWithoutForwardSocket, got {err:?}" + ); + assert!( + msg.contains("RELAY_INPUT_OFF_WITHOUT_FORWARD_SOCKET"), + "error must name the reject code: {msg}" + ); + } + + #[cfg(unix)] + #[test] + fn relay_input_off_with_forward_socket_is_ok() { + let args = CliArgs::try_parse_from([ + "buzz-acp", + "--private-key", + TEST_PRIVATE_KEY, + "--relay-input", + "false", + "--forward-socket", + "/tmp/bz-fwd.sock", + "--forward-peer-uid", + "10001", + "--forward-state", + "/tmp/acp-forward-test.jsonl", + ]) + .expect("cli parses"); + let config = Config::from_args(args).expect("off + forward must start"); + assert!(!config.relay_input); + assert!(config.forward_socket.is_some()); + } + + #[test] + fn relay_input_off_without_socket_beats_missing_peer_uid() { + let args = CliArgs::try_parse_from([ + "buzz-acp", + "--private-key", + TEST_PRIVATE_KEY, + "--relay-input", + "false", + ]) + .expect("cli parses"); + let err = Config::from_args(args).expect_err("silent seat"); + assert!( + matches!(err, ConfigError::RelayInputOffWithoutForwardSocket), + "got {err:?}" + ); + assert!(err + .to_string() + .contains("RELAY_INPUT_OFF_WITHOUT_FORWARD_SOCKET")); + } + + #[cfg(unix)] + #[test] + fn relay_input_off_with_socket_missing_peer_uid_is_not_the_silent_seat_code() { + let args = CliArgs::try_parse_from([ + "buzz-acp", + "--private-key", + TEST_PRIVATE_KEY, + "--relay-input", + "false", + "--forward-socket", + "/tmp/bz-fwd.sock", + ]) + .expect("cli parses"); + let err = Config::from_args(args).expect_err("peer uid required"); + let msg = err.to_string(); + assert!( + !matches!(err, ConfigError::RelayInputOffWithoutForwardSocket), + "socket present is not the silent-seat reject: {err:?}" + ); + assert!( + !msg.contains("RELAY_INPUT_OFF_WITHOUT_FORWARD_SOCKET"), + "stderr identifier must not fire when a socket was given: {msg}" + ); + assert!( + msg.contains("BUZZ_ACP_FORWARD_PEER_UID"), + "peer-uid error: {msg}" + ); + } + + #[cfg(not(unix))] + #[test] + fn forward_socket_is_rejected_on_non_unix() { + let args = CliArgs::try_parse_from([ + "buzz-acp", + "--private-key", + TEST_PRIVATE_KEY, + "--forward-socket", + "forward.sock", + "--forward-peer-uid", + "10001", + "--forward-state", + "forward.jsonl", + ]) + .expect("cli parses"); + let err = Config::from_args(args).expect_err("forward input is Unix-only"); + assert!(err.to_string().contains("forward input requires Unix")); + } + + #[test] + fn input_paths_log_line_names_both_switches() { + assert_eq!( + input_paths_log_line(true, false), + "input_paths relay=on forward=off" + ); + assert_eq!( + input_paths_log_line(false, true), + "input_paths relay=off(membership-only) forward=on" + ); + assert_eq!( + input_paths_log_line(false, false), + "input_paths relay=off(membership-only) forward=off" + ); + } } diff --git a/crates/buzz-acp/src/forward.rs b/crates/buzz-acp/src/forward.rs new file mode 100644 index 00000000000..2ebf5091cbe --- /dev/null +++ b/crates/buzz-acp/src/forward.rs @@ -0,0 +1,1330 @@ +//! Unix-socket consumer for Hermes Buzz gateway forward frames. +//! +//! One connection carries one length-prefixed JSON frame. Durable occupancy of +//! `dedupe_key` plus admission is the ack criterion (`accepted`), not agent +//! turn completion. The in-memory queue is not durable across a crash. + +use std::collections::{HashSet, VecDeque}; +use std::fs::{File, OpenOptions}; +use std::future::Future; +use std::io::{BufRead, BufReader, Write}; +use std::os::unix::fs::{FileTypeExt, PermissionsExt}; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use anyhow::{Context, Result}; +use buzz_core::kind::{ + KIND_AGENT_OBSERVER_FRAME, KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION, +}; +use nostr::Event; +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; +use sha2::{Digest, Sha256}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{UnixListener, UnixStream}; +use tokio::sync::{mpsc, oneshot}; +use uuid::Uuid; + +use crate::intake::{EnqueueResult, ForwardWork}; +use crate::relay::BuzzEvent; + +pub const FORWARD_FRAME_MAX: usize = 8 * 1024 * 1024; +/// Append-only file is unbounded; `load` keeps only the most recent N keys in +/// reload HashSet with bounded parsing memory. Runtime reservations can grow. +/// Keys that remain on +/// disk but fall outside this window are treated as new after restart and may +/// be processed again. +const DEDUPE_MEMORY_CAP: usize = 50_000; +const ACK_TIMEOUT: Duration = Duration::from_secs(10); + +pub fn compute_dedupe_key(event_id: &str, direction: &str, channel_id: &str) -> String { + let mut hasher = Sha256::new(); + hasher.update(event_id.as_bytes()); + hasher.update(direction.as_bytes()); + hasher.update(channel_id.as_bytes()); + hex::encode(hasher.finalize()) +} + +pub fn relay_dedupe_key(event: &Event, channel_id: Uuid) -> String { + compute_dedupe_key(&event.id.to_hex(), "inbound", &channel_id.to_string()) +} + +#[derive(Debug, Serialize, Deserialize)] +struct ForwardFrame { + event: Value, + direction: String, + /// 예약: wire-contract field. Not used for authorization. + #[allow(dead_code)] + chat_type: String, + channel_id: String, + dedupe_key: String, +} + +#[derive(Clone)] +pub struct DedupeStore { + path: PathBuf, + keys: HashSet, +} + +fn is_complete_dedupe_line(line: &str) -> bool { + line.len() == 64 && line.bytes().all(|b| matches!(b, b'0'..=b'9' | b'a'..=b'f')) +} + +impl DedupeStore { + pub fn load(path: impl AsRef) -> std::io::Result { + Self::load_with_cap(path, DEDUPE_MEMORY_CAP) + } + + fn load_with_cap(path: impl AsRef, cap: usize) -> std::io::Result { + let path = path.as_ref().to_path_buf(); + if let Some(parent) = path.parent() { + std::fs::create_dir_all(parent)?; + } + let mut ordered = VecDeque::new(); + if path.is_file() { + let file = File::open(&path)?; + for line in BufReader::new(file).lines() { + let line = line?; + let trimmed = line.trim(); + // Skip empty and truncated/partial last lines from a crash mid-write. + if is_complete_dedupe_line(trimmed) { + ordered.push_back(trimmed.to_string()); + if ordered.len() > cap { + ordered.pop_front(); + } + } + } + } + // Cap is the most recent N *raw complete lines*, not unique keys. + // Repeated duplicate appends can fill the window and drop older + // distinct keys on reload. + let keys = ordered.into_iter().collect(); + Ok(Self { path, keys }) + } + + #[cfg(test)] + pub fn contains(&self, key: &str) -> bool { + self.keys.contains(key) + } + + pub fn path(&self) -> &Path { + &self.path + } + + #[cfg(test)] + pub fn len(&self) -> usize { + self.keys.len() + } + + /// Memory-only check-and-insert. Critical section must not fsync. + /// Returns `true` when this call reserved the key. + pub fn reserve(&mut self, key: &str) -> bool { + self.keys.insert(key.to_string()) + } + + pub fn unreserve(&mut self, key: &str) { + self.keys.remove(key); + } + + fn append_key(path: &Path, key: &str) -> std::io::Result<()> { + // Serialize the full record even if write_all needs more than one write. + // This lock is only held by blocking threads, never the intake loop. + static APPEND_LOCK: Mutex<()> = Mutex::new(()); + let _append = APPEND_LOCK.lock().unwrap_or_else(|e| e.into_inner()); + let mut file = OpenOptions::new().create(true).append(true).open(path)?; + file.write_all(format!("{key}\n").as_bytes())?; + file.sync_all()?; + Ok(()) + } +} + +/// Append+fsync on a blocking thread. Caller must not hold the store mutex. +pub async fn persist_dedupe_key(path: PathBuf, key: String) -> std::io::Result<()> { + tokio::task::spawn_blocking(move || DedupeStore::append_key(&path, &key)) + .await + .unwrap_or_else(|e| Err(std::io::Error::other(e.to_string()))) +} + +/// Persist occupancy after the event was already handled. On I/O failure the +/// in-memory reservation stays — rollback is only valid for `InfraFailure`. +/// A crash between queue and persist is an accepted one-event window: restart +/// loses the HashSet, so a retry may enqueue again. +/// Sustained I/O failure opens a multi-event window: a `duplicate` ack from +/// memory alone lets the gateway advance while disk never recorded the keys, +/// and relay redelivery after restart would reprocess them. The duplicate +/// path therefore retries persist once before acking. +/// Residual shutdown window only: a duplicate-path persist retry can land on +/// disk while another connection's enqueue is still in flight; if that +/// enqueue then returns `InfraFailure` and unreserves, disk has the key and +/// memory does not. Reload treats the on-disk key as already processed. +/// `InfraFailure` fires only when the main loop is already gone. Not closed +/// in code (an in-flight marker would skip the duplicate retry). +pub async fn persist_occupancy( + store: &Arc>, + key: String, +) -> std::io::Result<()> { + let path = { + let guard = store.lock().unwrap_or_else(|e| e.into_inner()); + guard.path().to_path_buf() + }; + match persist_dedupe_key(path, key.clone()).await { + Ok(()) => Ok(()), + Err(e) => { + tracing::warn!(key = %key, error = %e, "forward dedupe persist failed"); + Err(e) + } + } +} + +#[derive(Clone)] +pub struct ForwardPolicy { + pub peer_uid: u32, + pub subscribed: Arc>>, + pub store: Arc>, +} + +fn peer_uid(stream: &UnixStream) -> Result { + #[cfg(target_os = "linux")] + { + let cred = nix::sys::socket::getsockopt(stream, nix::sys::socket::sockopt::PeerCredentials) + .context("SO_PEERCRED")?; + Ok(cred.uid()) + } + #[cfg(target_os = "macos")] + { + let cred = nix::sys::socket::getsockopt(stream, nix::sys::socket::sockopt::LocalPeerCred) + .context("LOCAL_PEERCRED")?; + Ok(cred.uid()) + } + #[cfg(not(any(target_os = "linux", target_os = "macos")))] + { + let _ = stream; + anyhow::bail!("peer credentials unsupported on this OS") + } +} + +#[cfg(test)] +fn current_uid() -> u32 { + nix::unistd::Uid::current().as_raw() +} + +fn log_status(event_id: &str, channel_id: &str, status: &str, reason: Option<&str>) { + let ev8: String = event_id.chars().take(8).collect(); + let ch8: String = channel_id.chars().take(8).collect(); + match reason { + Some(r) => { + tracing::info!(event_id = %ev8, channel_id = %ch8, status, reason = r, "forward") + } + None => tracing::info!(event_id = %ev8, channel_id = %ch8, status, "forward"), + } +} + +fn encode_ack(event_id: &str, status: &str, reason: Option<&str>) -> Vec { + let body = json!({ + "ack": event_id, + "status": status, + "reason": reason, + }); + let blob = serde_json::to_vec(&body).unwrap_or_else(|_| b"{}".to_vec()); + let mut out = Vec::with_capacity(4 + blob.len()); + out.extend_from_slice(&(blob.len() as u32).to_be_bytes()); + out.extend_from_slice(&blob); + out +} + +async fn write_ack(stream: &mut UnixStream, event_id: &str, status: &str, reason: Option<&str>) { + let bytes = encode_ack(event_id, status, reason); + if let Err(e) = stream.write_all(&bytes).await { + tracing::debug!(error = %e, "forward ack write failed (already accepted)"); + } +} + +fn event_has_h_tag(event: &Event, channel_id: Uuid) -> bool { + let expected = channel_id.to_string(); + event.tags.iter().any(|tag| { + let v = tag.as_slice(); + v.len() >= 2 && v[0] == "h" && v[1] == expected + }) +} + +enum ReadFrameError { + Timeout, + TooLarge, + Bad, +} + +async fn read_frame(stream: &mut UnixStream) -> std::result::Result { + let mut header = [0u8; 4]; + match tokio::time::timeout(ACK_TIMEOUT, stream.read_exact(&mut header)).await { + Err(_) => return Err(ReadFrameError::Timeout), + Ok(Err(_)) => return Err(ReadFrameError::Bad), + Ok(Ok(_)) => {} + } + let size = u32::from_be_bytes(header) as usize; + if size == 0 { + return Err(ReadFrameError::Bad); + } + if size > FORWARD_FRAME_MAX { + return Err(ReadFrameError::TooLarge); + } + let mut payload = vec![0u8; size]; + match tokio::time::timeout(ACK_TIMEOUT, stream.read_exact(&mut payload)).await { + Err(_) => return Err(ReadFrameError::Timeout), + Ok(Err(_)) => return Err(ReadFrameError::Bad), + Ok(Ok(_)) => {} + } + serde_json::from_slice(&payload).map_err(|_| ReadFrameError::Bad) +} + +pub async fn serve_connection(mut stream: UnixStream, policy: &ForwardPolicy, enqueue: F) +where + F: FnOnce(BuzzEvent) -> Fut, + Fut: Future, +{ + let uid = match peer_uid(&stream) { + Ok(uid) => uid, + Err(_) => { + log_status("unknown", "unknown", "rejected", Some("peer_uid")); + write_ack(&mut stream, "unknown", "rejected", Some("peer_uid")).await; + return; + } + }; + if uid != policy.peer_uid { + log_status("unknown", "unknown", "rejected", Some("peer_uid")); + write_ack(&mut stream, "unknown", "rejected", Some("peer_uid")).await; + return; + } + + let frame = match read_frame(&mut stream).await { + Ok(frame) => frame, + Err(ReadFrameError::Timeout) => return, + Err(ReadFrameError::TooLarge) => { + log_status("unknown", "unknown", "rejected", Some("frame_too_large")); + write_ack(&mut stream, "unknown", "rejected", Some("frame_too_large")).await; + return; + } + Err(ReadFrameError::Bad) => { + log_status("unknown", "unknown", "rejected", Some("bad_frame")); + write_ack(&mut stream, "unknown", "rejected", Some("bad_frame")).await; + return; + } + }; + + let event_id_hint = frame + .event + .get("id") + .and_then(|v| v.as_str()) + .unwrap_or("unknown") + .to_string(); + let channel_hint = frame.channel_id.clone(); + + if frame.direction != "inbound" { + log_status(&event_id_hint, &channel_hint, "rejected", Some("bad_frame")); + write_ack(&mut stream, &event_id_hint, "rejected", Some("bad_frame")).await; + return; + } + let channel_id = match Uuid::parse_str(&frame.channel_id) { + Ok(id) => id, + Err(_) => { + log_status(&event_id_hint, &channel_hint, "rejected", Some("bad_frame")); + write_ack(&mut stream, &event_id_hint, "rejected", Some("bad_frame")).await; + return; + } + }; + if frame.channel_id != channel_id.to_string() { + log_status(&event_id_hint, &channel_hint, "rejected", Some("bad_frame")); + write_ack(&mut stream, &event_id_hint, "rejected", Some("bad_frame")).await; + return; + } + let event: Event = match serde_json::from_value(frame.event.clone()) { + Ok(event) => event, + Err(_) => { + log_status(&event_id_hint, &channel_hint, "rejected", Some("bad_frame")); + write_ack(&mut stream, &event_id_hint, "rejected", Some("bad_frame")).await; + return; + } + }; + if event.verify().is_err() { + log_status( + &event_id_hint, + &channel_hint, + "rejected", + Some("bad_signature"), + ); + write_ack( + &mut stream, + &event_id_hint, + "rejected", + Some("bad_signature"), + ) + .await; + return; + } + let event_id = event.id.to_hex(); + if !event_has_h_tag(&event, channel_id) { + log_status( + &event_id, + &frame.channel_id, + "rejected", + Some("channel_mismatch"), + ); + write_ack(&mut stream, &event_id, "rejected", Some("channel_mismatch")).await; + return; + } + let subscribed = policy + .subscribed + .lock() + .unwrap_or_else(|e| e.into_inner()) + .contains(&channel_id); + if !subscribed { + log_status( + &event_id, + &frame.channel_id, + "rejected", + Some("channel_not_subscribed"), + ); + write_ack( + &mut stream, + &event_id, + "rejected", + Some("channel_not_subscribed"), + ) + .await; + return; + } + let kind = event.kind.as_u16() as u32; + if matches!( + kind, + KIND_MEMBER_ADDED_NOTIFICATION + | KIND_MEMBER_REMOVED_NOTIFICATION + | KIND_AGENT_OBSERVER_FRAME + ) { + log_status( + &event_id, + &frame.channel_id, + "rejected", + Some("wake_denied"), + ); + write_ack(&mut stream, &event_id, "rejected", Some("wake_denied")).await; + return; + } + let channel_s = channel_id.to_string(); + let expected = compute_dedupe_key(&event_id, "inbound", &channel_s); + if expected != frame.dedupe_key { + log_status( + &event_id, + &frame.channel_id, + "rejected", + Some("dedupe_key_mismatch"), + ); + write_ack( + &mut stream, + &event_id, + "rejected", + Some("dedupe_key_mismatch"), + ) + .await; + return; + } + + let reserved = { + let mut store = policy.store.lock().unwrap_or_else(|e| e.into_inner()); + store.reserve(&expected) + }; + if !reserved { + // Memory-only occupancy: retry persist once so a prior disk failure + // does not ack duplicate while the key is still missing on disk. + if persist_occupancy(&policy.store, expected.clone()) + .await + .is_err() + { + return; + } + log_status(&event_id, &frame.channel_id, "duplicate", None); + write_ack(&mut stream, &event_id, "duplicate", None).await; + return; + } + + // Forward frames have no relay connection generation. The shared author + // gate refreshes relay identity for this sentinel before attribution. + let buzz_event = BuzzEvent { + connection_generation: u64::MAX, + channel_id, + event, + }; + let outcome = enqueue(buzz_event).await; + if outcome == EnqueueResult::InfraFailure { + { + let mut store = policy.store.lock().unwrap_or_else(|e| e.into_inner()); + store.unreserve(&expected); + } + return; + } + + // Queued/Drop/Ignored already happened. Persist failure must not unreserve: + // a gateway retry then hits the in-memory key and acks duplicate. + if persist_occupancy(&policy.store, expected.clone()) + .await + .is_err() + { + return; + } + + match outcome { + EnqueueResult::Queued | EnqueueResult::Ignored => { + log_status(&event_id, &frame.channel_id, "accepted", None); + write_ack(&mut stream, &event_id, "accepted", None).await; + } + EnqueueResult::Drop => { + log_status( + &event_id, + &frame.channel_id, + "duplicate", + Some("queue_drop"), + ); + write_ack(&mut stream, &event_id, "duplicate", Some("queue_drop")).await; + } + EnqueueResult::InfraFailure => {} + } +} + +pub fn bind_listener(path: &Path) -> Result { + if let Ok(meta) = std::fs::symlink_metadata(path) { + anyhow::ensure!(meta.file_type().is_socket(), "forward path is not a socket"); + anyhow::ensure!( + std::os::unix::net::UnixStream::connect(path).is_err(), + "forward socket is already listening" + ); + std::fs::remove_file(path) + .with_context(|| format!("remove stale socket {}", path.display()))?; + } + if let Some(parent) = path.parent() { + std::fs::create_dir_all(parent)?; + } + let tmp = path.with_extension(format!("tmp-{}", std::process::id())); + // Never remove a pre-existing temporary path: it may be unrelated data. + // bind fails closed on any collision, including a dangling symlink. + let listener = UnixListener::bind(&tmp).with_context(|| format!("bind {}", tmp.display()))?; + if let Err(e) = std::fs::set_permissions(&tmp, std::fs::Permissions::from_mode(0o660)) { + let _ = std::fs::remove_file(&tmp); + return Err(e).with_context(|| format!("chmod {}", tmp.display())); + } + if let Err(e) = std::fs::rename(&tmp, path) { + let _ = std::fs::remove_file(&tmp); + return Err(e).with_context(|| format!("rename {} -> {}", tmp.display(), path.display())); + } + Ok(listener) +} + +pub async fn accept_loop( + listener: UnixListener, + policy: ForwardPolicy, + work_tx: mpsc::Sender, +) { + let slots = Arc::new(tokio::sync::Semaphore::new(32)); + loop { + let Ok(slot) = slots.clone().acquire_owned().await else { + return; + }; + let (stream, _) = match listener.accept().await { + Ok(pair) => pair, + Err(e) => { + tracing::warn!(error = %e, "forward accept failed"); + continue; + } + }; + let policy = policy.clone(); + let work_tx = work_tx.clone(); + tokio::spawn(async move { + let _slot = slot; + serve_connection(stream, &policy, |event| { + let work_tx = work_tx.clone(); + async move { + let (reply, rx) = oneshot::channel(); + if work_tx.send(ForwardWork { event, reply }).await.is_err() { + return EnqueueResult::InfraFailure; + } + rx.await.unwrap_or(EnqueueResult::InfraFailure) + } + }) + .await; + }); + } +} + +#[cfg(test)] +pub fn relay_event_is_duplicate(store: &DedupeStore, event: &Event, channel_id: Uuid) -> bool { + store.contains(&relay_dedupe_key(event, channel_id)) +} + +#[cfg(test)] +mod tests { + use super::*; + use buzz_core::kind::KIND_STREAM_MESSAGE; + use nostr::{EventBuilder, Keys, Kind, Tag}; + use std::sync::atomic::{AtomicU64, Ordering}; + use tokio::io::AsyncWriteExt; + + static SOCK_SEQ: AtomicU64 = AtomicU64::new(1); + + fn sock_path() -> PathBuf { + PathBuf::from(format!( + "/tmp/bz-fwd-{}-{}.sock", + std::process::id(), + SOCK_SEQ.fetch_add(1, Ordering::Relaxed) + )) + } + + fn signed_event(keys: &Keys, kind: u32, content: &str, channel: Uuid) -> Event { + let h = Tag::parse(["h", &channel.to_string()]).expect("h tag"); + EventBuilder::new(Kind::Custom(kind as u16), content) + .tags([h]) + .sign_with_keys(keys) + .expect("sign") + } + + fn frame_for(event: &Event, channel: Uuid, direction: &str) -> Value { + let event_id = event.id.to_hex(); + let channel_s = channel.to_string(); + json!({ + "event": serde_json::to_value(event).unwrap(), + "direction": direction, + "chat_type": "group", + "channel_id": channel_s, + "dedupe_key": compute_dedupe_key(&event_id, direction, &channel_s), + }) + } + + fn encode_frame(value: &Value) -> Vec { + let blob = serde_json::to_vec(value).unwrap(); + let mut out = Vec::with_capacity(4 + blob.len()); + out.extend_from_slice(&(blob.len() as u32).to_be_bytes()); + out.extend_from_slice(&blob); + out + } + + async fn read_ack(stream: &mut UnixStream) -> Value { + let mut header = [0u8; 4]; + stream.read_exact(&mut header).await.expect("ack header"); + let size = u32::from_be_bytes(header) as usize; + let mut payload = vec![0u8; size]; + stream.read_exact(&mut payload).await.expect("ack body"); + serde_json::from_slice(&payload).expect("ack json") + } + + struct Harness { + path: PathBuf, + store: Arc>, + queue: Arc>>, + _listener: UnixListener, + enqueue: EnqueueResult, + policy: ForwardPolicy, + } + + impl Harness { + fn new( + peer_uid: u32, + subscribed: HashSet, + _self_keys: &Keys, + enqueue: EnqueueResult, + ) -> Self { + let path = sock_path(); + let state = path.with_extension("jsonl"); + let store = Arc::new(Mutex::new(DedupeStore::load(&state).unwrap())); + let listener = bind_listener(&path).unwrap(); + let mode = std::fs::metadata(&path).unwrap().permissions().mode() & 0o777; + assert_eq!(mode, 0o660); + let policy = ForwardPolicy { + peer_uid, + subscribed: Arc::new(Mutex::new(subscribed)), + store: store.clone(), + }; + Self { + path, + store, + queue: Arc::new(Mutex::new(Vec::new())), + _listener: listener, + enqueue, + policy, + } + } + + fn block_persist(&self) { + let state = self.store.lock().unwrap().path().to_path_buf(); + let _ = std::fs::remove_file(&state); + std::fs::create_dir(&state).expect("block persist path"); + } + + fn unblock_persist(&self) { + let state = self.store.lock().unwrap().path().to_path_buf(); + let _ = std::fs::remove_dir(&state); + } + + async fn accept_one(&self) { + let (stream, _) = self._listener.accept().await.expect("accept"); + let queue = self.queue.clone(); + let enqueue = self.enqueue; + serve_connection(stream, &self.policy, move |event| { + let queue = queue.clone(); + async move { + if enqueue == EnqueueResult::Queued { + queue.lock().unwrap().push(event); + } + enqueue + } + }) + .await; + } + + async fn exchange(&self, frame: Value) -> Value { + let mut client = UnixStream::connect(&self.path).await.expect("connect"); + let accept = self.accept_one(); + let send = async { + client.write_all(&encode_frame(&frame)).await.unwrap(); + read_ack(&mut client).await + }; + let (_, ack) = tokio::join!(accept, send); + ack + } + } + + impl Drop for Harness { + fn drop(&mut self) { + let _ = std::fs::remove_file(&self.path); + let state = self.path.with_extension("jsonl"); + let _ = std::fs::remove_file(&state); + let _ = std::fs::remove_dir(&state); + } + } + + #[tokio::test] + async fn accepted_frame_queues_and_persists() { + let keys = Keys::generate(); + let seat = Keys::generate(); + let channel = Uuid::new_v4(); + let h = Harness::new( + current_uid(), + HashSet::from([channel]), + &seat, + EnqueueResult::Queued, + ); + let event = signed_event(&keys, KIND_STREAM_MESSAGE, "hello", channel); + let ack = h.exchange(frame_for(&event, channel, "inbound")).await; + assert_eq!(ack["status"], "accepted"); + assert_eq!(ack["ack"], event.id.to_hex()); + assert_eq!(h.queue.lock().unwrap().len(), 1); + assert_eq!(h.store.lock().unwrap().len(), 1); + } + + #[tokio::test] + async fn second_identical_frame_is_duplicate() { + let keys = Keys::generate(); + let seat = Keys::generate(); + let channel = Uuid::new_v4(); + let h = Harness::new( + current_uid(), + HashSet::from([channel]), + &seat, + EnqueueResult::Queued, + ); + let event = signed_event(&keys, KIND_STREAM_MESSAGE, "hello", channel); + let frame = frame_for(&event, channel, "inbound"); + let first = h.exchange(frame.clone()).await; + let second = h.exchange(frame).await; + assert_eq!(first["status"], "accepted"); + assert_eq!(second["status"], "duplicate"); + assert_eq!(h.queue.lock().unwrap().len(), 1); + } + + #[tokio::test] + async fn concurrent_same_key_one_accepted() { + let keys = Keys::generate(); + let seat = Keys::generate(); + let channel = Uuid::new_v4(); + let h = Harness::new( + current_uid(), + HashSet::from([channel]), + &seat, + EnqueueResult::Queued, + ); + let event = signed_event(&keys, KIND_STREAM_MESSAGE, "hello", channel); + let frame = frame_for(&event, channel, "inbound"); + let mut c1 = UnixStream::connect(&h.path).await.unwrap(); + let mut c2 = UnixStream::connect(&h.path).await.unwrap(); + let accept = async { + h.accept_one().await; + h.accept_one().await; + }; + let send = async { + let bytes = encode_frame(&frame); + c1.write_all(&bytes).await.unwrap(); + c2.write_all(&bytes).await.unwrap(); + let a = read_ack(&mut c1).await; + let b = read_ack(&mut c2).await; + (a, b) + }; + let (_, (a, b)) = tokio::join!(accept, send); + let statuses = [a["status"].as_str().unwrap(), b["status"].as_str().unwrap()]; + assert!(statuses.contains(&"accepted")); + assert!(statuses.contains(&"duplicate")); + assert_eq!(h.queue.lock().unwrap().len(), 1); + } + + #[tokio::test] + async fn restart_reloads_store() { + let keys = Keys::generate(); + let seat = Keys::generate(); + let channel = Uuid::new_v4(); + let h = Harness::new( + current_uid(), + HashSet::from([channel]), + &seat, + EnqueueResult::Queued, + ); + let event = signed_event(&keys, KIND_STREAM_MESSAGE, "hello", channel); + let frame = frame_for(&event, channel, "inbound"); + assert_eq!(h.exchange(frame.clone()).await["status"], "accepted"); + let reloaded = DedupeStore::load(h.path.with_extension("jsonl")).unwrap(); + assert!(relay_event_is_duplicate(&reloaded, &event, channel)); + *h.store.lock().unwrap() = reloaded; + assert_eq!(h.exchange(frame).await["status"], "duplicate"); + } + + #[tokio::test] + async fn bad_signature_rejected() { + let keys = Keys::generate(); + let seat = Keys::generate(); + let channel = Uuid::new_v4(); + let h = Harness::new( + current_uid(), + HashSet::from([channel]), + &seat, + EnqueueResult::Queued, + ); + let event = signed_event(&keys, KIND_STREAM_MESSAGE, "hello", channel); + let mut frame = frame_for(&event, channel, "inbound"); + frame["event"]["sig"] = json!("0".repeat(128)); + let ack = h.exchange(frame).await; + assert_eq!(ack["status"], "rejected"); + assert_eq!(ack["reason"], "bad_signature"); + assert!(h.queue.lock().unwrap().is_empty()); + assert_eq!(h.store.lock().unwrap().len(), 0); + } + + #[tokio::test] + async fn membership_set_change_is_visible_to_forward_gate() { + let keys = Keys::generate(); + let seat = Keys::generate(); + let channel = Uuid::new_v4(); + let h = Harness::new(current_uid(), HashSet::new(), &seat, EnqueueResult::Queued); + let first = signed_event(&keys, KIND_STREAM_MESSAGE, "hello", channel); + let ack = h.exchange(frame_for(&first, channel, "inbound")).await; + assert_eq!(ack["reason"], "channel_not_subscribed"); + + h.policy.subscribed.lock().unwrap().insert(channel); + let opened = signed_event(&keys, KIND_STREAM_MESSAGE, "opened", channel); + let ack = h.exchange(frame_for(&opened, channel, "inbound")).await; + assert_eq!(ack["status"], "accepted"); + + h.policy.subscribed.lock().unwrap().remove(&channel); + let closed = signed_event(&keys, KIND_STREAM_MESSAGE, "closed", channel); + let ack = h.exchange(frame_for(&closed, channel, "inbound")).await; + assert_eq!(ack["reason"], "channel_not_subscribed"); + } + + #[tokio::test] + async fn unsubscribed_channel_rejected() { + let keys = Keys::generate(); + let seat = Keys::generate(); + let channel = Uuid::new_v4(); + let h = Harness::new(current_uid(), HashSet::new(), &seat, EnqueueResult::Queued); + let event = signed_event(&keys, KIND_STREAM_MESSAGE, "hello", channel); + let ack = h.exchange(frame_for(&event, channel, "inbound")).await; + assert_eq!(ack["reason"], "channel_not_subscribed"); + assert!(h.queue.lock().unwrap().is_empty()); + assert_eq!(h.store.lock().unwrap().len(), 0); + } + + #[tokio::test] + async fn membership_kind_rejected() { + let keys = Keys::generate(); + let seat = Keys::generate(); + let channel = Uuid::new_v4(); + let h = Harness::new( + current_uid(), + HashSet::from([channel]), + &seat, + EnqueueResult::Queued, + ); + let event = signed_event(&keys, 44100, "membership", channel); + let ack = h.exchange(frame_for(&event, channel, "inbound")).await; + assert_eq!(ack["reason"], "wake_denied"); + assert!(h.queue.lock().unwrap().is_empty()); + assert_eq!(h.store.lock().unwrap().len(), 0); + } + + #[tokio::test] + async fn dedupe_key_mismatch_rejected() { + let keys = Keys::generate(); + let seat = Keys::generate(); + let channel = Uuid::new_v4(); + let h = Harness::new( + current_uid(), + HashSet::from([channel]), + &seat, + EnqueueResult::Queued, + ); + let event = signed_event(&keys, KIND_STREAM_MESSAGE, "hello", channel); + let mut frame = frame_for(&event, channel, "inbound"); + frame["dedupe_key"] = json!("ab".repeat(32)); + let ack = h.exchange(frame).await; + assert_eq!(ack["reason"], "dedupe_key_mismatch"); + assert_eq!(h.store.lock().unwrap().len(), 0); + } + + #[tokio::test] + async fn self_echo_reaches_common_ignore_self_policy() { + let seat = Keys::generate(); + let channel = Uuid::new_v4(); + let h = Harness::new( + current_uid(), + HashSet::from([channel]), + &seat, + EnqueueResult::Queued, + ); + let event = signed_event(&seat, KIND_STREAM_MESSAGE, "loop", channel); + let ack = h.exchange(frame_for(&event, channel, "inbound")).await; + assert_eq!(ack["status"], "accepted"); + assert_eq!(h.queue.lock().unwrap().len(), 1); + } + + #[tokio::test] + async fn peer_uid_mismatch_rejected() { + let keys = Keys::generate(); + let seat = Keys::generate(); + let channel = Uuid::new_v4(); + let h = Harness::new( + current_uid().wrapping_add(1), + HashSet::from([channel]), + &seat, + EnqueueResult::Queued, + ); + let event = signed_event(&keys, KIND_STREAM_MESSAGE, "hello", channel); + let ack = h.exchange(frame_for(&event, channel, "inbound")).await; + assert_eq!(ack["reason"], "peer_uid"); + assert_eq!(h.store.lock().unwrap().len(), 0); + } + + #[tokio::test] + async fn queue_drop_reports_duplicate() { + let keys = Keys::generate(); + let seat = Keys::generate(); + let channel = Uuid::new_v4(); + let h = Harness::new( + current_uid(), + HashSet::from([channel]), + &seat, + EnqueueResult::Drop, + ); + let event = signed_event(&keys, KIND_STREAM_MESSAGE, "hello", channel); + let ack = h.exchange(frame_for(&event, channel, "inbound")).await; + assert_eq!(ack["status"], "duplicate"); + assert_eq!(ack["reason"], "queue_drop"); + assert_eq!(h.store.lock().unwrap().len(), 1); + assert!(h.queue.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn ack_write_failure_keeps_occupancy() { + let keys = Keys::generate(); + let seat = Keys::generate(); + let channel = Uuid::new_v4(); + let h = Harness::new( + current_uid(), + HashSet::from([channel]), + &seat, + EnqueueResult::Queued, + ); + let event = signed_event(&keys, KIND_STREAM_MESSAGE, "hello", channel); + let frame = frame_for(&event, channel, "inbound"); + let mut client = UnixStream::connect(&h.path).await.unwrap(); + let accept = h.accept_one(); + let send = async { + client.write_all(&encode_frame(&frame)).await.unwrap(); + drop(client); + }; + tokio::join!(accept, send); + assert_eq!(h.store.lock().unwrap().len(), 1); + assert_eq!(h.queue.lock().unwrap().len(), 1); + } + + #[tokio::test] + async fn frame_too_large_rejected() { + let seat = Keys::generate(); + let channel = Uuid::new_v4(); + let h = Harness::new( + current_uid(), + HashSet::from([channel]), + &seat, + EnqueueResult::Queued, + ); + let mut client = UnixStream::connect(&h.path).await.unwrap(); + let accept = h.accept_one(); + let send = async { + let size = (FORWARD_FRAME_MAX as u32) + 1; + client.write_all(&size.to_be_bytes()).await.unwrap(); + client.write_all(&[0u8; 8]).await.ok(); + read_ack(&mut client).await + }; + let (_, ack) = tokio::join!(accept, send); + assert_eq!(ack["reason"], "frame_too_large"); + assert_eq!(h.store.lock().unwrap().len(), 0); + } + + #[tokio::test] + async fn h_tag_from_another_channel_rejected() { + let keys = Keys::generate(); + let seat = Keys::generate(); + let subscribed = Uuid::new_v4(); + let other = Uuid::new_v4(); + let h = Harness::new( + current_uid(), + HashSet::from([subscribed]), + &seat, + EnqueueResult::Queued, + ); + let event = signed_event(&keys, KIND_STREAM_MESSAGE, "hello", other); + let ack = h.exchange(frame_for(&event, subscribed, "inbound")).await; + assert_eq!(ack["status"], "rejected"); + assert_eq!(ack["reason"], "channel_mismatch"); + assert!(h.queue.lock().unwrap().is_empty()); + assert_eq!(h.store.lock().unwrap().len(), 0); + } + + #[tokio::test] + async fn relay_first_then_forward_is_duplicate() { + let keys = Keys::generate(); + let seat = Keys::generate(); + let channel = Uuid::new_v4(); + let h = Harness::new( + current_uid(), + HashSet::from([channel]), + &seat, + EnqueueResult::Queued, + ); + let event = signed_event(&keys, KIND_STREAM_MESSAGE, "hello", channel); + let key = relay_dedupe_key(&event, channel); + { + let mut store = h.store.lock().unwrap(); + assert!(store.reserve(&key)); + } + let persist_path = h.store.lock().unwrap().path().to_path_buf(); + persist_dedupe_key(persist_path, key).await.unwrap(); + let ack = h.exchange(frame_for(&event, channel, "inbound")).await; + assert_eq!(ack["status"], "duplicate"); + assert!(ack["reason"].is_null()); + assert!(h.queue.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn forward_first_then_relay_skips() { + let keys = Keys::generate(); + let seat = Keys::generate(); + let channel = Uuid::new_v4(); + let h = Harness::new( + current_uid(), + HashSet::from([channel]), + &seat, + EnqueueResult::Queued, + ); + let event = signed_event(&keys, KIND_STREAM_MESSAGE, "hello", channel); + let ack = h.exchange(frame_for(&event, channel, "inbound")).await; + assert_eq!(ack["status"], "accepted"); + let store = h.store.lock().unwrap(); + assert!(relay_event_is_duplicate(&store, &event, channel)); + assert!(!{ + let mut clone = store.clone(); + clone.reserve(&relay_dedupe_key(&event, channel)) + }); + } + + #[tokio::test] + async fn noncanonical_channel_id_rejected() { + let keys = Keys::generate(); + let seat = Keys::generate(); + let channel = Uuid::new_v4(); + let h = Harness::new( + current_uid(), + HashSet::from([channel]), + &seat, + EnqueueResult::Queued, + ); + let event = signed_event(&keys, KIND_STREAM_MESSAGE, "hello", channel); + let variants = [ + channel.to_string().to_uppercase(), + format!("{{{channel}}}"), + format!("urn:uuid:{channel}"), + ]; + for raw in variants { + let mut frame = frame_for(&event, channel, "inbound"); + frame["channel_id"] = json!(raw); + let ack = h.exchange(frame).await; + assert_eq!(ack["reason"], "bad_frame", "raw channel_id should reject"); + } + assert!(h.queue.lock().unwrap().is_empty()); + assert_eq!(h.store.lock().unwrap().len(), 0); + } + + #[tokio::test] + async fn outbound_direction_rejected() { + let keys = Keys::generate(); + let seat = Keys::generate(); + let channel = Uuid::new_v4(); + let h = Harness::new( + current_uid(), + HashSet::from([channel]), + &seat, + EnqueueResult::Queued, + ); + let event = signed_event(&keys, KIND_STREAM_MESSAGE, "hello", channel); + let ack = h.exchange(frame_for(&event, channel, "outbound")).await; + assert_eq!(ack["status"], "rejected"); + assert_eq!(ack["reason"], "bad_frame"); + assert_eq!(h.store.lock().unwrap().len(), 0); + } + + #[tokio::test] + async fn non_json_payload_is_bad_frame() { + let seat = Keys::generate(); + let channel = Uuid::new_v4(); + let h = Harness::new( + current_uid(), + HashSet::from([channel]), + &seat, + EnqueueResult::Queued, + ); + let mut client = UnixStream::connect(&h.path).await.unwrap(); + let accept = h.accept_one(); + let send = async { + let blob = b"not-json"; + let mut out = Vec::new(); + out.extend_from_slice(&(blob.len() as u32).to_be_bytes()); + out.extend_from_slice(blob); + client.write_all(&out).await.unwrap(); + read_ack(&mut client).await + }; + let (_, ack) = tokio::join!(accept, send); + assert_eq!(ack["reason"], "bad_frame"); + assert_eq!(h.store.lock().unwrap().len(), 0); + } + + #[tokio::test] + async fn connect_and_send_nothing_times_out_without_ack() { + let seat = Keys::generate(); + let channel = Uuid::new_v4(); + let h = Harness::new( + current_uid(), + HashSet::from([channel]), + &seat, + EnqueueResult::Queued, + ); + let mut client = UnixStream::connect(&h.path).await.unwrap(); + let accept = h.accept_one(); + let send = async { + let mut header = [0u8; 4]; + let res = tokio::time::timeout( + ACK_TIMEOUT + Duration::from_secs(2), + client.read_exact(&mut header), + ) + .await; + if let Ok(Ok(_)) = res { + panic!("timeout path must not send an ack"); + } + }; + tokio::join!(accept, send); + assert_eq!(h.store.lock().unwrap().len(), 0); + } + + #[tokio::test] + async fn infra_failure_closes_without_ack() { + let keys = Keys::generate(); + let seat = Keys::generate(); + let channel = Uuid::new_v4(); + let h = Harness::new( + current_uid(), + HashSet::from([channel]), + &seat, + EnqueueResult::InfraFailure, + ); + let event = signed_event(&keys, KIND_STREAM_MESSAGE, "hello", channel); + let frame = frame_for(&event, channel, "inbound"); + let mut client = UnixStream::connect(&h.path).await.unwrap(); + let accept = h.accept_one(); + let send = async { + client.write_all(&encode_frame(&frame)).await.unwrap(); + let mut header = [0u8; 4]; + let res = + tokio::time::timeout(Duration::from_secs(2), client.read_exact(&mut header)).await; + if let Ok(Ok(_)) = res { + panic!("infra failure must not ack"); + } + }; + tokio::join!(accept, send); + assert_eq!(h.store.lock().unwrap().len(), 0); + assert!(h.queue.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn persist_fail_after_queued_keeps_reservation() { + let keys = Keys::generate(); + let seat = Keys::generate(); + let channel = Uuid::new_v4(); + let h = Harness::new( + current_uid(), + HashSet::from([channel]), + &seat, + EnqueueResult::Queued, + ); + h.block_persist(); + let event = signed_event(&keys, KIND_STREAM_MESSAGE, "hello", channel); + let frame = frame_for(&event, channel, "inbound"); + let mut client = UnixStream::connect(&h.path).await.unwrap(); + let accept = h.accept_one(); + let send = async { + client.write_all(&encode_frame(&frame)).await.unwrap(); + let mut header = [0u8; 4]; + let res = + tokio::time::timeout(Duration::from_secs(2), client.read_exact(&mut header)).await; + if let Ok(Ok(_)) = res { + panic!("persist failure after queued must not ack"); + } + }; + tokio::join!(accept, send); + assert_eq!(h.store.lock().unwrap().len(), 1); + assert_eq!(h.queue.lock().unwrap().len(), 1); + let mut retry = UnixStream::connect(&h.path).await.unwrap(); + let accept_retry = h.accept_one(); + let send_retry = async { + retry.write_all(&encode_frame(&frame)).await.unwrap(); + let mut header = [0u8; 4]; + let res = + tokio::time::timeout(Duration::from_secs(2), retry.read_exact(&mut header)).await; + if let Ok(Ok(_)) = res { + panic!("duplicate persist retry still failing must not ack"); + } + }; + tokio::join!(accept_retry, send_retry); + h.unblock_persist(); + let ack = h.exchange(frame).await; + assert_eq!(ack["status"], "duplicate"); + assert_eq!(h.queue.lock().unwrap().len(), 1); + } + + #[tokio::test] + async fn persist_fail_after_relay_reserve_forward_is_duplicate() { + let keys = Keys::generate(); + let seat = Keys::generate(); + let channel = Uuid::new_v4(); + let h = Harness::new( + current_uid(), + HashSet::from([channel]), + &seat, + EnqueueResult::Queued, + ); + let event = signed_event(&keys, KIND_STREAM_MESSAGE, "hello", channel); + let key = relay_dedupe_key(&event, channel); + { + let mut store = h.store.lock().unwrap(); + assert!(store.reserve(&key)); + } + h.block_persist(); + let err = persist_occupancy(&h.store, key.clone()).await; + assert!(err.is_err()); + assert_eq!(h.store.lock().unwrap().len(), 1); + assert!(relay_event_is_duplicate( + &h.store.lock().unwrap(), + &event, + channel + )); + h.unblock_persist(); + let ack = h.exchange(frame_for(&event, channel, "inbound")).await; + assert_eq!(ack["status"], "duplicate"); + assert!(h.queue.lock().unwrap().is_empty()); + } + + #[test] + fn load_survives_truncated_last_line() { + let path = PathBuf::from(format!( + "/tmp/bz-dedupe-trunc-{}-{}.jsonl", + std::process::id(), + SOCK_SEQ.fetch_add(1, Ordering::Relaxed) + )); + let full = "ab".repeat(32); + std::fs::write(&path, format!("{full}\ntruncated")).unwrap(); + let store = DedupeStore::load(&path).unwrap(); + assert!(store.contains(&full)); + assert!(!store.contains("truncated")); + let _ = std::fs::remove_file(&path); + } + + #[test] + fn load_caps_memory_to_last_n_keys() { + let path = PathBuf::from(format!( + "/tmp/bz-dedupe-cap-{}-{}.jsonl", + std::process::id(), + SOCK_SEQ.fetch_add(1, Ordering::Relaxed) + )); + let k1 = "aa".repeat(32); + let k2 = "bb".repeat(32); + let k3 = "cc".repeat(32); + std::fs::write(&path, format!("{k1}\n{k2}\n{k3}\n")).unwrap(); + let store = DedupeStore::load_with_cap(&path, 2).unwrap(); + assert!(!store.contains(&k1)); + assert!(store.contains(&k2)); + assert!(store.contains(&k3)); + let _ = std::fs::remove_file(&path); + } + + #[tokio::test] + async fn bind_preserves_regular_files_and_temporary_collisions() { + let path = PathBuf::from(format!("/tmp/bz-bind-{}.sock", Uuid::new_v4())); + std::fs::write(&path, b"keep").unwrap(); + assert!(bind_listener(&path).is_err()); + assert_eq!(std::fs::read(&path).unwrap(), b"keep"); + std::fs::remove_file(&path).unwrap(); + let tmp = path.with_extension(format!("tmp-{}", std::process::id())); + std::fs::write(&tmp, b"also keep").unwrap(); + assert!(bind_listener(&path).is_err()); + assert_eq!(std::fs::read(&tmp).unwrap(), b"also keep"); + std::fs::remove_file(&tmp).unwrap(); + } + + #[tokio::test] + async fn bind_refuses_to_replace_live_listener() { + let path = PathBuf::from(format!("/tmp/bz-live-{}.sock", Uuid::new_v4())); + let listener = bind_listener(&path).unwrap(); + assert!(bind_listener(&path).is_err()); + assert!(UnixStream::connect(&path).await.is_ok()); + drop(listener); + std::fs::remove_file(&path).unwrap(); + } + + #[tokio::test] + async fn concurrent_appends_reload_every_complete_key() { + let path = PathBuf::from(format!("/tmp/bz-append-{}.jsonl", Uuid::new_v4())); + let mut tasks = tokio::task::JoinSet::new(); + for i in 0..128 { + tasks.spawn(persist_dedupe_key(path.clone(), format!("{i:064x}"))); + } + while let Some(result) = tasks.join_next().await { + result.unwrap().unwrap(); + } + let store = DedupeStore::load(&path).unwrap(); + assert_eq!(store.len(), 128); + for i in 0..128 { + assert!(store.contains(&format!("{i:064x}"))); + } + std::fs::remove_file(path).unwrap(); + } +} diff --git a/crates/buzz-acp/src/intake.rs b/crates/buzz-acp/src/intake.rs new file mode 100644 index 00000000000..45f346569e9 --- /dev/null +++ b/crates/buzz-acp/src/intake.rs @@ -0,0 +1,22 @@ +//! Typed admission results shared by relay and local forward ingress. +use crate::relay::BuzzEvent; +use tokio::sync::oneshot; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum EnqueueResult { + Queued, + /// Policy `DedupMode::Drop` discarded the in-flight channel event. + Drop, + /// Occupied but not queued (filter / author / self). Ack is still accepted. + Ignored, + /// Queue never happened (main loop gone, channel closed, oneshot dropped). + /// Must not be acked — the gateway treats silence as delivery-unknown. + #[cfg(unix)] + InfraFailure, +} + +#[derive(Debug)] +pub struct ForwardWork { + pub event: BuzzEvent, + pub reply: oneshot::Sender, +} diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 3ff1dc39898..6691feabe41 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -4,6 +4,9 @@ mod acp; mod config; mod engram_fetch; mod filter; +#[cfg(unix)] +mod forward; +mod intake; mod observer; mod pool; mod pool_lifecycle; @@ -37,7 +40,12 @@ use config::{ MultipleEventHandling, RespondTo, SubscribeMode, }; use filter::SubscriptionRule; +#[cfg(unix)] +use forward::{ + accept_loop, bind_listener, persist_occupancy, relay_dedupe_key, DedupeStore, ForwardPolicy, +}; use futures_util::FutureExt; +use intake::{EnqueueResult, ForwardWork}; use nostr::{PublicKey, ToBech32}; use pool::{ AgentPool, ControlSignal, IdleSwitchResult, OwnedAgent, PromptContext, PromptOutcome, @@ -474,11 +482,14 @@ mod inbound_author_gate { // Retry failed startup discovery on generation 0 as well as failed // reconnect refreshes. Only an authoritative result completes the // generation; transient failure retains the last verified key. - if refresh_needed(self.refreshed_generation, buzz_event.connection_generation) { + if buzz_event.connection_generation == u64::MAX + || refresh_needed(self.refreshed_generation, buzz_event.connection_generation) + { let (relay_self, completed) = refresh_relay_self(rest_client, self.relay_self.take(), "listener").await; self.relay_self = relay_self; - if completed { + if completed && buzz_event.connection_generation != u64::MAX { + // Forward input must not poison the relay generation cache. self.refreshed_generation = Some(buzz_event.connection_generation); } } @@ -2509,6 +2520,10 @@ async fn tokio_main() -> Result<()> { } tracing::info!("buzz-acp starting: {}", config.summary()); + tracing::info!( + "{}", + config::input_paths_log_line(config.relay_input, config.forward_socket.is_some()) + ); let observer = config .relay_observer @@ -2572,6 +2587,7 @@ async fn tokio_main() -> Result<()> { HarnessRelay::connect(&config.relay_url, &config.keys, &pubkey_hex, relay_auth_tag) .await .map_err(|e| anyhow::anyhow!("relay connect error: {e}"))?; + relay.set_inbound_subscribe(config.relay_input); // Tell the relay background task the watermark so it can use // `since = watermark - 5s` on the first REQ instead of `since=now`. @@ -2709,12 +2725,17 @@ async fn tokio_main() -> Result<()> { if channel_filters.is_empty() { tracing::warn!("no channel subscriptions resolved — agent will sit idle"); } - let mut subscribed_channel_ids = HashSet::with_capacity(channel_filters.len()); + let subscribed_channel_ids = Arc::new(std::sync::Mutex::new(HashSet::with_capacity( + channel_filters.len(), + ))); 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 { - subscribed_channel_ids.insert(*channel_id); + subscribed_channel_ids + .lock() + .unwrap_or_else(|e| e.into_inner()) + .insert(*channel_id); tracing::info!("subscribed to channel {channel_id}"); } } @@ -2737,6 +2758,39 @@ async fn tokio_main() -> Result<()> { let mut queue = EventQueue::new(dedup_mode).with_in_flight_deadline(config.max_turn_duration_secs); + #[cfg(unix)] + let forward_store = if config.forward_socket.is_some() { + let path = config + .forward_state + .as_ref() + .ok_or_else(|| anyhow::anyhow!("missing forward state path"))?; + Some(Arc::new(std::sync::Mutex::new(DedupeStore::load(path)?))) + } else { + None + }; + #[cfg(unix)] + let mut forward_rx = if let Some(path) = &config.forward_socket { + let uid = config + .forward_peer_uid + .ok_or_else(|| anyhow::anyhow!("missing forward peer UID"))?; + let store = forward_store + .clone() + .ok_or_else(|| anyhow::anyhow!("missing forward store"))?; + let (tx, rx) = mpsc::channel::(32); + let policy = ForwardPolicy { + peer_uid: uid, + subscribed: subscribed_channel_ids.clone(), + store, + }; + let listener = bind_listener(path)?; + tokio::spawn(accept_loop(listener, policy, tx)); + Some(rx) + } else { + None + }; + #[cfg(not(unix))] + let mut forward_rx: Option> = None; + // Online means the harness can receive work, not merely that its socket is // connected. Publishing after channel subscriptions gives desktop callers // a durable readiness boundary before they send a startup mention. @@ -3215,11 +3269,29 @@ async fn tokio_main() -> Result<()> { None } // Remaining branches don't touch pool — evaluated when pool is idle. - buzz_event = relay.next_event() => { + incoming = async { + tokio::select! { + event = relay.next_event() => (event, None), + work = async { match forward_rx.as_mut() { + Some(rx) => rx.recv().await, + None => std::future::pending().await, + }} => match work { + Some(work) => (Some(work.event), Some(work.reply)), + None => std::future::pending().await, + } + } + } => { + let (buzz_event, forward_reply) = incoming; let _ = result_rx; // end split borrow before relay handling match buzz_event { Some(buzz_event) => { let kind_u32 = buzz_event.event.kind.as_u16() as u32; + if forward_reply.is_some() && !subscribed_channel_ids.lock().unwrap_or_else(|e| e.into_inner()).contains(&buzz_event.channel_id) { + if let Some(reply) = forward_reply { let _ = reply.send(EnqueueResult::Ignored); } + continue; + } + if forward_reply.is_none() && !config.relay_input && !matches!(kind_u32, KIND_MEMBER_ADDED_NOTIFICATION | KIND_MEMBER_REMOVED_NOTIFICATION) { continue; } + if kind_u32 == KIND_MEMBER_ADDED_NOTIFICATION || kind_u32 == KIND_MEMBER_REMOVED_NOTIFICATION @@ -3278,20 +3350,20 @@ async fn tokio_main() -> Result<()> { // stripped for a legitimately re-added channel. removed_channels.remove(&ch); - if subscribed_channel_ids.contains(&ch) { + if subscribed_channel_ids.lock().unwrap_or_else(|e| e.into_inner()).contains(&ch) { tracing::debug!(channel_id = %ch, "membership notification: channel already subscribed"); } else if let Some(filter) = config::resolve_dynamic_channel_filter(&config, ch, &rules) { tracing::info!(channel_id = %ch, "membership notification: subscribing to new channel"); if let Err(e) = relay.subscribe_channel_from(ch, filter, Some(ts)).await { tracing::warn!("failed to subscribe to new channel {ch}: {e}"); } else { - subscribed_channel_ids.insert(ch); + subscribed_channel_ids.lock().unwrap_or_else(|e| e.into_inner()).insert(ch); } } else { tracing::debug!(channel_id = %ch, "membership notification: no matching rules — skipping"); } } else { - subscribed_channel_ids.remove(&ch); + subscribed_channel_ids.lock().unwrap_or_else(|e| e.into_inner()).remove(&ch); tracing::info!(channel_id = %ch, "membership notification: unsubscribing from channel"); if let Err(e) = relay.unsubscribe_channel(ch).await { tracing::warn!("failed to unsubscribe from channel {ch}: {e}"); @@ -3339,226 +3411,21 @@ async fn tokio_main() -> Result<()> { continue; } - if config.ignore_self && buzz_event.event.pubkey.to_hex() == pubkey_hex { - tracing::debug!(channel_id = %buzz_event.channel_id, "dropping self-authored event"); - continue; - } - - // Check: kind:9, content "!shutdown", from owner, mentions THIS agent. - let is_shutdown = is_owner_control_command( - &buzz_event.event, - kind_u32, - "!shutdown", - &pubkey_hex, - ); - if is_shutdown { - let owner = owner_cache.get(); - if let Some(owner) = owner { - if buzz_event.event.pubkey.to_hex() == *owner { - tracing::info!( - channel_id = %buzz_event.channel_id, - sender = %buzz_event.event.pubkey.to_hex(), - "shutdown command from owner — exiting gracefully" - ); - let _ = shutdown_tx.send(()); - continue; - } - } - // Not from owner — fall through to normal prompt handling. - // Don't drop it — it's a regular message that happens to - // contain "!shutdown" from a non-owner. - } - - // Mirrors !shutdown: kind:9, content "!cancel", from - // owner, mentions THIS agent. Must be BEFORE - // queue.push() — the event content is moved by push. - // - // Mode-independent: !cancel fires regardless of - // --multiple-event-handling. It is explicit user - // intent, not an automatic policy decision. - let is_cancel = is_owner_control_command( - &buzz_event.event, - kind_u32, - "!cancel", - &pubkey_hex, - ); - if is_cancel { - let from_owner = owner_cache.get().is_some_and(|owner| { - buzz_event.event.pubkey.to_hex() == *owner - }); - if from_owner { - // Scope-exact: an owner's !cancel in thread A - // must cancel thread A's turn, never a sibling - // thread running in the same channel. Under - // the default channel policy the scope is the - // channel's sole conversation, so this is - // byte-for-byte the prior behavior. - let scope = scope::SessionScope::derive( - config.session_policy, - buzz_event.channel_id, - is_dm_channel(buzz_event.channel_id, &ctx.channel_info) - .await, - &buzz_event.event, - ); - let fired = signal_in_flight_task_for_scope( - &mut pool, - &scope, - ControlSignal::Cancel, - ); - if !fired { - tracing::warn!( - channel_id = %buzz_event.channel_id, - scope = %scope.telemetry_label(), - "!cancel received but no in-flight task — no-op" - ); - } - continue; // consume event — do NOT push to queue - } - // Not from owner — fall through to normal prompt handling. - } - - // Mirrors !shutdown / !cancel: kind:9, content - // "!rotate", from owner, mentions THIS agent. - // - // Rotation is explicit owner intent to start the - // next turn in this channel with a fresh ACP - // session. It is consumed by the harness and never - // forwarded to the agent. If a turn is in-flight, - // cancel it, drop its triggering batch, and - // invalidate the channel session when the task - // returns. If idle, invalidate the cached channel - // session immediately. Queued future events remain - // queued and will create a fresh session on dispatch. - let is_rotate = is_owner_control_command( - &buzz_event.event, - kind_u32, - "!rotate", - &pubkey_hex, - ); - if is_rotate { - let from_owner = owner_cache.get().is_some_and(|owner| { - buzz_event.event.pubkey.to_hex() == *owner - }); - if from_owner { - // Scope-exact: rotate only the thread the - // owner's !rotate belongs to. Under the - // default channel policy the scope is the - // channel's sole conversation, matching the - // prior channel-wide rotate. - let scope = scope::SessionScope::derive( - config.session_policy, - buzz_event.channel_id, - is_dm_channel(buzz_event.channel_id, &ctx.channel_info) - .await, - &buzz_event.event, - ); - let fired = signal_in_flight_task_for_scope( - &mut pool, - &scope, - ControlSignal::Rotate, - ); - if fired { - tracing::info!( - channel_id = %buzz_event.channel_id, - scope = %scope.telemetry_label(), - "!rotate received — cancelling in-flight turn and rotating session" - ); - } else { - let invalidated = - pool.invalidate_scope_session(&scope); - tracing::info!( - channel_id = %buzz_event.channel_id, - scope = %scope.telemetry_label(), - invalidated, - "!rotate received — invalidated idle session for scope" - ); - } - continue; // consume event — do NOT push to queue - } - // Not from owner — fall through to normal prompt handling. + #[cfg(unix)] + let occupancy = if forward_reply.is_none() { + forward_store.as_ref().map(|store| (store.clone(), relay_dedupe_key(&buzz_event.event, buzz_event.channel_id))) + } else { None }; + #[cfg(unix)] + if let Some((store, key)) = &occupancy { + if !store.lock().unwrap_or_else(|e| e.into_inner()).reserve(key) { continue; } } - - // Coarse security policy: drop events from disallowed - // authors before they reach subscription rules or the - // agent. Must be AFTER !shutdown (owner can always - // shut down regardless of gate mode). - // - // Both OwnerOnly and Allowlist accept events from - // "siblings" — pubkeys whose agent_owner_pubkey - // matches this agent's owner (e.g. other bots - // launched by the same human). Allowlist adds the - // explicit pubkey list on top, for external people; - // it never revokes same-owner team bots. - let Some(authorized_event) = authorize_normal_listener_event( - &mut author_gate_ctx, - buzz_event, - &config.respond_to, - &config.respond_to_allowlist, - &owner_cache, - &ctx.channel_info, - &ctx.rest_client, - ) - .await - else { - continue; - }; - let Some(ingress) = - AuthorizedNormalListenerEvent(authorized_event) - .match_subscription(&rules, &pubkey_hex) - .await - else { - tracing::debug!("authorized event matched no rule — dropping"); - continue; - }; - // Derive the session scope once, at admission, from - // the operator policy, DM status, and NIP-10 thread - // tags. Under the default `channel` policy this is - // always a conversation scope, preserving today's - // channel-keyed routing. Telemetry only for now — - // queue/pool partitioning by scope lands in a - // follow-up (see ticket outline steps 2–4). - let session_scope = scope::SessionScope::derive( - config.session_policy, - ingress.buzz_event.channel_id, - is_dm_channel( - ingress.buzz_event.channel_id, - &ctx.channel_info, - ) - .await, - &ingress.buzz_event.event, - ); - tracing::debug!( - channel_id = %session_scope.channel_id(), - scope = %session_scope.telemetry_label(), - thread_scoped = session_scope.is_thread(), - thread_root = session_scope.root_event_id().unwrap_or("-"), - policy = %config.session_policy, - "admitted event — resolved session scope" - ); - let queued = ingress.push(&mut queue, session_scope); - // 👀 — immediate "seen" reaction, only if the event - // was actually queued (not dropped by DedupMode::Drop). - // Fire-and-forget: on rare fast-failure paths the - // guard's cleanup may race with this add, leaving a - // cosmetic stale 👀. Acceptable — see ReactionGuard docs. - queued.mark_seen(&ctx.rest_client); - // Event is already queued. The authorized ingress - // retains its verified author, resolved scope, and - // event data through the optional steer/interrupt - // decision. - queued.steer_or_interrupt( - config.multiple_event_handling, - owner_cache.get(), - &mut pool, - &mut queue, - &steer_ack_tx, - ); - if pool_ready { - for (scope, thread_tags) in - dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity, observer.as_ref()) - { - typing_channels.insert(scope, thread_tags); - } + let outcome = admit_inbound(buzz_event, &config, &pubkey_hex, &owner_cache, + &shutdown_tx, &mut author_gate_ctx, &mut pool, &mut queue, &ctx, &rules, + &steer_ack_tx, pool_ready, &mut typing_channels, &mut last_activity, observer.as_ref()).await; + if let Some(reply) = forward_reply { let _ = reply.send(outcome); } + #[cfg(unix)] + if let Some((store, key)) = occupancy { + tokio::spawn(async move { let _ = persist_occupancy(&store, key).await; }); } } None => { @@ -9176,6 +9043,10 @@ mod build_mcp_servers_tests { agent_owner: None, no_base_prompt: false, base_prompt_content: None, + forward_socket: None, + forward_peer_uid: None, + forward_state: None, + relay_input: true, } } @@ -9402,6 +9273,10 @@ mod error_outcome_emission_tests { agent_owner: None, no_base_prompt: false, base_prompt_content: None, + forward_socket: None, + forward_peer_uid: None, + forward_state: None, + relay_input: true, } } @@ -11519,3 +11394,379 @@ mod observer_payload_trim_tests { assert!(leaf.contains("[elided")); } } + +/// Admit either transport through the same author, mention, scope and queue boundary. +/// A terminal result means admission/intentional filtering, never completed execution. +#[allow(clippy::too_many_arguments)] +async fn admit_inbound( + buzz_event: relay::BuzzEvent, + config: &Config, + pubkey_hex: &str, + owner_cache: &OwnerCache, + shutdown_tx: &watch::Sender<()>, + author_gate_ctx: &mut InboundAuthorGate, + pool: &mut AgentPool, + queue: &mut EventQueue, + ctx: &Arc, + rules: &[SubscriptionRule], + steer_ack_tx: &mpsc::UnboundedSender, + pool_ready: bool, + typing_channels: &mut HashMap, + last_activity: &mut tokio::time::Instant, + observer: Option<&observer::ObserverHandle>, +) -> EnqueueResult { + let kind_u32 = buzz_event.event.kind.as_u16() as u32; + if config.ignore_self && buzz_event.event.pubkey.to_hex() == pubkey_hex { + tracing::debug!(channel_id = %buzz_event.channel_id, "dropping self-authored event"); + return EnqueueResult::Ignored; + } + + // Check: kind:9, content "!shutdown", from owner, mentions THIS agent. + let is_shutdown = + is_owner_control_command(&buzz_event.event, kind_u32, "!shutdown", pubkey_hex); + if is_shutdown { + let owner = owner_cache.get(); + if let Some(owner) = owner { + if buzz_event.event.pubkey.to_hex() == *owner { + tracing::info!( + channel_id = %buzz_event.channel_id, + sender = %buzz_event.event.pubkey.to_hex(), + "shutdown command from owner — exiting gracefully" + ); + let _ = shutdown_tx.send(()); + return EnqueueResult::Ignored; + } + } + // Not from owner — fall through to normal prompt handling. + // Don't drop it — it's a regular message that happens to + // contain "!shutdown" from a non-owner. + } + + // Mirrors !shutdown: kind:9, content "!cancel", from + // owner, mentions THIS agent. Must be BEFORE + // queue.push() — the event content is moved by push. + // + // Mode-independent: !cancel fires regardless of + // --multiple-event-handling. It is explicit user + // intent, not an automatic policy decision. + let is_cancel = is_owner_control_command(&buzz_event.event, kind_u32, "!cancel", pubkey_hex); + if is_cancel { + let from_owner = owner_cache + .get() + .is_some_and(|owner| buzz_event.event.pubkey.to_hex() == *owner); + if from_owner { + // Scope-exact: an owner's !cancel in thread A + // must cancel thread A's turn, never a sibling + // thread running in the same channel. Under + // the default channel policy the scope is the + // channel's sole conversation, so this is + // byte-for-byte the prior behavior. + let scope = scope::SessionScope::derive( + config.session_policy, + buzz_event.channel_id, + is_dm_channel(buzz_event.channel_id, &ctx.channel_info).await, + &buzz_event.event, + ); + let fired = signal_in_flight_task_for_scope(pool, &scope, ControlSignal::Cancel); + if !fired { + tracing::warn!( + channel_id = %buzz_event.channel_id, + scope = %scope.telemetry_label(), + "!cancel received but no in-flight task — no-op" + ); + } + return EnqueueResult::Ignored; // consume event — do NOT push to queue + } + // Not from owner — fall through to normal prompt handling. + } + + // Mirrors !shutdown / !cancel: kind:9, content + // "!rotate", from owner, mentions THIS agent. + // + // Rotation is explicit owner intent to start the + // next turn in this channel with a fresh ACP + // session. It is consumed by the harness and never + // forwarded to the agent. If a turn is in-flight, + // cancel it, drop its triggering batch, and + // invalidate the channel session when the task + // returns. If idle, invalidate the cached channel + // session immediately. Queued future events remain + // queued and will create a fresh session on dispatch. + let is_rotate = is_owner_control_command(&buzz_event.event, kind_u32, "!rotate", pubkey_hex); + if is_rotate { + let from_owner = owner_cache + .get() + .is_some_and(|owner| buzz_event.event.pubkey.to_hex() == *owner); + if from_owner { + // Scope-exact: rotate only the thread the + // owner's !rotate belongs to. Under the + // default channel policy the scope is the + // channel's sole conversation, matching the + // prior channel-wide rotate. + let scope = scope::SessionScope::derive( + config.session_policy, + buzz_event.channel_id, + is_dm_channel(buzz_event.channel_id, &ctx.channel_info).await, + &buzz_event.event, + ); + let fired = signal_in_flight_task_for_scope(pool, &scope, ControlSignal::Rotate); + if fired { + tracing::info!( + channel_id = %buzz_event.channel_id, + scope = %scope.telemetry_label(), + "!rotate received — cancelling in-flight turn and rotating session" + ); + } else { + let invalidated = pool.invalidate_scope_session(&scope); + tracing::info!( + channel_id = %buzz_event.channel_id, + scope = %scope.telemetry_label(), + invalidated, + "!rotate received — invalidated idle session for scope" + ); + } + return EnqueueResult::Ignored; // consume event — do NOT push to queue + } + // Not from owner — fall through to normal prompt handling. + } + + // Coarse security policy: drop events from disallowed + // authors before they reach subscription rules or the + // agent. Must be AFTER !shutdown (owner can always + // shut down regardless of gate mode). + // + // Both OwnerOnly and Allowlist accept events from + // "siblings" — pubkeys whose agent_owner_pubkey + // matches this agent's owner (e.g. other bots + // launched by the same human). Allowlist adds the + // explicit pubkey list on top, for external people; + // it never revokes same-owner team bots. + let Some(authorized_event) = authorize_normal_listener_event( + author_gate_ctx, + buzz_event, + &config.respond_to, + &config.respond_to_allowlist, + owner_cache, + &ctx.channel_info, + &ctx.rest_client, + ) + .await + else { + return EnqueueResult::Ignored; + }; + let Some(ingress) = AuthorizedNormalListenerEvent(authorized_event) + .match_subscription(rules, pubkey_hex) + .await + else { + tracing::debug!("authorized event matched no rule — dropping"); + return EnqueueResult::Ignored; + }; + // Derive the session scope once, at admission, from + // the operator policy, DM status, and NIP-10 thread + // tags. Under the default `channel` policy this is + // always a conversation scope, preserving today's + // channel-keyed routing. Telemetry only for now — + // queue/pool partitioning by scope lands in a + // follow-up (see ticket outline steps 2–4). + let session_scope = scope::SessionScope::derive( + config.session_policy, + ingress.buzz_event.channel_id, + is_dm_channel(ingress.buzz_event.channel_id, &ctx.channel_info).await, + &ingress.buzz_event.event, + ); + tracing::debug!( + channel_id = %session_scope.channel_id(), + scope = %session_scope.telemetry_label(), + thread_scoped = session_scope.is_thread(), + thread_root = session_scope.root_event_id().unwrap_or("-"), + policy = %config.session_policy, + "admitted event — resolved session scope" + ); + let queued = ingress.push(queue, session_scope); + let outcome = if queued.accepted { + EnqueueResult::Queued + } else { + EnqueueResult::Drop + }; + // 👀 — immediate "seen" reaction, only if the event + // was actually queued (not dropped by DedupMode::Drop). + // Fire-and-forget: on rare fast-failure paths the + // guard's cleanup may race with this add, leaving a + // cosmetic stale 👀. Acceptable — see ReactionGuard docs. + queued.mark_seen(&ctx.rest_client); + // Event is already queued. The authorized ingress + // retains its verified author, resolved scope, and + // event data through the optional steer/interrupt + // decision. + queued.steer_or_interrupt( + config.multiple_event_handling, + owner_cache.get(), + pool, + queue, + steer_ack_tx, + ); + if pool_ready { + for (scope, thread_tags) in dispatch_pending(pool, queue, ctx, last_activity, observer) { + typing_channels.insert(scope, thread_tags); + } + } + outcome +} + +#[cfg(test)] +mod forward_admission_tests { + use super::*; + use nostr::{EventBuilder, Kind, Tag}; + + async fn admit_case( + mentioned: bool, + respond_to: RespondTo, + self_echo: bool, + ignore_self: bool, + ) -> (EnqueueResult, usize) { + use clap::Parser; + let seat = nostr::Keys::generate(); + let mut config = Config::from_args( + config::CliArgs::try_parse_from([ + "buzz-acp", + "--private-key", + &seat.secret_key().to_secret_hex(), + ]) + .expect("args"), + ) + .expect("config"); + config.ignore_self = ignore_self; + config.respond_to = respond_to; + let author = if self_echo { + seat.clone() + } else { + nostr::Keys::generate() + }; + let channel = Uuid::new_v4(); + let rest = relay::RestClient { + http: reqwest::Client::builder() + .timeout(Duration::from_millis(100)) + .build() + .expect("http"), + base_url: "http://127.0.0.1:0".into(), + keys: seat.clone(), + auth_tag_json: None, + }; + let info = pool::ChannelInfoResolver::new( + HashMap::from([( + channel, + relay::ChannelInfo { + name: "test".into(), + channel_type: "stream".into(), + description: None, + }, + )]), + rest.clone(), + ); + let ctx = Arc::new(PromptContext { + mcp_servers: vec![], + initial_message: None, + idle_timeout: Duration::from_secs(1), + max_turn_duration: Duration::from_secs(1), + turn_liveness_interval: Duration::ZERO, + dedup_mode: DedupMode::Queue, + system_prompt: None, + session_title: None, + team_instructions: None, + heartbeat_prompt: None, + base_prompt: None, + cwd: ".".into(), + rest_client: rest.clone(), + channel_info: info, + context_message_limit: 0, + max_turns_per_session: 0, + permission_mode: config::PermissionMode::Default, + agent_keys: seat.clone(), + agent_owner_pubkey: None, + memory_enabled: false, + harness_name: "test".into(), + relay_url: "ws://127.0.0.1:0".into(), + }); + let pk = seat.public_key().to_hex(); + let mut gate = InboundAuthorGate::connect(&rest, &pk, "test").await; + let owner = OwnerCache::new(Some(pk.clone())); + owner.cache_sibling(author.public_key().to_hex(), false); + let rules = vec![SubscriptionRule { + name: "mentions".into(), + channels: filter::ChannelScope::All("all".into()), + kinds: vec![9], + require_mention: true, + filter: None, + compiled_filter: None, + consecutive_timeouts: Arc::new(std::sync::atomic::AtomicU32::new(0)), + prompt_tag: None, + }]; + let mut tags = vec![Tag::parse(["h", &channel.to_string()]).expect("tag")]; + if mentioned { + tags.push(Tag::parse(["p", &pk]).expect("p tag")); + } + let event = EventBuilder::new(Kind::Custom(9), "hello") + .allow_self_tagging() + .tags(tags) + .sign_with_keys(&author) + .expect("sign"); + let event = relay::BuzzEvent { + connection_generation: u64::MAX, + channel_id: channel, + event, + }; + let (shutdown, _) = watch::channel(()); + let (steer, _) = mpsc::unbounded_channel(); + let mut pool = AgentPool::from_slots(vec![]); + let mut queue = EventQueue::new(DedupMode::Queue); + let mut typing = HashMap::new(); + let mut last_activity = tokio::time::Instant::now(); + let outcome = admit_inbound( + event, + &config, + &pk, + &owner, + &shutdown, + &mut gate, + &mut pool, + &mut queue, + &ctx, + &rules, + &steer, + false, + &mut typing, + &mut last_activity, + None, + ) + .await; + (outcome, queue.pending_channels()) + } + + #[tokio::test] + async fn socket_origin_cannot_bypass_mention_or_author_filter() { + assert_eq!( + admit_case(false, RespondTo::Anyone, false, true).await, + (EnqueueResult::Ignored, 0) + ); + assert_eq!( + admit_case(true, RespondTo::OwnerOnly, false, true).await, + (EnqueueResult::Ignored, 0) + ); + assert_eq!( + admit_case(true, RespondTo::Anyone, false, true).await, + (EnqueueResult::Queued, 1) + ); + } + + #[tokio::test] + async fn socket_origin_honors_configured_self_filter() { + assert_eq!( + admit_case(true, RespondTo::Anyone, true, true).await, + (EnqueueResult::Ignored, 0) + ); + assert_eq!( + admit_case(true, RespondTo::Anyone, true, false).await, + (EnqueueResult::Queued, 1) + ); + } +} diff --git a/crates/buzz-acp/src/relay.rs b/crates/buzz-acp/src/relay.rs index a019e758341..c3e2dd9e4bc 100644 --- a/crates/buzz-acp/src/relay.rs +++ b/crates/buzz-acp/src/relay.rs @@ -712,6 +712,10 @@ pub struct HarnessRelay { /// Wrapped in `Option` so `shutdown()` can take ownership without conflicting /// with `Drop` (which only has `&mut self`). bg_handle: Option>, + /// When false, channel and observer REQ subscriptions are not opened. + /// Membership REQ is independent (does not start a turn). Outbound + /// `PublishEvent` still goes through. Conventions §4 single input path. + inbound_subscribe: bool, } /// Cloneable publisher handle for signed events on the relay background socket. @@ -807,9 +811,18 @@ impl HarnessRelay { keys: keys.clone(), auth_tag, bg_handle: Some(bg_handle), + inbound_subscribe: true, }) } + /// Disable (or re-enable) inbound relay subscriptions. + /// + /// Outbound publish is unchanged. Used when `--relay-input false` so a + /// forward sidecar is the only intake path. + pub fn set_inbound_subscribe(&mut self, enabled: bool) { + self.inbound_subscribe = enabled; + } + /// Discover channels the agent is a member of. /// /// Queries kind:39002 (NIP-29 group members) events where `#p` includes @@ -914,6 +927,10 @@ impl HarnessRelay { filter: ChannelFilter, replay_since: Option, ) -> Result<(), RelayError> { + if !self.inbound_subscribe { + debug!("inbound subscribe disabled; skipping channel {channel_id}"); + return Ok(()); + } self.cmd_tx .send(RelayCommand::Subscribe { channel_id, @@ -927,6 +944,10 @@ impl HarnessRelay { } /// Subscribe to membership notifications for this agent. + /// + /// Membership REQ stays open even when channel inbound is off (§4 + /// single input: membership does not start a turn; it only updates + /// the subscribed set the forward gate reads). pub async fn subscribe_membership_notifications(&mut self) -> Result<(), RelayError> { self.cmd_tx .send(RelayCommand::SubscribeMembership) @@ -937,6 +958,10 @@ impl HarnessRelay { /// Subscribe to encrypted observer control frames addressed to this agent. pub async fn subscribe_observer_controls(&mut self) -> Result<(), RelayError> { + if !self.inbound_subscribe { + debug!("inbound subscribe disabled; skipping observer controls"); + return Ok(()); + } self.cmd_tx .send(RelayCommand::SubscribeObserverControls) .await @@ -6974,4 +6999,76 @@ mod tests { "channel_dropped_since must be cleared on successful drain" ); } + + fn test_stub_relay(inbound_subscribe: bool) -> (HarnessRelay, mpsc::Receiver) { + let (cmd_tx, cmd_rx) = mpsc::channel(16); + let (_event_tx, event_rx) = mpsc::channel(1); + let relay = HarnessRelay { + event_rx, + observer_control_rx: None, + cmd_tx, + http: reqwest::Client::new(), + relay_url: "ws://localhost:3000".into(), + keys: Keys::generate(), + auth_tag: None, + bg_handle: None, + inbound_subscribe, + }; + (relay, cmd_rx) + } + + #[tokio::test] + async fn inbound_off_keeps_membership_req_but_skips_channel_and_observer() { + let (mut relay, mut cmd_rx) = test_stub_relay(false); + relay + .subscribe_membership_notifications() + .await + .expect("membership stays on"); + let membership = cmd_rx.try_recv().expect("membership REQ queued"); + assert!(matches!(membership, RelayCommand::SubscribeMembership)); + relay + .subscribe_observer_controls() + .await + .expect("noop observer"); + relay + .subscribe_channel(Uuid::nil(), test_channel_filter()) + .await + .expect("noop channel"); + assert!( + cmd_rx.try_recv().is_err(), + "inbound-off must send 0 channel/observer subscribe commands" + ); + + let event = EventBuilder::new(Kind::TextNote, "outbound") + .tags([]) + .sign_with_keys(&relay.keys) + .expect("sign"); + relay + .event_publisher() + .publish_event(event) + .await + .expect("publish still works"); + let cmd = cmd_rx.try_recv().expect("publish command queued"); + assert!( + matches!(cmd, RelayCommand::PublishEvent { .. }), + "outbound publish must still reach the background task" + ); + } + + #[tokio::test] + async fn inbound_on_still_queues_subscribe_commands() { + let (mut relay, mut cmd_rx) = test_stub_relay(true); + relay + .subscribe_membership_notifications() + .await + .expect("membership"); + let cmd = cmd_rx.try_recv().expect("membership command queued"); + assert!(matches!(cmd, RelayCommand::SubscribeMembership)); + relay + .subscribe_channel(Uuid::nil(), test_channel_filter()) + .await + .expect("channel"); + let cmd = cmd_rx.try_recv().expect("channel command queued"); + assert!(matches!(cmd, RelayCommand::Subscribe { .. })); + } } diff --git a/crates/buzz-acp/src/setup_mode.rs b/crates/buzz-acp/src/setup_mode.rs index c924e26f0fd..30e49416cec 100644 --- a/crates/buzz-acp/src/setup_mode.rs +++ b/crates/buzz-acp/src/setup_mode.rs @@ -43,6 +43,18 @@ use nostr::EventId; use serde::{Deserialize, Serialize}; use uuid::Uuid; +/// Setup-mode's only intake is the relay. A launcher pin of `relay_input=false` +/// would wait forever on `next_event()`. Force inbound on; the single-input +/// path applies to ready mode only. +fn setup_mode_force_relay_inbound(relay_input: bool) -> bool { + if !relay_input { + tracing::warn!( + "setup-mode: relay input forced on — single input path applies to ready mode only" + ); + } + true +} + // ── Availability mirror ──────────────────────────────────────────────────────── /// Granular install/auth state for a CLI-backed ACP harness. @@ -342,6 +354,7 @@ pub(crate) async fn run_setup_listener(config: Config, payload: SetupPayload) -> HarnessRelay::connect(&config.relay_url, &config.keys, &pubkey_hex, relay_auth_tag) .await .map_err(|e| anyhow::anyhow!("setup-mode relay connect error: {e}"))?; + relay.set_inbound_subscribe(setup_mode_force_relay_inbound(config.relay_input)); if let Err(e) = relay.set_startup_watermark(startup_watermark).await { tracing::warn!("setup-mode: failed to set startup watermark: {e}"); @@ -702,6 +715,12 @@ async fn publish_setup_nudge( mod tests { use super::*; + #[test] + fn setup_mode_forces_relay_inbound_on_when_pin_is_off() { + assert!(setup_mode_force_relay_inbound(false)); + assert!(setup_mode_force_relay_inbound(true)); + } + #[test] fn setup_payload_from_raw_returns_none_when_absent() { // None → Ok(None): normal startup, no setup payload.