From 5a6c427c24d28cbccf22be18a70d8f7731666106 Mon Sep 17 00:00:00 2001 From: a-arad Date: Sun, 6 Sep 2026 22:36:28 -0400 Subject: [PATCH 1/3] feat(agents): recover private reminders as durable follow-ups Signed-off-by: a-arad --- Cargo.lock | 2 + crates/buzz-acp/Cargo.toml | 2 + crates/buzz-acp/README.md | 32 ++ crates/buzz-acp/src/base_prompt.md | 13 + crates/buzz-acp/src/lib.rs | 79 ++++- crates/buzz-acp/src/pool.rs | 39 ++- crates/buzz-acp/src/relay.rs | 9 + crates/buzz-acp/src/reminder_receipts.rs | 83 +++++ crates/buzz-acp/src/reminders.rs | 279 ++++++++++++++++ crates/buzz-acp/src/reminders_tests.rs | 200 ++++++++++++ crates/buzz-cli/src/commands/mod.rs | 1 + crates/buzz-cli/src/commands/reminders.rs | 304 ++++++++++++++++++ .../buzz-cli/src/commands/reminders_tests.rs | 91 ++++++ crates/buzz-cli/src/lib.rs | 5 + crates/buzz-sdk/src/lib.rs | 1 + crates/buzz-sdk/src/reminders.rs | 289 +++++++++++++++++ crates/buzz-sdk/src/reminders_json.rs | 82 +++++ crates/buzz-sdk/src/reminders_tests.rs | 190 +++++++++++ 18 files changed, 1680 insertions(+), 21 deletions(-) create mode 100644 crates/buzz-acp/src/reminder_receipts.rs create mode 100644 crates/buzz-acp/src/reminders.rs create mode 100644 crates/buzz-acp/src/reminders_tests.rs create mode 100644 crates/buzz-cli/src/commands/reminders.rs create mode 100644 crates/buzz-cli/src/commands/reminders_tests.rs create mode 100644 crates/buzz-sdk/src/reminders.rs create mode 100644 crates/buzz-sdk/src/reminders_json.rs create mode 100644 crates/buzz-sdk/src/reminders_tests.rs diff --git a/Cargo.lock b/Cargo.lock index d5f80a5fc82..c624fd8a9a3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -837,6 +837,7 @@ dependencies = [ "chrono", "clap", "evalexpr", + "fs2", "futures-util", "hex", "httparse", @@ -847,6 +848,7 @@ dependencies = [ "serde", "serde_json", "sha2 0.11.0", + "tempfile", "thiserror 2.0.18", "tokio", "tokio-tungstenite 0.29.0", diff --git a/crates/buzz-acp/Cargo.toml b/crates/buzz-acp/Cargo.toml index d047849806f..5d4d5b3c407 100644 --- a/crates/buzz-acp/Cargo.toml +++ b/crates/buzz-acp/Cargo.toml @@ -16,6 +16,7 @@ name = "buzz-acp" path = "src/main.rs" [dependencies] +fs2 = "0.4" # Internal buzz-core = { workspace = true } buzz-sdk = { workspace = true } @@ -77,5 +78,6 @@ evalexpr = { workspace = true } nix = { version = "0.31", default-features = false, features = ["signal"] } [dev-dependencies] +tempfile = "3" tokio = { workspace = true, features = ["test-util"] } httparse = "1" diff --git a/crates/buzz-acp/README.md b/crates/buzz-acp/README.md index 41d9a214bdd..d5856abe335 100644 --- a/crates/buzz-acp/README.md +++ b/crates/buzz-acp/README.md @@ -23,6 +23,38 @@ cargo build --release -p buzz-acp export PATH="$PWD/target/release:$PATH" ``` +## Private reminders + +Agents can schedule deferred follow-ups with `buzz reminders create --after 7d +--note 'Inspect experiment X and decide whether to continue' --link +'buzz://message?channel=&id='`. `--at` accepts an absolute RFC3339 +time with timezone. `list`, `get`, `snooze`, `complete`, and `cancel` manage the +author's encrypted NIP-ER state; no channel message is published automatically. + +On relays advertising NIP-ER and NIP-42, the harness queries the author's current +reminder heads every 30 seconds, with paginated recovery and no creation-time +lower bound. Due work uses the existing private session and agent pool after +queued messages, without interrupting active work. It rechecks the head before +dispatch, so snoozes and cancellations supersede waiting intent. The reminder +note and target provide context when the originating session no longer exists. + +A normally completed turn gets a durable delivery receipt; the agent separately +chooses whether to complete, snooze, or cancel the reminder. Pending reminders +already delivered remain inspectable with `buzz reminders list`; they do not +repeatedly wake the agent. Failed, interrupted, or limited turns remain eligible +with per-version backoff. A crash before the receipt is durable can redeliver: +agents should inspect their retained artifacts before repeating side effects. + +Receipts default to `$XDG_STATE_HOME/buzz-acp/reminders` or +`$HOME/.local/state/buzz-acp/reminders`. Set `BUZZ_ACP_REMINDER_STATE_DIR` to a +persistent volume in replaceable runtimes. Files are scoped by relay and author; +an exclusive local lock prevents competing consumers sharing that directory. +Keep one harness per identity; simultaneous devices do not have distributed +exactly-once delivery. Losing receipts can redeliver pending reminders, while +done/cancelled state remains on the relay. Bookmark reminders without a due time +never wake an agent. Relays lacking the advertised private-read contract disable +reminder recovery without changing ordinary messaging. + ## Generating Keys Each agent needs a Nostr keypair — this is the agent's identity in Buzz. Use `buzz-admin` to generate one: diff --git a/crates/buzz-acp/src/base_prompt.md b/crates/buzz-acp/src/base_prompt.md index 305d5b8880d..91c75112e4e 100644 --- a/crates/buzz-acp/src/base_prompt.md +++ b/crates/buzz-acp/src/base_prompt.md @@ -15,6 +15,7 @@ The `buzz` CLI is your primary interface. Auth env vars: `BUZZ_RELAY_URL`, `BUZZ | `buzz dms` | `list`, `open` | | `buzz users` | `get`, `set-profile`, `presence` | | `buzz workflows` | `list`, `trigger`, `runs` | +| `buzz reminders` | `create`, `list`, `get`, `snooze`, `complete`, `cancel` | | `buzz feed` | `get` | | `buzz social` | `publish`, `notes` | | `buzz repos` | `create`, `get`, `list` | @@ -49,6 +50,18 @@ Open an owner-reviewed draft with `buzz agents draft-create --channel &id='`. +Use `--at` with an RFC3339 timezone for an absolute time. The link is optional. +Due reminders return to your identity in a private session, including after a +harness restart; they do not restore the originating channel's in-memory context. +Inspect current evidence and complete, snooze, or cancel the reminder as appropriate. +Use existing completion notifications for jobs and peer replies; time reminders +are useful when waiting itself lets evidence accumulate. A reminder is your +retained intention to reconsider, not an obligation to carry out a stale plan. + ### Mentions - For a notifying `@mention`, use the person's **exact display name as shown in Buzz** (e.g., `@Alice Smith`, not `@Alice`, when the displayed name is `Alice Smith`). Do not expand a short display name, infer a surname, or spend tool calls looking for a “fuller” name merely to address someone. Partial names fail silently. diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index ddd594b142d..b272fd75f03 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -12,6 +12,8 @@ mod prompt_framing; mod prompt_project; mod queue; mod relay; +mod reminder_receipts; +mod reminders; mod scope; mod setup_mode; mod usage; @@ -2846,6 +2848,14 @@ async fn tokio_main() -> Result<()> { None }; let mut heartbeat_in_flight = false; + let mut reminders = match reminders::Reminders::open(&ctx.rest_client) { + Ok(state) => Some(state), + Err(error) => { + tracing::error!(%error, "reminder delivery disabled: cannot open durable receipts"); + None + } + }; + let (mut reminder_rx, reminder_poller) = reminders::start_polling(ctx.rest_client.clone()); let mut presence_heartbeat = if config.presence_enabled { let interval = Duration::from_secs(60); @@ -3008,6 +3018,16 @@ async fn tokio_main() -> Result<()> { } loop { + if let Some(state) = reminders.as_mut() { + state.recover_missing_turn(pool.task_map().values().map(|meta| meta.turn_id.clone())); + } + let next_reminder = match reminders.as_ref().map(|state| state.next()).transpose() { + Ok(candidate) => candidate.flatten(), + Err(error) => { + tracing::error!(%error, "cannot read reminder receipts; delivery deferred"); + None + } + }; // Whether buffered work is waiting on a lazy pool. Also gates the // retry-deadline sleep arm below: a `Failed` lifecycle keeps its // (possibly past) `retry_at` until the next wake, so sleeping on it @@ -3015,7 +3035,7 @@ async fn tokio_main() -> Result<()> { // busy spin — whenever the queued work drained after a failed wake. let mut lazy_wake_work_pending = false; if config.lazy_pool && !pool_ready { - lazy_wake_work_pending = queue.has_flushable_work(); + lazy_wake_work_pending = queue.has_flushable_work() || next_reminder.is_some(); if let Some(attempt) = pool_lifecycle .start_wake_if_due(lazy_wake_work_pending, tokio::time::Instant::now()) { @@ -3145,6 +3165,22 @@ async fn tokio_main() -> Result<()> { } } + if pool_ready && !queue.has_flushable_work() && !heartbeat_in_flight { + if let Some(reminder) = next_reminder { + if let Some(turn_id) = dispatch_private( + &mut pool, + &ctx, + &mut heartbeat_in_flight, + Some(reminder.clone()), + ) { + if let Some(state) = reminders.as_mut() { + state.started(turn_id, reminder); + } + last_activity = tokio::time::Instant::now(); + } + } + } + // Borrow result_rx and join_set simultaneously via split-borrow helper. let pool_event: Option = { let (result_rx, join_set) = pool.rx_and_join_set(); @@ -3666,6 +3702,10 @@ async fn tokio_main() -> Result<()> { } None } + Some(heads) = reminder_rx.recv() => { + if let Some(state) = reminders.as_mut() { state.refresh(heads); } + None + } _ = async { match heartbeat.as_mut() { Some(hb) => hb.tick().await, @@ -3742,6 +3782,9 @@ async fn tokio_main() -> Result<()> { match pool_event { Some(PoolEvent::Result(result)) => { + if let Some(state) = reminders.as_mut() { + state.finished(&result.turn_id, &result.outcome); + } // Stop the typing indicator for the completed turn's exact scope, // not the whole channel — a sibling thread still running in the // same channel must keep its indicator. @@ -4119,6 +4162,7 @@ async fn tokio_main() -> Result<()> { } // Cancel any in-flight presence heartbeat before sending offline. + reminder_poller.abort(); if let Some(h) = presence_task.take() { h.abort(); } @@ -4795,7 +4839,7 @@ fn handle_prompt_result( match &result.source { PromptSource::Channel(scope) => queue.mark_complete(scope.clone()), - PromptSource::Heartbeat => *heartbeat_in_flight = false, + PromptSource::Heartbeat | PromptSource::Reminder => *heartbeat_in_flight = false, } // Strip sessions for channels the agent was removed from while this @@ -5188,13 +5232,21 @@ fn dispatch_heartbeat( ctx: &Arc, heartbeat_in_flight: &mut bool, ) { + if dispatch_private(pool, ctx, heartbeat_in_flight, None).is_some() { + tracing::info!("heartbeat_fired"); + } +} + +fn dispatch_private( + pool: &mut AgentPool, + ctx: &Arc, + heartbeat_in_flight: &mut bool, + reminder: Option, +) -> Option { if *heartbeat_in_flight { - return; + return None; } - let agent = match pool.try_claim(None) { - Some(a) => a, - None => return, - }; + let agent = pool.try_claim(None)?; let prompt_text = ctx .heartbeat_prompt @@ -5207,10 +5259,17 @@ fn dispatch_heartbeat( let task_turn_id = turn_id.clone(); let abort_handle = pool.join_set.spawn(async move { + if let Some(reminder) = reminder { + reminders::run(agent, reminder, ctx_clone, result_tx, task_turn_id).await; + return; + } pool::run_prompt_task( agent, None, - Some(prompt_text), + Some(pool::PrivatePrompt { + text: prompt_text, + source: PromptSource::Heartbeat, + }), ctx_clone, result_tx, None, @@ -5225,7 +5284,7 @@ fn dispatch_heartbeat( agent_index, channel_id: None, scope: None, - turn_id, + turn_id: turn_id.clone(), recoverable_batch: None, control_tx: None, steer_tx: None, @@ -5233,7 +5292,7 @@ fn dispatch_heartbeat( }, ); *heartbeat_in_flight = true; - tracing::info!(agent = agent_index, "heartbeat_fired"); + Some(turn_id) } #[cfg(test)] diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 06383d456d3..b049df2356c 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -150,7 +150,7 @@ impl SessionState { PromptSource::Channel(scope) => { self.invalidate_scope(scope); } - PromptSource::Heartbeat => { + PromptSource::Heartbeat | PromptSource::Reminder => { self.heartbeat_session = None; self.heartbeat_turn_count = 0; self.heartbeat_standing_context_sent = false; @@ -362,10 +362,20 @@ pub struct PromptResult { /// (conversation or thread), not just the channel id, so completion and /// invalidation target the exact session. Use [`channel_id`](PromptSource::channel_id) /// where only the channel is needed. -#[derive(Debug)] +#[derive(Debug, Clone)] pub enum PromptSource { Channel(SessionScope), Heartbeat, + /// A due, author-owned NIP-ER reminder. + Reminder, +} + +/// A private prompt that does not publish channel activity. +pub struct PrivatePrompt { + /// Context delivered to the native agent. + pub text: String, + /// Whether this is a heartbeat or a due reminder. + pub source: PromptSource, } impl PromptSource { @@ -373,7 +383,7 @@ impl PromptSource { pub fn channel_id(&self) -> Option { match self { Self::Channel(scope) => Some(scope.channel_id()), - Self::Heartbeat => None, + Self::Heartbeat | Self::Reminder => None, } } @@ -386,7 +396,7 @@ impl PromptSource { pub fn scope(&self) -> Option<&SessionScope> { match self { Self::Channel(scope) => Some(scope), - Self::Heartbeat => None, + Self::Heartbeat | Self::Reminder => None, } } } @@ -2120,7 +2130,7 @@ fn send_prompt_result( pub async fn run_prompt_task( mut agent: OwnedAgent, batch: Option, - prompt_text: Option, + private_prompt: Option, ctx: Arc, result_tx: mpsc::UnboundedSender, control_rx: Option>, @@ -2149,6 +2159,7 @@ pub async fn run_prompt_task( "source": match &source { PromptSource::Channel(_) => "channel", PromptSource::Heartbeat => "heartbeat", + PromptSource::Reminder => "reminder", }, "triggeringEventIds": triggering_event_ids, }), @@ -2355,7 +2366,7 @@ pub async fn run_prompt_task( .get(scope) .cloned() .or_else(|| pending_canvas.as_ref().map(|(_, s)| s.clone())), - PromptSource::Heartbeat => None, + PromptSource::Heartbeat | PromptSource::Reminder => None, }; let (session_id, is_new_session) = match &source { @@ -2429,7 +2440,7 @@ pub async fn run_prompt_task( } } } - PromptSource::Heartbeat => { + PromptSource::Heartbeat | PromptSource::Reminder => { if let Some(sid) = &agent.state.heartbeat_session { (sid.clone(), false) } else { @@ -2526,7 +2537,9 @@ pub async fn run_prompt_task( .deliveries .get(scope) .is_some_and(|delivery| delivery.standing_context_sent), - PromptSource::Heartbeat => agent.state.heartbeat_standing_context_sent, + PromptSource::Heartbeat | PromptSource::Reminder => { + agent.state.heartbeat_standing_context_sent + } }; if is_new_session { @@ -2830,7 +2843,7 @@ pub async fn run_prompt_task( let prompt_bytes: usize = prompt_blocks.iter().map(|block| block.len()).sum(); let has_standing_context = match &source { PromptSource::Channel(_) => !standing.sections().is_empty(), - PromptSource::Heartbeat => ctx.base_prompt.is_some(), + PromptSource::Heartbeat | PromptSource::Reminder => ctx.base_prompt.is_some(), }; let standing_context_included = !agent.has_system_prompt_support() && !standing_context_sent && has_standing_context; @@ -3075,7 +3088,7 @@ pub async fn run_prompt_task( *count += 1; *count >= limit } - PromptSource::Heartbeat => { + PromptSource::Heartbeat | PromptSource::Reminder => { agent.state.heartbeat_turn_count += 1; agent.state.heartbeat_turn_count >= limit } @@ -4552,6 +4565,7 @@ fn prompt_label(source: &PromptSource) -> String { scope.telemetry_label() ), PromptSource::Heartbeat => "heartbeat".to_string(), + PromptSource::Reminder => "reminder".to_string(), } } @@ -6594,7 +6608,10 @@ done"# run_prompt_task( agent, None, - Some(format!("heartbeat-{turn}")), + Some(PrivatePrompt { + text: format!("heartbeat-{turn}"), + source: PromptSource::Heartbeat, + }), Arc::clone(&ctx), result_tx.clone(), None, diff --git a/crates/buzz-acp/src/relay.rs b/crates/buzz-acp/src/relay.rs index 23ed454fa2e..2302ec43514 100644 --- a/crates/buzz-acp/src/relay.rs +++ b/crates/buzz-acp/src/relay.rs @@ -484,6 +484,15 @@ impl RestClient { /// Accepts a slice of `nostr::Filter` (serialized as JSON array). /// Returns the events as a `serde_json::Value` (JSON array of event objects). pub async fn query(&self, filters: &[nostr::Filter]) -> Result { + self.query_json( + &serde_json::to_value(filters) + .map_err(|e| RelayError::Http(format!("filter serialize error: {e}")))?, + ) + .await + } + + /// Query bridge filters, including its composite pagination cursor. + pub async fn query_json(&self, filters: &Value) -> Result { let body_bytes = serde_json::to_vec(filters) .map_err(|e| RelayError::Http(format!("filter serialize error: {e}")))?; let resp = self.bridge_post("/query", &body_bytes).await?; diff --git a/crates/buzz-acp/src/reminder_receipts.rs b/crates/buzz-acp/src/reminder_receipts.rs new file mode 100644 index 00000000000..d6a004897bc --- /dev/null +++ b/crates/buzz-acp/src/reminder_receipts.rs @@ -0,0 +1,83 @@ +use std::fs::{self, File, OpenOptions}; +use std::io::{Read, Write}; +use std::path::{Path, PathBuf}; + +use anyhow::{Context, Result}; +use buzz_sdk::reminders::Reminder; +use sha2::{Digest, Sha256}; + +pub(super) struct Receipts { + directory: PathBuf, + _lock: File, +} + +impl Receipts { + pub(super) fn open(base: &Path, relay: &str, author: &str) -> Result { + let scope = hex::encode(Sha256::digest(format!("{relay}\n{author}"))); + let directory = base.join(scope); + fs::create_dir_all(&directory).context("create reminder receipt directory")?; + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + fs::set_permissions(&directory, fs::Permissions::from_mode(0o700))?; + } + let lock = private_file(&directory.join("lock"), false)?; + fs2::FileExt::try_lock_exclusive(&lock) + .context("another harness owns reminder delivery for this identity")?; + Ok(Self { + directory, + _lock: lock, + }) + } + + fn path(&self, reminder: &Reminder) -> PathBuf { + self.directory + .join(hex::encode(Sha256::digest(&reminder.id))) + } + + pub(super) fn contains(&self, reminder: &Reminder) -> Result { + let mut value = String::new(); + match File::open(self.path(reminder)) { + Ok(mut file) => { + file.read_to_string(&mut value)?; + } + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(false), + Err(error) => return Err(error.into()), + } + let value = value.trim(); + anyhow::ensure!( + value.len() == 64 && value.bytes().all(|c| c.is_ascii_hexdigit()), + "invalid reminder delivery receipt" + ); + Ok(value == reminder.event_id) + } + + pub(super) fn record(&self, reminder: &Reminder) -> Result<()> { + let temporary = self + .directory + .join(format!(".{}.tmp", uuid::Uuid::new_v4())); + let mut file = private_file(&temporary, true)?; + writeln!(file, "{}", reminder.event_id)?; + file.sync_all()?; + fs::rename(&temporary, self.path(reminder))?; + #[cfg(unix)] + File::open(&self.directory)?.sync_all()?; + Ok(()) + } +} + +fn private_file(path: &Path, exclusive: bool) -> Result { + let mut options = OpenOptions::new(); + options.read(true).write(true); + if exclusive { + options.create_new(true); + } else { + options.create(true); + } + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + options.mode(0o600); + } + Ok(options.open(path)?) +} diff --git a/crates/buzz-acp/src/reminders.rs b/crates/buzz-acp/src/reminders.rs new file mode 100644 index 00000000000..d4c792de24e --- /dev/null +++ b/crates/buzz-acp/src/reminders.rs @@ -0,0 +1,279 @@ +use std::path::PathBuf; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use anyhow::{Context, Result}; +use buzz_sdk::reminders::{current_heads, relay_supports_private_reminders, Reminder}; +use nostr::{Event, Timestamp}; +use serde_json::{json, Value}; +use tokio::sync::mpsc; + +use crate::pool::{ + self, OwnedAgent, PrivatePrompt, PromptContext, PromptOutcome, PromptResult, PromptSource, +}; +use crate::relay::RestClient; +use crate::reminder_receipts::Receipts; + +const POLL_INTERVAL: Duration = Duration::from_secs(30); + +pub(super) struct Reminders { + receipts: Receipts, + pending: Vec, + in_flight: Option<(String, Reminder)>, + retries: std::collections::HashMap, + delivered: std::collections::HashSet, +} + +impl Reminders { + pub(super) fn open(client: &RestClient) -> Result { + let base = std::env::var_os("BUZZ_ACP_REMINDER_STATE_DIR") + .map(PathBuf::from) + .or_else(|| { + std::env::var_os("XDG_STATE_HOME") + .map(|p| PathBuf::from(p).join("buzz-acp/reminders")) + }) + .or_else(|| { + std::env::var_os("HOME") + .map(|p| PathBuf::from(p).join(".local/state/buzz-acp/reminders")) + }) + .context("no state directory for reminder delivery receipts")?; + Ok(Self { + receipts: Receipts::open(&base, &client.base_url, &client.keys.public_key().to_hex())?, + pending: Vec::new(), + in_flight: None, + retries: Default::default(), + delivered: Default::default(), + }) + } + + pub(super) fn refresh(&mut self, heads: Vec) { + self.delivered + .retain(|id| heads.iter().any(|head| &head.event_id == id)); + self.pending = heads; + } + + pub(super) fn next(&self) -> Result> { + if self.in_flight.is_some() { + return Ok(None); + } + for reminder in &self.pending { + if reminder.is_due(Timestamp::now().as_secs()) + && self + .retries + .get(&reminder.event_id) + .is_none_or(|(_, after)| Instant::now() >= *after) + && !self.delivered.contains(&reminder.event_id) + && !self.receipts.contains(reminder)? + { + return Ok(Some(reminder.clone())); + } + } + Ok(None) + } + + pub(super) fn started(&mut self, turn_id: String, reminder: Reminder) { + tracing::info!(%turn_id, reminder_id = %reminder.id, event_id = %reminder.event_id, "reminder_dispatched"); + self.in_flight = Some((turn_id, reminder)); + } + + pub(super) fn finished(&mut self, turn_id: &str, outcome: &PromptOutcome) { + if self + .in_flight + .as_ref() + .is_none_or(|(turn, _)| turn != turn_id) + { + return; + } + let Some((_, reminder)) = self.in_flight.take() else { + return; + }; + if matches!(outcome, PromptOutcome::Ok(crate::acp::StopReason::EndTurn)) { + self.delivered.insert(reminder.event_id.clone()); + if let Err(error) = self.receipts.record(&reminder) { + tracing::error!(%error, reminder_id = %reminder.id, "reminder receipt persistence failed; restart may redeliver"); + } + self.retries.remove(&reminder.event_id); + tracing::info!(reminder_id = %reminder.id, event_id = %reminder.event_id, "reminder_delivered; disposition remains author-owned"); + } else { + self.backoff(&reminder.event_id); + tracing::warn!(reminder_id = %reminder.id, "reminder turn failed; retained for recovery"); + } + } + + fn backoff(&mut self, event_id: &str) { + let (failures, after) = self + .retries + .entry(event_id.into()) + .or_insert((0, Instant::now())); + *failures = failures.saturating_add(1); + *after = Instant::now() + Duration::from_secs(30 * 2u64.pow((*failures).min(7))); + } + + pub(super) fn recover_missing_turn(&mut self, active: impl Iterator) { + if let Some((turn, _)) = &self.in_flight { + if !active.into_iter().any(|id| &id == turn) { + if let Some((_, reminder)) = self.in_flight.take() { + self.backoff(&reminder.event_id); + } + tracing::warn!("reminder task disappeared; retained for recovery"); + } + } + } +} + +pub(super) fn start_polling( + client: RestClient, +) -> (mpsc::Receiver>, tokio::task::JoinHandle<()>) { + let (tx, rx) = mpsc::channel(1); + let task = tokio::spawn(async move { + let mut interval = tokio::time::interval(POLL_INTERVAL); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + let mut supported = false; + loop { + interval.tick().await; + if tx.is_closed() { + return; + } + if !supported { + match relay_support(&client).await { + Ok(true) => supported = true, + Ok(false) => { + tracing::info!( + "relay does not advertise NIP-ER; reminder polling disabled" + ); + return; + } + Err(error) => { + tracing::warn!(%error,"reminder capability check failed; will retry"); + continue; + } + } + } + match fetch_heads(&client, None).await { + Ok(heads) => { + if tx.send(heads).await.is_err() { + return; + } + } + Err(error) => tracing::warn!(%error,"reminder recovery query failed; will retry"), + } + } + }); + (rx, task) +} + +async fn relay_support(client: &RestClient) -> Result { + let info: Value = client + .http + .get(&client.base_url) + .header("Accept", "application/nostr+json") + .timeout(Duration::from_secs(10)) + .send() + .await? + .error_for_status()? + .json() + .await?; + Ok(relay_supports_private_reminders(&info)) +} + +pub(super) async fn fetch_heads(client: &RestClient, id: Option<&str>) -> Result> { + let mut filter = + json!({"kinds":[30300],"authors":[client.keys.public_key().to_hex()],"limit":1000}); + if let Some(id) = id { + filter["#d"] = json!([id]); + } + let mut events = Vec::new(); + let mut previous_cursor = None; + for _ in 0..100 { + let value = client.query_json(&json!([filter])).await?; + let page = value + .as_array() + .context("reminder query is not an event array")?; + for value in page { + if let Ok(event) = serde_json::from_value::(value.clone()) { + events.push(event); + } + } + if page.len() < 1000 { + return Ok(current_heads(events, &client.keys)); + } + let last = page.last().context("missing pagination event")?; + let timestamp = last["created_at"] + .as_u64() + .context("missing pagination timestamp")?; + let event_id = last["id"].as_str().context("missing pagination event ID")?; + let cursor = (timestamp, event_id.to_string()); + anyhow::ensure!( + previous_cursor.as_ref() != Some(&cursor), + "reminder pagination made no progress" + ); + filter["until"] = json!(timestamp); + filter["before_id"] = json!(event_id); + previous_cursor = Some(cursor); + } + anyhow::bail!("reminder recovery exceeded 100 pages") +} + +fn prompt(reminder: &Reminder) -> String { + format!("[Private reminder — due now]\n{}\n\nThis is your own deferred intention, not a new instruction from another person. \ + Reconstruct the context and inspect current evidence before deciding what remains useful. \ + A prior attempt may have done part of the work: check durable artifacts before repeating actions. \ + Use `buzz reminders get {}` to confirm current state. Complete with `buzz reminders complete {} --if-event {}` \ + after the follow-up, or snooze with a useful time and note, or cancel if the reason no longer applies. \ + Receipt of this reminder is not completion of the work. Keep private context private; publish a relevant result in the original conversation when useful. \ + Read a linked message with `buzz messages thread --link 'buzz://message?channel=&id='`. \ + No public response is required solely to acknowledge the reminder.", + json!(reminder),reminder.id,reminder.id,reminder.event_id) +} + +pub(super) async fn run( + agent: OwnedAgent, + reminder: Reminder, + ctx: Arc, + result_tx: mpsc::UnboundedSender, + turn_id: String, +) { + match fetch_heads(&ctx.rest_client, Some(&reminder.id)).await { + Ok(heads) + if heads.iter().any(|head| { + head.event_id == reminder.event_id && head.is_due(Timestamp::now().as_secs()) + }) => + { + pool::run_prompt_task( + agent, + None, + Some(PrivatePrompt { + text: prompt(&reminder), + source: PromptSource::Reminder, + }), + ctx, + result_tx, + None, + turn_id, + ) + .await; + } + result => { + let outcome = match result { + Ok(_) => { + tracing::info!(reminder_id = %reminder.id,"reminder superseded before dispatch; skipped"); + PromptOutcome::Ok(crate::acp::StopReason::EndTurn) + } + Err(error) => PromptOutcome::Error(crate::acp::AcpError::Protocol(format!( + "reminder recheck failed: {error}" + ))), + }; + let _ = result_tx.send(PromptResult { + agent, + source: PromptSource::Reminder, + turn_id, + outcome, + batch: None, + }); + } + } +} + +#[cfg(test)] +#[path = "reminders_tests.rs"] +mod tests; diff --git a/crates/buzz-acp/src/reminders_tests.rs b/crates/buzz-acp/src/reminders_tests.rs new file mode 100644 index 00000000000..9e9828e8c92 --- /dev/null +++ b/crates/buzz-acp/src/reminders_tests.rs @@ -0,0 +1,200 @@ +use super::*; +use crate::acp::{AcpError, StopReason}; +use buzz_sdk::reminders::{build, Content, Status}; +use nostr::Keys; +use std::collections::{HashMap, HashSet}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; + +fn event(keys: &Keys, id: &str, due: u64, created: u64) -> Event { + build( + keys, + id, + &Content { + status: Status::Pending, + note: Some("Inspect the durable experiment results".into()), + target: None, + extra: Default::default(), + }, + Some(due), + created, + ) + .unwrap() + .sign_with_keys(keys) + .unwrap() +} + +fn state(directory: &std::path::Path, keys: &Keys) -> Reminders { + Reminders { + receipts: Receipts::open(directory, "https://relay.test", &keys.public_key().to_hex()) + .unwrap(), + pending: vec![], + in_flight: None, + retries: HashMap::new(), + delivered: HashSet::new(), + } +} + +#[test] +fn old_created_due_reminder_recovers_after_restart_and_delivered_version_does_not() { + let directory = tempfile::tempdir().unwrap(); + let keys = Keys::generate(); + let reminder = Reminder::decrypt(&event(&keys, "work", 2, 1), &keys).unwrap(); + let mut first = state(directory.path(), &keys); + first.refresh(vec![reminder.clone()]); + assert!(first.next().unwrap().is_some()); + first.started("interrupted-turn".into(), reminder.clone()); + drop(first); + + let mut restarted = state(directory.path(), &keys); + restarted.refresh(vec![reminder.clone()]); + assert!(restarted.next().unwrap().is_some()); + restarted.started("successful-turn".into(), reminder.clone()); + restarted.finished("successful-turn", &PromptOutcome::Ok(StopReason::EndTurn)); + assert!(restarted.next().unwrap().is_none()); + assert_eq!(restarted.pending[0].content.status, Status::Pending); + drop(restarted); + + let mut again = state(directory.path(), &keys); + again.refresh(vec![reminder.clone()]); + assert!(again.next().unwrap().is_none()); + let snoozed = Reminder::decrypt(&event(&keys, "work", 3, 2), &keys).unwrap(); + again.refresh(vec![snoozed]); + assert!(again.next().unwrap().is_some()); +} + +#[test] +fn failed_cancelled_and_limited_turns_back_off_without_starving_other_work() { + let directory = tempfile::tempdir().unwrap(); + let keys = Keys::generate(); + let reminder = Reminder::decrypt(&event(&keys, "work", 2, 1), &keys).unwrap(); + let other = Reminder::decrypt(&event(&keys, "other", 2, 1), &keys).unwrap(); + let mut state = state(directory.path(), &keys); + state.refresh(vec![reminder.clone(), other.clone()]); + for outcome in [ + PromptOutcome::Error(AcpError::Protocol("offline".into())), + PromptOutcome::Cancelled, + PromptOutcome::Ok(StopReason::MaxTokens), + PromptOutcome::Ok(StopReason::Cancelled), + ] { + state.started("turn".into(), reminder.clone()); + state.finished("turn", &outcome); + assert!(!state.receipts.contains(&reminder).unwrap()); + assert_eq!(state.next().unwrap().unwrap().id, other.id); + } + state.started("panicked".into(), other.clone()); + state.recover_missing_turn(std::iter::empty()); + assert!(state.in_flight.is_none()); + assert!(!state.receipts.contains(&other).unwrap()); +} + +#[test] +fn future_snoozed_and_cancelled_heads_supersede_queued_intent() { + let directory = tempfile::tempdir().unwrap(); + let keys = Keys::generate(); + let mut state = state(directory.path(), &keys); + let old = event(&keys, "id", 2, 1); + state.refresh(current_heads(vec![old.clone()], &keys)); + assert!(state.next().unwrap().is_some()); + let future = event(&keys, "id", Timestamp::now().as_secs() + 86_400, 2); + state.refresh(current_heads(vec![old, future.clone()], &keys)); + assert!(state.next().unwrap().is_none()); + let mut content = Reminder::decrypt(&future, &keys).unwrap().content; + content.status = Status::Cancelled; + let cancelled = build(&keys, "id", &content, None, 3) + .unwrap() + .sign_with_keys(&keys) + .unwrap(); + state.refresh(current_heads(vec![future, cancelled], &keys)); + assert!(state.next().unwrap().is_none()); +} + +#[test] +fn receipt_lock_and_scope_prevent_local_competing_consumers() { + let directory = tempfile::tempdir().unwrap(); + let first = Receipts::open(directory.path(), "https://one.test", "author").unwrap(); + assert!(Receipts::open(directory.path(), "https://one.test", "author").is_err()); + assert!(Receipts::open(directory.path(), "https://two.test", "author").is_ok()); + assert!(Receipts::open(directory.path(), "https://one.test", "other").is_ok()); + drop(first); + assert!(Receipts::open(directory.path(), "https://one.test", "author").is_ok()); +} + +async fn mock_relay( + keys: Keys, + responses: Vec, +) -> (RestClient, tokio::task::JoinHandle>) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let task = tokio::spawn(async move { + let mut requests = Vec::new(); + for response in responses { + let (mut socket, _) = listener.accept().await.unwrap(); + let mut bytes = Vec::new(); + let (end, length) = loop { + let mut buf = [0u8; 4096]; + let count = socket.read(&mut buf).await.unwrap(); + assert!(count > 0); + bytes.extend_from_slice(&buf[..count]); + if let Some(end) = bytes.windows(4).position(|part| part == b"\r\n\r\n") { + let header = String::from_utf8_lossy(&bytes[..end]); + assert!(header.to_lowercase().contains("authorization: nostr ")); + let length: usize = header + .lines() + .find_map(|line| { + line.to_lowercase() + .strip_prefix("content-length:") + .map(|v| v.trim().parse().unwrap()) + }) + .unwrap(); + break (end + 4, length); + } + }; + while bytes.len() < end + length { + let mut buf = [0u8; 4096]; + let count = socket.read(&mut buf).await.unwrap(); + assert!(count > 0); + bytes.extend_from_slice(&buf[..count]); + } + requests.push(serde_json::from_slice(&bytes[end..end + length]).unwrap()); + let body = response.to_string(); + socket.write_all(format!("HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",body.len()).as_bytes()).await.unwrap(); + } + requests + }); + ( + RestClient { + http: reqwest::Client::new(), + base_url: format!("http://{address}"), + keys, + auth_tag_json: None, + }, + task, + ) +} + +#[tokio::test] +async fn authenticated_recovery_pages_old_created_heads_and_ignores_bad_items() { + let keys = Keys::generate(); + let recent = event(&keys, "future", Timestamp::now().as_secs() + 86_400, 100); + let old = event(&keys, "due", 2, 1); + let first = vec![json!(recent); 1000]; + let (client, server) = mock_relay( + keys.clone(), + vec![json!(first), json!([old,{"malformed":true}])], + ) + .await; + let heads = fetch_heads(&client, None).await.unwrap(); + assert_eq!(heads.len(), 2); + assert!(heads + .iter() + .any(|r| r.id == "due" && r.is_due(Timestamp::now().as_secs()))); + let requests = server.await.unwrap(); + assert!(requests[0][0].get("since").is_none()); + assert!(requests[0][0].get("until").is_none()); + assert_eq!( + requests[0][0]["authors"], + json!([keys.public_key().to_hex()]) + ); + assert_eq!(requests[1][0]["until"], 100); + assert_eq!(requests[1][0]["before_id"], recent.id.to_hex()); +} diff --git a/crates/buzz-cli/src/commands/mod.rs b/crates/buzz-cli/src/commands/mod.rs index 7ed03f9d060..3c99d9da64d 100644 --- a/crates/buzz-cli/src/commands/mod.rs +++ b/crates/buzz-cli/src/commands/mod.rs @@ -16,6 +16,7 @@ pub mod pr; pub mod project_channel; pub mod projects; pub mod reactions; +pub mod reminders; pub mod repos; pub mod social; pub mod upload; diff --git a/crates/buzz-cli/src/commands/reminders.rs b/crates/buzz-cli/src/commands/reminders.rs new file mode 100644 index 00000000000..6571df7ea3c --- /dev/null +++ b/crates/buzz-cli/src/commands/reminders.rs @@ -0,0 +1,304 @@ +//! Manage the caller's encrypted NIP-ER reminders. + +use buzz_sdk::reminders::{self, Content, Reminder, Status}; +use clap::{Args, Subcommand}; +use nostr::{Event, Timestamp}; +use serde_json::{json, Value}; + +use crate::{client::BuzzClient, error::CliError, links::parse_message_link}; + +#[derive(Subcommand)] +pub enum RemindersCmd { + /// Schedule a private follow-up. Preserve why, what to inspect, and context. + Create { + #[command(flatten)] + time: DueTime, + /// Reason and context for the follow-up; use '-' to read stdin. + #[arg(long)] + note: String, + /// Optional original Buzz message link. + #[arg(long)] + link: Option, + }, + /// List current reminder heads, decrypted for your identity. + List { + /// Show terminal states as well as pending reminders. + #[arg(long)] + all: bool, + /// Show only reminders that are due now. + #[arg(long, conflicts_with = "all")] + due: bool, + }, + /// Inspect a reminder's current version before acting. + Get { id: String }, + /// Reconsider later; optionally replace the reason or context. + Snooze { + id: String, + #[command(flatten)] + time: DueTime, + #[arg(long)] + note: Option, + /// Refuse the update if another client changed this event version. + #[arg(long)] + if_event: Option, + }, + /// Record that you have completed or acknowledged this follow-up. + Complete { + id: String, + #[arg(long)] + if_event: Option, + }, + /// Cancel a follow-up that is no longer useful. + Cancel { + id: String, + #[arg(long)] + if_event: Option, + }, +} + +#[derive(Args)] +pub struct DueTime { + /// Absolute RFC3339 time with a timezone (e.g. 2026-09-14T14:00:00Z). + #[arg(long, required_unless_present = "after", conflicts_with = "after")] + at: Option, + /// Delay such as 30s, 20m, 4h, or 7d. + #[arg(long, required_unless_present = "at", conflicts_with = "at")] + after: Option, +} + +impl DueTime { + fn resolve(&self, now: u64) -> Result { + let time = if let Some(at) = &self.at { + let timestamp = chrono::DateTime::parse_from_rfc3339(at) + .map_err(|_| { + CliError::Usage("--at needs an RFC3339 timestamp with timezone".into()) + })? + .timestamp(); + u64::try_from(timestamp) + .map_err(|_| CliError::Usage("--at predates Unix time".into()))? + } else { + let raw = self.after.as_deref().unwrap_or(""); + let split = raw + .len() + .checked_sub(1) + .filter(|&i| raw.is_char_boundary(i)) + .ok_or_else(|| { + CliError::Usage("--after needs a duration such as 20m or 7d".into()) + })?; + let (number, unit) = raw.split_at(split); + let multiplier = match unit { + "s" => 1, + "m" => 60, + "h" => 3600, + "d" => 86_400, + _ => return Err(CliError::Usage("--after units are s, m, h, d".into())), + }; + let seconds = reminders::parse_not_before(number) + .map_err(input_error)? + .checked_mul(multiplier) + .ok_or_else(|| CliError::Usage("duration overflow".into()))?; + now.checked_add(seconds) + .ok_or_else(|| CliError::Usage("due time overflow".into()))? + }; + if time <= now { + return Err(CliError::Usage("choose a future reminder time".into())); + } + reminders::parse_not_before(&time.to_string()).map_err(input_error) + } +} + +fn input_error(error: buzz_sdk::SdkError) -> CliError { + CliError::Usage(error.to_string()) +} + +fn read_note(note: String) -> Result { + if note != "-" { + return Ok(note); + } + std::io::read_to_string(std::io::stdin()) + .map_err(|error| CliError::Other(format!("cannot read reminder note: {error}"))) +} + +async fn fetch(client: &BuzzClient, id: Option<&str>) -> Result, CliError> { + let mut filter = json!({"kinds": [30300], "authors": [client.keys().public_key().to_hex()]}); + if let Some(id) = id { + filter["#d"] = json!([id]); + } + let values = client.query_all(filter).await?; + let events = values + .into_iter() + .filter_map(|v| serde_json::from_value::(v).ok()) + .collect(); + Ok(reminders::current_heads(events, client.keys())) +} + +async fn head(client: &BuzzClient, id: &str) -> Result { + fetch(client, Some(id)) + .await? + .into_iter() + .next() + .ok_or_else(|| CliError::NotFound("no valid reminder for this identity and ID".into())) +} + +async fn target(client: &BuzzClient, link: &str) -> Result { + let link = parse_message_link(link)?; + let raw = client + .query(&json!({"ids": [link.message_id], "kinds": [9, 11, 1111, 40002, 40007]})) + .await?; + let events: Vec = serde_json::from_str(&raw) + .map_err(|_| CliError::Other("invalid message response".into()))?; + let event = events + .into_iter() + .find(|event| event.id.to_hex() == link.message_id) + .ok_or_else(|| CliError::NotFound("reminder target message not found".into()))?; + event + .verify() + .map_err(|_| CliError::Other("invalid target message signature".into()))?; + if !event + .tags + .iter() + .any(|tag| tag.kind().as_str() == "h" && tag.content() == Some(&link.channel_id)) + { + return Err(CliError::Usage( + "target message does not belong to the linked channel".into(), + )); + } + Ok( + json!({"id": event.id.to_hex(), "eventId": event.id.to_hex(), "channelId": link.channel_id, + "preview": event.content.chars().take(500).collect::(), "authorPubkey": event.pubkey.to_hex()}), + ) +} + +async fn publish( + client: &BuzzClient, + id: &str, + content: &Content, + due: Option, + created_at: u64, +) -> Result<(), CliError> { + let builder = + reminders::build(client.keys(), id, content, due, created_at).map_err(input_error)?; + let event = client.sign_event(builder)?; + let event_id = event.id.to_hex(); + let raw = client.submit_event(event).await?; + super::parse_write_response(&raw, "reminder was superseded; inspect its current head")?; + let current = head(client, id).await?; + if current.event_id != event_id { + return Err(CliError::Conflict( + "reminder changed concurrently; inspect its current head".into(), + )); + } + println!( + "{}", + json!({"accepted":true,"event_id":event_id,"reminder_id":id,"not_before":due,"status":content.status}) + ); + Ok(()) +} + +async fn update( + client: &BuzzClient, + id: String, + status: Status, + time: Option, + note: Option, + expected: Option, +) -> Result<(), CliError> { + let mut reminder = head(client, &id).await?; + if expected + .as_deref() + .is_some_and(|event| event != reminder.event_id) + { + return Err(CliError::Conflict( + "reminder changed; inspect its current head".into(), + )); + } + if reminder.content.status != Status::Pending { + return Err(CliError::Usage( + "reminder is already terminal; create a new reminder if needed".into(), + )); + } + let now = Timestamp::now().as_secs(); + if status == Status::Done && reminder.not_before.is_some_and(|time| time > now) { + return Err(CliError::Usage( + "reminder is not due yet; cancel it if the reason no longer applies".into(), + )); + } + let due = time.as_ref().map(|time| time.resolve(now)).transpose()?; + reminder.content.status = status; + if let Some(note) = note { + reminder.content.note = Some(read_note(note)?); + } + publish( + client, + &id, + &reminder.content, + due, + now.max(reminder.created_at.saturating_add(1)), + ) + .await +} + +pub async fn dispatch(command: RemindersCmd, client: &BuzzClient) -> Result<(), CliError> { + if matches!(command, RemindersCmd::Create { .. }) { + let info: Value = reqwest::Client::new() + .get(client.relay_url()) + .header("Accept", "application/nostr+json") + .timeout(std::time::Duration::from_secs(10)) + .send() + .await? + .error_for_status()? + .json() + .await?; + if !reminders::relay_supports_private_reminders(&info) { + return Err(CliError::Usage( + "relay does not advertise author-private NIP-ER and NIP-42 support".into(), + )); + } + } + match command { + RemindersCmd::Create { time, note, link } => { + let now = Timestamp::now().as_secs(); + let due = time.resolve(now)?; + let content = Content { + status: Status::Pending, + note: Some(read_note(note)?), + target: match link { + Some(link) => Some(target(client, &link).await?), + None => None, + }, + extra: Default::default(), + }; + publish(client, &reminders::new_id(), &content, Some(due), now).await + } + RemindersCmd::List { all, due } => { + let now = Timestamp::now().as_secs(); + let reminders: Vec<_> = fetch(client, None) + .await? + .into_iter() + .filter(|r| (all || r.content.status == Status::Pending) && (!due || r.is_due(now))) + .collect(); + println!("{}", json!(reminders)); + Ok(()) + } + RemindersCmd::Get { id } => { + println!("{}", json!(head(client, &id).await?)); + Ok(()) + } + RemindersCmd::Snooze { + id, + time, + note, + if_event, + } => update(client, id, Status::Pending, Some(time), note, if_event).await, + RemindersCmd::Complete { id, if_event } => { + update(client, id, Status::Done, None, None, if_event).await + } + RemindersCmd::Cancel { id, if_event } => { + update(client, id, Status::Cancelled, None, None, if_event).await + } + } +} + +#[cfg(test)] +#[path = "reminders_tests.rs"] +mod tests; diff --git a/crates/buzz-cli/src/commands/reminders_tests.rs b/crates/buzz-cli/src/commands/reminders_tests.rs new file mode 100644 index 00000000000..c3e6eb2fb0a --- /dev/null +++ b/crates/buzz-cli/src/commands/reminders_tests.rs @@ -0,0 +1,91 @@ +use super::*; +use clap::Parser; + +#[derive(Parser)] +struct TestCli { + #[command(subcommand)] + command: RemindersCmd, +} + +#[test] +fn create_requires_one_unambiguous_due_time() { + assert!(TestCli::try_parse_from(["buzz", "create", "--note", "inspect results"]).is_err()); + assert!(TestCli::try_parse_from([ + "buzz", + "create", + "--note", + "inspect results", + "--after", + "7d", + "--at", + "2026-09-14T14:00:00Z" + ]) + .is_err()); + assert!(TestCli::try_parse_from([ + "buzz", + "create", + "--note", + "inspect results", + "--after", + "7d" + ]) + .is_ok()); +} + +#[test] +fn due_time_handles_units_timezone_and_rejects_past_or_overflow() { + for (raw, expected) in [("30s", 130), ("20m", 1300), ("4h", 14500), ("7d", 604900)] { + assert_eq!( + DueTime { + at: None, + after: Some(raw.into()) + } + .resolve(100) + .unwrap(), + expected + ); + } + for raw in [ + "0s", + "-1h", + "3weeks", + "1.5h", + "18446744073709551615d", + "é", + " 5m", + ] { + assert!(DueTime { + at: None, + after: Some(raw.into()) + } + .resolve(100) + .is_err()); + } + let first = DueTime { + at: Some("2026-09-14T14:00:00Z".into()), + after: None, + } + .resolve(100) + .unwrap(); + assert_eq!( + first, + DueTime { + at: Some("2026-09-14T10:00:00-04:00".into()), + after: None + } + .resolve(100) + .unwrap() + ); + assert!(DueTime { + at: Some("2026-09-14T14:00:00".into()), + after: None + } + .resolve(100) + .is_err()); + assert!(DueTime { + at: Some("1970-01-01T00:00:00Z".into()), + after: None + } + .resolve(100) + .is_err()); +} diff --git a/crates/buzz-cli/src/lib.rs b/crates/buzz-cli/src/lib.rs index 3f2bea73979..47bb74a6514 100644 --- a/crates/buzz-cli/src/lib.rs +++ b/crates/buzz-cli/src/lib.rs @@ -180,6 +180,9 @@ enum Cmd { /// Send, read, search, and manage messages #[command(subcommand)] Messages(MessagesCmd), + /// Schedule and manage your private follow-ups + #[command(subcommand)] + Reminders(commands::reminders::RemindersCmd), /// Create, configure, and manage channels #[command(subcommand)] Channels(ChannelsCmd), @@ -2104,6 +2107,7 @@ async fn run(cli: Cli) -> Result<(), CliError> { match cli.command { Cmd::Agents(sub) => commands::agents::dispatch(sub, &client).await, Cmd::Messages(sub) => commands::messages::dispatch(sub, &client, &cli.format).await, + Cmd::Reminders(sub) => commands::reminders::dispatch(sub, &client).await, Cmd::Channels(sub) => commands::channels::dispatch(sub, &client, &cli.format).await, Cmd::Canvas(sub) => commands::channels::dispatch_canvas(sub, &client).await, Cmd::Reactions(sub) => commands::reactions::dispatch(sub, &client).await, @@ -2270,6 +2274,7 @@ mod tests { "pr", "projects", "reactions", + "reminders", "repos", "social", "upload", diff --git a/crates/buzz-sdk/src/lib.rs b/crates/buzz-sdk/src/lib.rs index 845505c56d5..9a71f7d9df9 100644 --- a/crates/buzz-sdk/src/lib.rs +++ b/crates/buzz-sdk/src/lib.rs @@ -16,6 +16,7 @@ pub mod broker; pub mod builders; pub mod mentions; pub mod nip_oa; +pub mod reminders; pub use builders::*; diff --git a/crates/buzz-sdk/src/reminders.rs b/crates/buzz-sdk/src/reminders.rs new file mode 100644 index 00000000000..487b5d87612 --- /dev/null +++ b/crates/buzz-sdk/src/reminders.rs @@ -0,0 +1,289 @@ +//! Encrypted, author-owned event reminders (NIP-ER). + +use std::collections::HashMap; + +use nostr::{nips::nip44, Event, EventBuilder, Keys, Kind, Tag, Timestamp}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +use crate::{kind::KIND_EVENT_REMINDER, SdkError}; + +#[path = "reminders_json.rs"] +mod strict_json; + +/// An author's disposition of a reminder. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum Status { + /// Awaiting the author's follow-up. + Pending, + /// Acknowledged or completed by its author. + Done, + /// No longer wanted by its author. + Cancelled, +} + +/// Private reminder content. Unknown fields survive updates. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Content { + /// Current author-owned disposition. + pub status: Status, + /// Optional NIP-ER target; also accepts Buzz Desktop's eventId/channelId shape. + #[serde(skip_serializing_if = "Option::is_none")] + pub target: Option, + /// Context and reason for returning to this work. + #[serde(skip_serializing_if = "Option::is_none")] + pub note: Option, + /// Forward-compatible private fields. + #[serde(flatten)] + pub extra: HashMap, +} + +/// A validated reminder head, decrypted for its owner. +#[derive(Debug, Clone, Serialize)] +pub struct Reminder { + /// The opaque address identifier (`d`), stable across replacements. + pub id: String, + /// Signed event ID for this particular version. + pub event_id: String, + /// Version timestamp, not the due time. + pub created_at: u64, + /// Earliest permissible delivery time, in Unix seconds. + pub not_before: Option, + /// Optional NIP-40 expiration; expired reminders must not wake their owner. + #[serde(skip_serializing_if = "Option::is_none")] + pub expiration: Option, + /// Decrypted private content. + #[serde(flatten)] + pub content: Content, +} + +fn invalid(message: &str) -> SdkError { + SdkError::InvalidInput(message.into()) +} + +/// Whether a NIP-11 document advertises the private reminder read contract. +pub fn relay_supports_private_reminders(info: &Value) -> bool { + info.get("supported_extensions") + .and_then(Value::as_array) + .is_some_and(|items| items.iter().any(|item| item == "nip-er")) + && info + .get("supported_nips") + .and_then(Value::as_array) + .is_some_and(|items| items.iter().any(|item| item.as_u64() == Some(42))) +} + +fn hex_id(value: &str) -> bool { + value.len() == 64 + && value + .bytes() + .all(|c| c.is_ascii_digit() || (b'a'..=b'f').contains(&c)) +} + +fn validate_content(content: &Content) -> Result<(), SdkError> { + let mut has_reference = false; + if let Some(target) = &content.target { + let target = target + .as_object() + .ok_or_else(|| invalid("reminder target must be an object"))?; + for field in ["id", "eventId"] { + if let Some(value) = target.get(field) { + if !value.as_str().is_some_and(hex_id) { + return Err(invalid("invalid reminder target event ID")); + } + has_reference = true; + } + } + if let (Some(id), Some(legacy)) = (target.get("id"), target.get("eventId")) { + if id != legacy { + return Err(invalid("conflicting reminder target IDs")); + } + } + if let Some(address) = target.get("a") { + let address = address + .as_str() + .ok_or_else(|| invalid("invalid reminder target address"))?; + nostr::nips::nip01::Coordinate::parse(address) + .map_err(|_| invalid("invalid reminder target address"))?; + has_reference = true; + } + for field in ["preview", "channelId", "authorPubkey"] { + if target.get(field).is_some_and(|value| !value.is_string()) { + return Err(invalid("invalid reminder target text")); + } + } + if target.get("relays").is_some_and(|value| !value.is_array()) { + return Err(invalid("reminder relays must be an array")); + } + } + if content.status == Status::Pending + && !has_reference + && content.note.as_deref().is_none_or(|n| n.trim().is_empty()) + { + return Err(invalid( + "a pending reminder needs a target or a non-empty note", + )); + } + Ok(()) +} + +fn single_tag<'a>(event: &'a Event, name: &str) -> Result, SdkError> { + let mut tags = event.tags.iter().filter(|tag| tag.kind().as_str() == name); + let value = tags.next(); + if tags.next().is_some() { + return Err(invalid("duplicate reminder tag")); + } + value + .map(|tag| tag.content().ok_or_else(|| invalid("empty reminder tag"))) + .transpose() +} + +/// Parse the exact integer format required for NIP-ER due times. +pub fn parse_not_before(raw: &str) -> Result { + if raw.is_empty() + || !raw.bytes().all(|c| c.is_ascii_digit()) + || (raw.len() > 1 && raw.starts_with('0')) + { + return Err(invalid("malformed not_before")); + } + raw.parse::() + .ok() + .filter(|value| *value <= 9_007_199_254_740_991) + .ok_or_else(|| invalid("not_before out of range")) +} + +impl Reminder { + /// Verify the signature and owner before decrypting a reminder. + pub fn decrypt(event: &Event, keys: &Keys) -> Result { + if event.kind.as_u16() != KIND_EVENT_REMINDER as u16 || event.pubkey != keys.public_key() { + return Err(invalid("reminder kind or owner mismatch")); + } + event + .verify() + .map_err(|_| invalid("invalid reminder signature"))?; + let id = single_tag(event, "d")? + .filter(|value| !value.is_empty()) + .ok_or_else(|| invalid("reminder needs one non-empty d tag"))?; + let not_before = single_tag(event, "not_before")? + .map(parse_not_before) + .transpose()?; + let plaintext = nip44::decrypt(keys.secret_key(), &keys.public_key(), &event.content) + .map_err(|_| invalid("cannot decrypt reminder"))?; + let value = strict_json::parse(&plaintext).map_err(|_| invalid("invalid reminder JSON"))?; + if value.get("note").is_some_and(|note| !note.is_string()) + || value + .get("target") + .is_some_and(|target| !target.is_object()) + { + return Err(invalid("invalid reminder note or target")); + } + let content: Content = + serde_json::from_value(value).map_err(|_| invalid("invalid reminder content"))?; + validate_content(&content)?; + Ok(Self { + id: id.into(), + event_id: event.id.to_hex(), + created_at: event.created_at.as_secs(), + not_before, + expiration: single_tag(event, "expiration")? + .map(parse_not_before) + .transpose()?, + content, + }) + } + + /// Whether this pending head may be delivered at `now`. + pub fn is_due(&self, now: u64) -> bool { + self.content.status == Status::Pending + && self.not_before.is_some_and(|time| time <= now) + && self.expiration.is_none_or(|time| time > now) + } +} + +/// Select signed current heads before decrypting, so an invalid newer version +/// cannot resurrect an older pending reminder. +pub fn current_heads(events: Vec, keys: &Keys) -> Vec { + let mut heads: HashMap = HashMap::new(); + for event in events { + if event.pubkey != keys.public_key() + || event.kind.as_u16() != KIND_EVENT_REMINDER as u16 + || event.verify().is_err() + { + continue; + } + let Ok(Some(id)) = single_tag(&event, "d") else { + continue; + }; + let replace = heads.get(id).is_none_or(|previous| { + event.created_at > previous.created_at + || (event.created_at == previous.created_at && event.id < previous.id) + }); + if replace { + heads.insert(id.into(), event); + } + } + let mut reminders: Vec<_> = heads + .values() + .filter_map(|event| Reminder::decrypt(event, keys).ok()) + .collect(); + reminders.sort_by_key(|reminder| (reminder.not_before, reminder.id.clone())); + reminders +} + +/// Generate an opaque address with more than 128 bits of random entropy. +pub fn new_id() -> String { + format!( + "{}{}", + uuid::Uuid::new_v4().simple(), + uuid::Uuid::new_v4().simple() + ) +} + +/// Build a self-encrypted reminder; callers add community/auth tags and sign. +/// Terminal updates omit `not_before` and expire after 30–90 days. +pub fn build( + keys: &Keys, + id: &str, + content: &Content, + not_before: Option, + created_at: u64, +) -> Result { + validate_content(content)?; + if id.is_empty() { + return Err(invalid("empty reminder ID")); + } + if content.status != Status::Pending && not_before.is_some() { + return Err(invalid("terminal reminders must omit not_before")); + } + let plaintext = + serde_json::to_string(content).map_err(|_| invalid("cannot serialize reminder"))?; + let encrypted = nip44::encrypt( + keys.secret_key(), + &keys.public_key(), + plaintext, + nip44::Version::V2, + ) + .map_err(|_| invalid("cannot encrypt reminder"))?; + let mut tags = vec![Tag::identifier(id), Tag::alt("Encrypted reminder")]; + if let Some(time) = not_before { + parse_not_before(&time.to_string())?; + tags.push( + Tag::parse(["not_before", &time.to_string()]) + .map_err(|error| SdkError::InvalidTag(error.to_string()))?, + ); + } else if content.status != Status::Pending { + let jitter = uuid::Uuid::new_v4().as_u128() % (60 * 86_400); + tags.push(Tag::expiration(Timestamp::from( + created_at.saturating_add(30 * 86_400 + jitter as u64), + ))); + } + Ok( + EventBuilder::new(Kind::Custom(KIND_EVENT_REMINDER as u16), encrypted) + .tags(tags) + .custom_created_at(Timestamp::from(created_at)), + ) +} + +#[cfg(test)] +#[path = "reminders_tests.rs"] +mod tests; diff --git a/crates/buzz-sdk/src/reminders_json.rs b/crates/buzz-sdk/src/reminders_json.rs new file mode 100644 index 00000000000..9aa29934c41 --- /dev/null +++ b/crates/buzz-sdk/src/reminders_json.rs @@ -0,0 +1,82 @@ +pub(super) fn parse(text: &str) -> Result { + use serde::de::{DeserializeSeed, Deserializer, MapAccess, SeqAccess, Visitor}; + use serde_json::Value; + use std::collections::HashSet; + use std::fmt; + + struct StrictValue; + + impl<'de> DeserializeSeed<'de> for StrictValue { + type Value = Value; + fn deserialize>(self, d: D) -> Result { + d.deserialize_any(StrictValue) + } + } + + impl<'de> Visitor<'de> for StrictValue { + type Value = Value; + + fn expecting(&self, f: &mut fmt::Formatter) -> fmt::Result { + f.write_str("any valid JSON value (objects must have unique keys)") + } + + fn visit_bool(self, v: bool) -> Result { + Ok(Value::Bool(v)) + } + fn visit_i64(self, v: i64) -> Result { + Ok(Value::Number(v.into())) + } + fn visit_u64(self, v: u64) -> Result { + Ok(Value::Number(v.into())) + } + fn visit_f64(self, v: f64) -> Result { + serde_json::Number::from_f64(v) + .map(Value::Number) + .ok_or_else(|| E::custom("non-finite float")) + } + fn visit_str(self, v: &str) -> Result { + Ok(Value::String(v.to_owned())) + } + fn visit_string(self, v: String) -> Result { + Ok(Value::String(v)) + } + fn visit_unit(self) -> Result { + Ok(Value::Null) + } + fn visit_none(self) -> Result { + Ok(Value::Null) + } + fn visit_some>(self, d: D) -> Result { + d.deserialize_any(StrictValue) + } + + fn visit_seq>(self, mut seq: A) -> Result { + let mut out = Vec::with_capacity(seq.size_hint().unwrap_or(0)); + while let Some(v) = seq.next_element_seed(StrictValue)? { + out.push(v); + } + Ok(Value::Array(out)) + } + + fn visit_map>(self, mut map: A) -> Result { + use serde::de::Error; + let mut seen: HashSet = HashSet::new(); + let mut out = serde_json::Map::new(); + while let Some(k) = map.next_key::()? { + if !seen.insert(k.clone()) { + return Err(A::Error::custom(format!( + "duplicate object member name: {k}" + ))); + } + let v = map.next_value_seed(StrictValue)?; + out.insert(k, v); + } + Ok(Value::Object(out)) + } + } + + let mut de = serde_json::Deserializer::from_str(text); + let value = StrictValue.deserialize(&mut de)?; + de.end()?; + Ok(value) +} diff --git a/crates/buzz-sdk/src/reminders_tests.rs b/crates/buzz-sdk/src/reminders_tests.rs new file mode 100644 index 00000000000..4028b554602 --- /dev/null +++ b/crates/buzz-sdk/src/reminders_tests.rs @@ -0,0 +1,190 @@ +use super::*; +use serde_json::json; + +fn pending() -> Content { + Content { + status: Status::Pending, + note: Some("Inspect experiment X; decide continue/revise/stop".into()), + target: None, + extra: Default::default(), + } +} + +fn event(keys: &Keys, id: &str, content: &Content, due: Option, created: u64) -> Event { + build(keys, id, content, due, created) + .unwrap() + .sign_with_keys(keys) + .unwrap() +} + +#[test] +fn encrypted_lifecycle_roundtrip_and_time_boundary() { + let keys = Keys::generate(); + let initial = event(&keys, "opaque-id", &pending(), Some(200), 100); + assert!(!initial.content.contains("Inspect")); + assert!(initial + .tags + .iter() + .all(|tag| !matches!(tag.kind().as_str(), "h" | "e" | "p"))); + let reminder = Reminder::decrypt(&initial, &keys).unwrap(); + assert!(!reminder.is_due(199)); + assert!(reminder.is_due(200)); + assert!(Reminder::decrypt(&initial, &Keys::generate()).is_err()); + + let snoozed = event(&keys, "opaque-id", &pending(), Some(400), 201); + let heads = current_heads(vec![initial.clone(), snoozed.clone(), initial], &keys); + assert_eq!(heads.len(), 1); + assert!(!heads[0].is_due(200)); + let mut completed = pending(); + completed.status = Status::Done; + let done = event(&keys, "opaque-id", &completed, None, 401); + assert!(done.tags.expiration().is_some()); + let heads = current_heads(vec![done, snoozed], &keys); + assert_eq!(heads[0].content.status, Status::Done); + assert!(!heads[0].is_due(u64::MAX)); +} + +#[test] +fn malformed_new_head_never_resurrects_previous_intent() { + let keys = Keys::generate(); + let previous = event(&keys, "id", &pending(), Some(2), 1); + let bad = EventBuilder::new(Kind::Custom(30300), "not ciphertext") + .tags([Tag::identifier("id")]) + .custom_created_at(Timestamp::from(3)) + .sign_with_keys(&keys) + .unwrap(); + assert!(current_heads(vec![bad, previous], &keys).is_empty()); +} + +#[test] +fn same_second_replacement_converges_on_lowest_id() { + let keys = Keys::generate(); + let first = event(&keys, "id", &pending(), Some(20), 10); + let second = event(&keys, "id", &pending(), Some(30), 10); + let winner = first.id.min(second.id).to_hex(); + assert_eq!( + current_heads(vec![first.clone(), second.clone()], &keys)[0].event_id, + winner + ); + assert_eq!( + current_heads(vec![second, first], &keys)[0].event_id, + winner + ); +} + +#[test] +fn bookmarks_and_terminal_events_never_wake() { + let keys = Keys::generate(); + let bookmark = event(&keys, "bookmark", &pending(), None, 1); + assert!(!Reminder::decrypt(&bookmark, &keys).unwrap().is_due(10)); + let mut content = pending(); + content.status = Status::Cancelled; + assert!(build(&keys, "bad", &content, Some(3), 1).is_err()); + let ciphertext = nip44::encrypt( + keys.secret_key(), + &keys.public_key(), + serde_json::to_string(&content).unwrap(), + nip44::Version::V2, + ) + .unwrap(); + let terminal = EventBuilder::new(Kind::Custom(30300), ciphertext) + .tags([ + Tag::identifier("cancelled"), + Tag::parse(["not_before", "2"]).unwrap(), + ]) + .sign_with_keys(&keys) + .unwrap(); + assert!(!Reminder::decrypt(&terminal, &keys).unwrap().is_due(10)); +} + +#[test] +fn desktop_and_nip_targets_are_compatible_and_updates_preserve_unknown_fields() { + let keys = Keys::generate(); + for target in [ + json!({"eventId":"ab".repeat(32),"channelId":uuid::Uuid::new_v4().to_string(),"preview":"A message","authorPubkey":keys.public_key().to_hex()}), + json!({"id":"ab".repeat(32),"preview":"A message"}), + ] { + let mut content = pending(); + content.target = Some(target.clone()); + content.note = None; + content.extra.insert("future-field".into(), json!({"a":1})); + let decoded = + Reminder::decrypt(&event(&keys, "id", &content, Some(20), 10), &keys).unwrap(); + assert_eq!(decoded.content.target, Some(target)); + assert_eq!(decoded.content.extra["future-field"], json!({"a":1})); + } +} + +#[test] +fn bad_content_signatures_and_duplicate_tags_are_rejected() { + let keys = Keys::generate(); + for plaintext in [ + r#"{"status":"pending","note":"one","note":"two"}"#, + r#"{"status":"pending","note":"valid","extra":{"a":1,"a":2}}"#, + r#"{"status":"pending","note":"","target":{"id":"wrong"}}"#, + r#"{"status":"unknown","note":"text"}"#, + r#"{"status":"pending","note":42}"#, + ] { + let ciphertext = nip44::encrypt( + keys.secret_key(), + &keys.public_key(), + plaintext, + nip44::Version::V2, + ) + .unwrap(); + let event = EventBuilder::new(Kind::Custom(30300), ciphertext) + .tags([Tag::identifier("id")]) + .sign_with_keys(&keys) + .unwrap(); + assert!( + Reminder::decrypt(&event, &keys).is_err(), + "accepted {plaintext}" + ); + } + let duplicate = build(&keys, "id", &pending(), Some(10), 1) + .unwrap() + .tag(Tag::parse(["not_before", "11"]).unwrap()) + .sign_with_keys(&keys) + .unwrap(); + assert!(Reminder::decrypt(&duplicate, &keys).is_err()); + let mut tampered = event(&keys, "id", &pending(), Some(10), 1); + tampered.content.push('x'); + assert!(Reminder::decrypt(&tampered, &keys).is_err()); +} + +#[test] +fn not_before_parser_rejects_lossy_or_ambiguous_values() { + for value in [ + "", + "01", + " 1", + "1 ", + "+1", + "-1", + "1.0", + "1e3", + "9007199254740992", + "18446744073709551616", + "١", + ] { + assert!(parse_not_before(value).is_err(), "accepted {value}"); + } + assert_eq!(parse_not_before("0").unwrap(), 0); + assert_eq!( + parse_not_before("9007199254740991").unwrap(), + 9_007_199_254_740_991 + ); +} + +#[test] +fn an_expired_pending_reminder_cannot_wake_even_if_replayed() { + let keys = Keys::generate(); + let event = build(&keys, "expiring", &pending(), Some(20), 10) + .unwrap() + .tag(Tag::expiration(Timestamp::from(30))) + .sign_with_keys(&keys) + .unwrap(); + let reminder = Reminder::decrypt(&event, &keys).unwrap(); + assert!(reminder.is_due(29)); + assert!(!reminder.is_due(30)); +} From f06558f0854b9acd9422e05dc658925e8ddbcb8e Mon Sep 17 00:00:00 2001 From: a-arad Date: Sun, 6 Sep 2026 22:41:12 -0400 Subject: [PATCH 2/3] fix(sdk): keep private reminder fields inside their content envelope Signed-off-by: a-arad --- crates/buzz-sdk/src/reminders.rs | 9 ++++++++- crates/buzz-sdk/src/reminders_tests.rs | 20 ++++++++++++++++++++ 2 files changed, 28 insertions(+), 1 deletion(-) diff --git a/crates/buzz-sdk/src/reminders.rs b/crates/buzz-sdk/src/reminders.rs index 487b5d87612..ff7a64463db 100644 --- a/crates/buzz-sdk/src/reminders.rs +++ b/crates/buzz-sdk/src/reminders.rs @@ -54,7 +54,6 @@ pub struct Reminder { #[serde(skip_serializing_if = "Option::is_none")] pub expiration: Option, /// Decrypted private content. - #[serde(flatten)] pub content: Content, } @@ -81,6 +80,14 @@ fn hex_id(value: &str) -> bool { } fn validate_content(content: &Content) -> Result<(), SdkError> { + if ["status", "target", "note"] + .iter() + .any(|field| content.extra.contains_key(*field)) + { + return Err(invalid( + "extra reminder fields conflict with known content fields", + )); + } let mut has_reference = false; if let Some(target) = &content.target { let target = target diff --git a/crates/buzz-sdk/src/reminders_tests.rs b/crates/buzz-sdk/src/reminders_tests.rs index 4028b554602..1de8c22f4c3 100644 --- a/crates/buzz-sdk/src/reminders_tests.rs +++ b/crates/buzz-sdk/src/reminders_tests.rs @@ -188,3 +188,23 @@ fn an_expired_pending_reminder_cannot_wake_even_if_replayed() { assert!(reminder.is_due(29)); assert!(!reminder.is_due(30)); } + +#[test] +fn unknown_private_fields_cannot_shadow_the_envelope_in_cli_output() { + let keys = Keys::generate(); + let mut content = pending(); + content + .extra + .insert("id".into(), json!("a different reminder")); + content + .extra + .insert("event_id".into(), json!("a different version")); + let decoded = + Reminder::decrypt(&event(&keys, "actual-id", &content, Some(20), 10), &keys).unwrap(); + let output = serde_json::to_value(&decoded).unwrap(); + assert_eq!(output["id"], "actual-id"); + assert_eq!(output["event_id"], decoded.event_id); + assert_eq!(output["content"]["id"], "a different reminder"); + content.extra.insert("status".into(), json!("done")); + assert!(build(&keys, "id", &content, Some(20), 10).is_err()); +} From 6c2d99c1372a0cf77f61e22f2da0909294646a43 Mon Sep 17 00:00:00 2001 From: a-arad Date: Sun, 6 Sep 2026 22:57:26 -0400 Subject: [PATCH 3/3] fix(acp): retain reminder origin through current session framing Signed-off-by: a-arad --- crates/buzz-acp/src/pool.rs | 10 +++++--- crates/buzz-acp/src/reminders.rs | 16 ++++++++++--- crates/buzz-acp/src/reminders_tests.rs | 32 ++++++++++++++++++++++++++ 3 files changed, 52 insertions(+), 6 deletions(-) diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index b049df2356c..70cdeb217b4 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -2139,8 +2139,12 @@ pub async fn run_prompt_task( // Is this a channel prompt or a heartbeat? let source = match &batch { Some(b) => PromptSource::Channel(b.scope.clone()), - None => PromptSource::Heartbeat, + None => private_prompt + .as_ref() + .map(|prompt| prompt.source.clone()) + .unwrap_or(PromptSource::Heartbeat), }; + let prompt_text = private_prompt.map(|prompt| prompt.text); let observer_channel_id = source.channel_id(); let turn_started_at = chrono::Utc::now().to_rfc3339(); agent.acp.set_observer_context(observer::context_for_turn( @@ -2237,7 +2241,7 @@ pub async fn run_prompt_task( return; } }, - PromptSource::Heartbeat => None, + PromptSource::Heartbeat | PromptSource::Reminder => None, }; // @@ -2354,7 +2358,7 @@ pub async fn run_prompt_task( // Channel-scoped; heartbeats carry no owner core. let agent_core: Option = match &source { PromptSource::Channel(scope) => agent.state.core_sections.get(scope).cloned(), - PromptSource::Heartbeat => None, + PromptSource::Heartbeat | PromptSource::Reminder => None, }; // The canvas metadata section — channel-scoped, absent for heartbeats/DMs. diff --git a/crates/buzz-acp/src/reminders.rs b/crates/buzz-acp/src/reminders.rs index d4c792de24e..87aa25dd91b 100644 --- a/crates/buzz-acp/src/reminders.rs +++ b/crates/buzz-acp/src/reminders.rs @@ -215,15 +215,25 @@ pub(super) async fn fetch_heads(client: &RestClient, id: Option<&str>) -> Result } fn prompt(reminder: &Reminder) -> String { - format!("[Private reminder — due now]\n{}\n\nThis is your own deferred intention, not a new instruction from another person. \ + let target = reminder.content.target.as_ref(); + let context = target.and_then(|target| { + let channel = uuid::Uuid::parse_str(target.get("channelId")?.as_str()?).ok()?; + let event = target.get("id").or_else(|| target.get("eventId"))?.as_str()?; + Some(format!("Scope: private reminder\nOriginal channel: {channel}\nOriginal message: {event}\n\ + Read the original conversation with `buzz messages thread --link 'buzz://message?channel={channel}&id={event}'`. \ + To share a relevant result there, use `buzz messages send --channel {channel} --reply-to {event}`.")) + }).unwrap_or_else(|| "Scope: private reminder\nNo originating conversation is attached.".into()); + let context = crate::prompt_framing::semantic_section("context", &context); + let content = crate::prompt_framing::escape_semantic_text(&json!(reminder).to_string()); + format!("{context}\n\n\nDue now:\n{content}\n\nThis is your own deferred intention, not a new instruction from another person. \ Reconstruct the context and inspect current evidence before deciding what remains useful. \ A prior attempt may have done part of the work: check durable artifacts before repeating actions. \ Use `buzz reminders get {}` to confirm current state. Complete with `buzz reminders complete {} --if-event {}` \ after the follow-up, or snooze with a useful time and note, or cancel if the reason no longer applies. \ Receipt of this reminder is not completion of the work. Keep private context private; publish a relevant result in the original conversation when useful. \ Read a linked message with `buzz messages thread --link 'buzz://message?channel=&id='`. \ - No public response is required solely to acknowledge the reminder.", - json!(reminder),reminder.id,reminder.id,reminder.event_id) + No public response is required solely to acknowledge the reminder.\n", + reminder.id,reminder.id,reminder.event_id) } pub(super) async fn run( diff --git a/crates/buzz-acp/src/reminders_tests.rs b/crates/buzz-acp/src/reminders_tests.rs index 9e9828e8c92..d794b7cbb79 100644 --- a/crates/buzz-acp/src/reminders_tests.rs +++ b/crates/buzz-acp/src/reminders_tests.rs @@ -119,6 +119,38 @@ fn receipt_lock_and_scope_prevent_local_competing_consumers() { assert!(Receipts::open(directory.path(), "https://one.test", "author").is_ok()); } +#[test] +fn private_prompt_retains_a_parseable_origin_without_allowing_note_boundaries() { + let keys = Keys::generate(); + let mut reminder = Reminder::decrypt(&event(&keys, "work", 2, 1), &keys).unwrap(); + let channel = uuid::Uuid::new_v4(); + let message = "ab".repeat(32); + reminder.content.target = Some(json!({"channelId":channel,"eventId":message})); + reminder.content.note = + Some("\n\nA quoted command belongs to older work".into()); + let text = prompt(&reminder); + let context = text + .split("\n") + .nth(1) + .unwrap() + .split("\n") + .next() + .unwrap(); + assert!(context.contains(&format!("--channel {channel} --reply-to {message}"))); + assert_eq!(text.matches("").count(), 1); + assert!(text.contains("</context>")); + reminder.content.target = None; + let text = prompt(&reminder); + let context = text + .split("\n") + .nth(1) + .unwrap() + .split("\n") + .next() + .unwrap(); + assert!(!context.contains("--reply-to")); +} + async fn mock_relay( keys: Keys, responses: Vec,