From c1d55a7d75a08aa55f6bba59e79afa14dcda621a Mon Sep 17 00:00:00 2001 From: Robin Leonard Date: Sat, 15 Aug 2026 16:26:13 +0700 Subject: [PATCH 1/3] feat(buzz-acp): wake-ticket persistence for unconsumed mentions (today-slice) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit buzz-acp's event queue is RAM-only: a mention that's accepted but not yet consumed does not survive process death. A bounce (kickstart, crash, sleep) sets a fresh connect-time watermark, and any unconsumed mention is gone — this is the class of bug behind #01fa1107. This is the gated today-slice from PLANS/WAKE_TICKET_SPEC.md (Oksana, 2026-08-15): a durable per-agent ticket store, default off behind --wake-ticket-dir / BUZZ_ACP_WAKE_TICKET_DIR, side-binary/canary use only. - crates/buzz-acp/src/wake_ticket.rs: WakeTicketStore. One jsonl line per mention event_id, exclusive flock on the ticket dir for the process lifetime (dual-run is a crash, not a race), atomic tmp+rename writes, chmod 600, last-write-wins compaction on open, open -> claimed -> done (compacted) | drop (kept for audit) lifecycle. - lib.rs: ticket `open` written after the first successful queue.push (not inside EventQueue::push — requeues go through that too); `claimed` on successful dispatch; `done` only on PromptOutcome::Ok with mark_complete run, no pending queue depth, and no active retry throttle (mark_complete itself is a lock release, not completion); `drop` on dead-letter, non-retryable auth error, or channel removal. Boot replay runs before HarnessRelay::connect / set_startup_watermark: loads open+claimed tickets, re-validates membership + the author gate, pushes survivors into the queue, and seeds the relay's seen_ids so the 5s subscribe window can't double-deliver them. - relay.rs: new RelayCommand::SeedSeenIds plumbing for the above. - queue.rs: EventQueue::is_retry_throttled, and queued_event_count is no longer test-only (needed for the done predicate). Unit tests cover the ticket store directly (open/claimed/done-predicate/ drop/replay-after-restart/lock-fail/cap), per the gate. Out of scope for this slice, per the gate: nothing points launchd at the side binary, the shipped /Applications/Buzz.app binary is untouched, and no canary has run. That's the next step, gated on Oksana's review here plus a Zar-only bounce proof. Signed-off-by: Robin Leonard --- Cargo.lock | 36 +- crates/buzz-acp/Cargo.toml | 4 + crates/buzz-acp/src/config.rs | 13 + crates/buzz-acp/src/lib.rs | 234 ++++++++++- crates/buzz-acp/src/pool.rs | 5 + crates/buzz-acp/src/queue.rs | 12 +- crates/buzz-acp/src/relay.rs | 33 ++ crates/buzz-acp/src/wake_ticket.rs | 633 +++++++++++++++++++++++++++++ 8 files changed, 952 insertions(+), 18 deletions(-) create mode 100644 crates/buzz-acp/src/wake_ticket.rs diff --git a/Cargo.lock b/Cargo.lock index 6c46beedf2f..132b737a59f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -836,6 +836,7 @@ dependencies = [ "chrono", "clap", "evalexpr", + "fs2", "futures-util", "hex", "httparse", @@ -846,6 +847,7 @@ dependencies = [ "serde", "serde_json", "sha2 0.11.0", + "tempfile", "thiserror 2.0.18", "tokio", "tokio-tungstenite 0.29.0", @@ -1712,7 +1714,7 @@ version = "3.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "faf9468729b8cbcea668e36183cb69d317348c2e08e994829fb56ebfdfbaac34" dependencies = [ - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -2333,7 +2335,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ccc2776f0c61eca1ca32528f85548abd1a4be8fb53d1b21c013e4f18da1e7090" dependencies = [ "data-encoding", - "syn 1.0.109", + "syn 2.0.117", ] [[package]] @@ -2544,7 +2546,7 @@ dependencies = [ "libc", "option-ext", "redox_users", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -2767,7 +2769,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -3032,6 +3034,16 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "fs2" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9564fc758e15025b46aa6643b1b77d047d1a56a1aea6e01002ac0c7026876213" +dependencies = [ + "libc", + "winapi", +] + [[package]] name = "fs_extra" version = "1.3.0" @@ -3183,7 +3195,7 @@ dependencies = [ "libc", "log", "rustversion", - "windows-link 0.1.3", + "windows-link 0.2.1", "windows-result 0.4.1", ] @@ -6028,7 +6040,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -7501,7 +7513,7 @@ dependencies = [ "once_cell", "socket2", "tracing", - "windows-sys 0.59.0", + "windows-sys 0.60.2", ] [[package]] @@ -8185,7 +8197,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -8244,7 +8256,7 @@ dependencies = [ "security-framework 3.7.0", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -8527,7 +8539,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5b55fb86dfd3a2f5f76ea78310a88f96c4ea21a3031f8d212443d56123fd0521" dependencies = [ "libc", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -9652,7 +9664,7 @@ dependencies = [ "getrandom 0.4.3", "once_cell", "rustix 1.1.4", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -11019,7 +11031,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] diff --git a/crates/buzz-acp/Cargo.toml b/crates/buzz-acp/Cargo.toml index d047849806f..107d853984b 100644 --- a/crates/buzz-acp/Cargo.toml +++ b/crates/buzz-acp/Cargo.toml @@ -71,6 +71,9 @@ toml = "1.0" # Filter expressions evalexpr = { workspace = true } +# Wake-ticket dir locking (exclusive flock) — same crate rustup itself uses. +fs2 = "0.4" + # 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] @@ -79,3 +82,4 @@ nix = { version = "0.31", default-features = false, features = ["signal"] } [dev-dependencies] tokio = { workspace = true, features = ["test-util"] } httparse = "1" +tempfile = "3" diff --git a/crates/buzz-acp/src/config.rs b/crates/buzz-acp/src/config.rs index f9e7bf1ed8a..ada388a6bf3 100644 --- a/crates/buzz-acp/src/config.rs +++ b/crates/buzz-acp/src/config.rs @@ -489,6 +489,13 @@ pub struct CliArgs { /// Requires `--lazy-pool`; ignored otherwise. 0 disables idle re-sleep. #[arg(long, env = "BUZZ_ACP_IDLE_POOL_SLEEP", default_value_t = 0)] pub idle_pool_sleep: u64, + + /// Directory for wake-ticket persistence (durable record of unconsumed + /// mentions, so a bounced process can resume them). Default off — when + /// unset, no ticket store is created and behavior is unchanged from + /// today. Side-binary / canary use only; see `PLANS/WAKE_TICKET_SPEC.md`. + #[arg(long, env = "BUZZ_ACP_WAKE_TICKET_DIR")] + pub wake_ticket_dir: Option, } /// Merged NIP-01 subscription filter for a single channel. @@ -579,6 +586,10 @@ pub struct Config { /// `from_cli()`. `None` when using the compiled-in default or when /// `--no-base-prompt` is set. pub base_prompt_content: Option, + /// Directory for wake-ticket persistence. `None` (the default) disables + /// the feature entirely — no ticket store is opened, no boot replay + /// runs, and the hot path is unchanged from pre-wake-ticket behavior. + pub wake_ticket_dir: Option, } /// Maximum length, in characters, of a session title sent to the adapter. @@ -1122,6 +1133,7 @@ impl Config { agent_owner: args.agent_owner.map(|s| s.trim().to_ascii_lowercase()), no_base_prompt: args.no_base_prompt, base_prompt_content, + wake_ticket_dir: args.wake_ticket_dir, }; Ok(config) @@ -1494,6 +1506,7 @@ mod tests { agent_owner: None, no_base_prompt: false, base_prompt_content: None, + wake_ticket_dir: None, } } diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 7fd40b83db1..e9bc50bdb9f 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -11,6 +11,7 @@ mod queue; mod relay; mod setup_mode; mod usage; +mod wake_ticket; pub use usage::TurnUsage; @@ -46,6 +47,7 @@ use relay::{HarnessRelay, RelayEventPublisher}; use tokio::sync::{mpsc, watch}; use tracing_subscriber::EnvFilter; use uuid::Uuid; +use wake_ticket::WakeTicketStore; /// Check if argv[1] matches a subcommand name, before any clap parsing. /// @@ -286,6 +288,40 @@ pub(crate) async fn is_dm_channel( } } +/// Re-validate that this agent is still a member of `channel_id`, for +/// wake-ticket boot replay (`PLANS/WAKE_TICKET_SPEC.md` §Gate: "Replay +/// re-validates membership + author gate. Fail → drop."). +/// +/// Queries kind:39002 (NIP-29 group members) scoped to both this pubkey and +/// the one channel — the same shape as `HarnessRelay::discover_channels`'s +/// Step 1, narrowed so replay doesn't refetch the whole membership set per +/// ticket. Fail-closed: a query error means "not a member" rather than +/// risking a stale mention replaying into a channel this unit lost access to. +async fn is_still_member(channel_id: Uuid, pubkey_hex: &str, rest: &relay::RestClient) -> bool { + use nostr::{Alphabet, SingleLetterTag}; + + let p_tag = SingleLetterTag::lowercase(Alphabet::P); + let d_tag = SingleLetterTag::lowercase(Alphabet::D); + let filter = nostr::Filter::new() + .kind(nostr::Kind::Custom( + buzz_core::kind::KIND_NIP29_GROUP_MEMBERS as u16, + )) + .custom_tags(p_tag, [pubkey_hex.to_string()]) + .custom_tags(d_tag, [channel_id.to_string()]); + + match rest.query(&[filter]).await { + Ok(v) => v.as_array().is_some_and(|arr| !arr.is_empty()), + Err(e) => { + tracing::warn!( + channel_id = %channel_id, + error = %e, + "wake-ticket replay: membership check failed — treating as not-a-member (fail closed)" + ); + false + } + } +} + /// Query an author's kind:0 profile and check if their NIP-OA auth tag /// proves the same owner as us. async fn check_sibling_via_profile( @@ -1973,6 +2009,103 @@ async fn tokio_main() -> Result<()> { .filter(|s| !s.is_empty()) .and_then(|s| buzz_sdk::nip_oa::parse_auth_tag(&s).ok()); + // ── Wake-ticket store + boot replay ───────────────────────────────────── + // + // Must run before `HarnessRelay::connect()` / `set_startup_watermark()` + // below — see `PLANS/WAKE_TICKET_SPEC.md` §Gate ("Snapshot, not + // REQ-by-id"). Replayed events need to already be in `queue` before the + // relay subscribes, so the connect-time watermark can never treat them + // as pre-history; `seed_seen_ids` (sent right after connect, before any + // channel subscribe) needs their ids ready by then too. `queue` itself + // therefore also moves up from its old post-connect declaration. + let wake_tickets: Option> = match &config.wake_ticket_dir { + Some(dir) => match WakeTicketStore::open(dir) { + Ok(store) => Some(Arc::new(store)), + Err(e) => { + // Lock-fail / unavailable dir is fatal: dual-run of the same + // identity against the same ticket dir is a crash, not a race. + anyhow::bail!("wake-ticket store at {} failed to open: {e}", dir.display()); + } + }, + None => None, + }; + + let dedup_mode = config.dedup_mode; + let mut queue = + EventQueue::new(dedup_mode).with_in_flight_deadline(config.max_turn_duration_secs); + let mut wake_ticket_replay_seen_ids: Vec = Vec::new(); + + if let Some(store) = &wake_tickets { + let mut pending = store.pending_for_replay(); + // Deterministic order: oldest mention first, so a crash-loop doesn't + // keep reshuffling replay order and starve an old ticket. + pending.sort_by_key(|t| t.created_at); + + if !pending.is_empty() { + // Lightweight, pre-connect REST client sharing this run's HTTP + // credentials — mirrors `HarnessRelay::rest_client()`, but no + // `HarnessRelay` exists yet this early in boot. + let replay_rest_client = relay::RestClient { + http: reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(10)) + .connect_timeout(std::time::Duration::from_secs(5)) + .build() + .map_err(|e| { + anyhow::anyhow!("wake-ticket replay: failed to build HTTP client: {e}") + })?, + base_url: relay::relay_ws_to_http(&config.relay_url), + keys: config.keys.clone(), + auth_tag_json: relay_auth_tag + .as_ref() + .and_then(|t| serde_json::to_string(t.as_slice()).ok()), + }; + let replay_owner_cache = OwnerCache::new(resolve_agent_owner(&config)); + let replay_channel_info = + pool::ChannelInfoResolver::new(HashMap::new(), replay_rest_client.clone()); + + let mut replayed = 0usize; + let mut dropped = 0usize; + for ticket in pending { + let is_dm = is_dm_channel(ticket.channel_id, &replay_channel_info).await; + let author = ticket.event.pubkey.to_hex(); + let allowed = author_allowed( + &config.respond_to, + &config.respond_to_allowlist, + &author, + is_dm, + &replay_owner_cache, + &replay_rest_client, + ) + .await; + let still_member = + is_still_member(ticket.channel_id, &pubkey_hex, &replay_rest_client).await; + + if !allowed || !still_member { + tracing::info!( + event_id = %ticket.event_id, + channel_id = %ticket.channel_id, + allowed, + still_member, + "wake-ticket replay: re-validation failed — dropping" + ); + store.drop_unvalidated(&ticket.event_id); + dropped += 1; + continue; + } + + queue.push(QueuedEvent { + channel_id: ticket.channel_id, + event: ticket.event.clone(), + received_at: std::time::Instant::now(), + prompt_tag: ticket.prompt_tag.clone(), + }); + wake_ticket_replay_seen_ids.push(ticket.event_id.clone()); + replayed += 1; + } + tracing::info!(replayed, dropped, "wake-ticket boot replay complete"); + } + } + let mut relay = HarnessRelay::connect(&config.relay_url, &config.keys, &pubkey_hex, relay_auth_tag) .await @@ -1986,6 +2119,16 @@ async fn tokio_main() -> Result<()> { tracing::warn!("failed to set startup watermark: {e}"); } + // Seed seen_ids for every wake-ticket event replayed above, before any + // channel subscription is sent — otherwise the 5s subscribe-window dedup + // cannot tell a replayed ticket apart from a genuine live re-delivery of + // the same event and would hand it to the agent twice. + if !wake_ticket_replay_seen_ids.is_empty() { + if let Err(e) = relay.seed_seen_ids(wake_ticket_replay_seen_ids).await { + tracing::warn!("failed to seed seen_ids from wake-ticket replay: {e}"); + } + } + tracing::info!("connected to relay at {}", config.relay_url); relay @@ -2134,9 +2277,6 @@ async fn tokio_main() -> Result<()> { } let runtime_start_nonce = std::env::var("BUZZ_MANAGED_AGENT_START_NONCE").unwrap_or_default(); - let dedup_mode = config.dedup_mode; - let mut queue = - EventQueue::new(dedup_mode).with_in_flight_deadline(config.max_turn_duration_secs); // Online means the harness can receive work, not merely that its socket is // connected. Publishing after channel subscriptions gives desktop callers @@ -2194,6 +2334,7 @@ async fn tokio_main() -> Result<()> { memory_enabled: config.memory_enabled, harness_name: crate::config::normalize_agent_command_identity(&config.agent_command), relay_url: config.relay_url.clone(), + wake_tickets: wake_tickets.clone(), }); if !config.memory_enabled { @@ -2680,6 +2821,12 @@ async fn tokio_main() -> Result<()> { // complete normally (the relay may reject actions if // the agent lost access). let drained_ids = queue.drain_channel(ch); + if let Some(store) = &ctx.wake_tickets { + // Channel-removed: drop, never replay + // (gate §Gate — dead-letter / auth-fail / + // channel-removed all drop, keep for audit). + store.mark_drop(&drained_ids); + } let invalidated = if pool_ready { pool.invalidate_channel_sessions(ch) } else { @@ -2893,6 +3040,26 @@ async fn tokio_main() -> Result<()> { received_at: std::time::Instant::now(), prompt_tag, }); + // Wake ticket: durable `open` record, written only for + // events the in-memory queue actually accepted (a + // Drop-mode discard for an in-flight channel is not a + // ticket — see `EventQueue::push`). Uses `event_for_steer` + // (already cloned above) since `buzz_event.event` was + // just moved into the queue push. + if accepted { + if let Some(store) = &ctx.wake_tickets { + store.write_open( + &event_for_steer, + buzz_event.channel_id, + &pubkey_hex, + &prompt_tag_for_steer, + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs(), + ); + } + } // 👀 — immediate "seen" reaction, only if the event // was actually queued (not dropped by DedupMode::Drop). // Fire-and-forget: on rare fast-failure paths the @@ -3130,6 +3297,7 @@ async fn tokio_main() -> Result<()> { &mut respawn_tasks, observer.clone(), Some(&ctx.rest_client), + ctx.wake_tickets.as_deref(), ) == LoopAction::Exit { break; @@ -3719,6 +3887,11 @@ fn dispatch_pending( }; tracing::debug!(agent = agent.index, channel = %channel_id, affinity_hit, "agent_claimed"); + if let Some(store) = &ctx.wake_tickets { + let ids: Vec = batch.events.iter().map(|e| e.event.id.to_hex()).collect(); + store.mark_claimed(&ids); + } + let recoverable_batch = match ctx.dedup_mode { DedupMode::Queue => Some(batch.clone()), DedupMode::Drop => None, @@ -3848,6 +4021,7 @@ fn handle_prompt_result( respawn_tasks: &mut tokio::task::JoinSet<()>, observer: Option, rest_client: Option<&relay::RestClient>, + wake_tickets: Option<&wake_ticket::WakeTicketStore>, ) -> LoopAction { let before = pool.task_map().len(); let agent_index = result.agent.index; @@ -3884,6 +4058,17 @@ fn handle_prompt_result( // branch below records what actually happened; only the hard-timeout // match arm in the death_message construction reads it. let mut hard_timeout_fate_suffix: Option<&'static str> = None; + // Wake-ticket fate, decided alongside the batch's requeue/dead-letter + // fate below (same branches, same reasoning) rather than re-derived from + // `result.outcome` afterward — `result.batch` is `None` on `Ok` (nothing + // to requeue) and also `None` in `DedupMode::Drop` regardless of + // outcome, so "was this batch actually dead-lettered" can only be + // answered inside this match, not reconstructed from the outcome alone. + // `true` = every ticket still `claimed` for this channel becomes `drop` + // (dead-lettered / auth-failed / channel-removed — never replayed). + // `false` leaves tickets as-is (open/claimed) unless the `Ok` path below + // separately marks them `done`. + let mut should_drop_tickets = false; // Requeue BEFORE mark_complete: requeue() sets retry_after with a future // deadline, and mark_complete() checks for it to decide whether to preserve @@ -3932,6 +4117,7 @@ fn handle_prompt_result( ); spawn_failure_notice(rest_client, &batch, content); hard_timeout_fate_suffix = Some(" — dead-lettered (no recent activity)"); + should_drop_tickets = true; } else if matches!( result.outcome, PromptOutcome::Timeout(TimeoutKind::Hard { @@ -3950,6 +4136,7 @@ fn handle_prompt_result( ); spawn_failure_notice(rest_client, &dead, content); hard_timeout_fate_suffix = Some(" — dead-lettered (retry budget exhausted)"); + should_drop_tickets = true; } else { hard_timeout_fate_suffix = Some(" — requeued for retry (recently active)"); } @@ -3968,6 +4155,7 @@ fn handle_prompt_result( and then re-send." .to_string(); spawn_failure_notice(rest_client, &batch, content); + should_drop_tickets = true; } else if let Some(dead) = queue.requeue(batch) { let reason = match &result.outcome { PromptOutcome::Timeout(TimeoutKind::Idle) => "the turn timed out".to_string(), @@ -3982,6 +4170,7 @@ fn handle_prompt_result( "⚠️ I couldn't process the last request after multiple retries ({reason}). Please re-send if it's still needed." ); spawn_failure_notice(rest_client, &dead, content); + should_drop_tickets = true; } } else { tracing::debug!( @@ -3990,11 +4179,33 @@ fn handle_prompt_result( "dropping failed batch for removed channel" ); hard_timeout_fate_suffix = Some(" — batch dropped (channel removed)"); + should_drop_tickets = true; } } match &result.source { - PromptSource::Channel(ch) => queue.mark_complete(*ch), + PromptSource::Channel(ch) => { + queue.mark_complete(*ch); + // `done`/`drop` predicate — see `PLANS/WAKE_TICKET_SPEC.md` §Gate + // ("`mark_complete` is not `done`"). Looked up by channel rather + // than threaded from the dispatched batch: `result.batch` is + // `None` on the `Ok` path (nothing to requeue) and always `None` + // in `DedupMode::Drop`, so it cannot carry the completed event + // ids here. + if let Some(store) = wake_tickets { + if should_drop_tickets { + store.mark_drop(&store.claimed_event_ids_for_channel(*ch)); + } else if matches!(result.outcome, PromptOutcome::Ok(_)) + && queue.queued_event_count(ch) == 0 + && !queue.is_retry_throttled(ch) + { + store.mark_done(&store.claimed_event_ids_for_channel(*ch)); + } + // Cancelled/CancelDrainTimeout/requeued-with-budget-remaining/ + // pool-exhausted/panic: leave tickets `open` or `claimed` — + // the work is still pending, not lost and not finished. + } + } PromptSource::Heartbeat => *heartbeat_in_flight = false, } @@ -6765,6 +6976,7 @@ mod build_mcp_servers_tests { agent_owner: None, no_base_prompt: false, base_prompt_content: None, + wake_ticket_dir: None, } } @@ -6988,6 +7200,7 @@ mod error_outcome_emission_tests { agent_owner: None, no_base_prompt: false, base_prompt_content: None, + wake_ticket_dir: None, } } @@ -7093,6 +7306,7 @@ mod error_outcome_emission_tests { &mut respawn_tasks, None, None, + None, ); let returned = pool.agents_mut()[0].as_ref().expect("returned agent"); @@ -7165,6 +7379,7 @@ mod error_outcome_emission_tests { &mut respawn_tasks, None, None, + None, ); let returned = pool.agents_mut()[0].as_ref().expect("returned agent"); @@ -7279,6 +7494,7 @@ mod error_outcome_emission_tests { &mut respawn_tasks, None, None, + None, ); let returned = pool.agents_mut()[0].as_ref().expect("returned agent"); @@ -7342,6 +7558,7 @@ mod error_outcome_emission_tests { &mut respawn_tasks, Some(observer.clone()), None, + None, ); let turn_errors: Vec<_> = observer @@ -7509,6 +7726,7 @@ mod error_outcome_emission_tests { &mut respawn_tasks, Some(observer.clone()), None, + None, ); let events = observer.snapshot(); let turn_error = events.iter().find(|e| e.kind == "turn_error").unwrap(); @@ -7600,6 +7818,7 @@ mod error_outcome_emission_tests { &mut respawn_tasks, None, None, + None, ); ( queue.pending_channels(), @@ -7706,6 +7925,7 @@ mod error_outcome_emission_tests { &mut respawn_tasks, None, None, + None, ); ( queue.pending_channels(), @@ -7798,6 +8018,7 @@ mod error_outcome_emission_tests { &mut respawn_tasks, Some(observer.clone()), None, + None, ); let events = observer.snapshot(); @@ -7892,6 +8113,7 @@ mod error_outcome_emission_tests { &mut respawn_tasks, Some(observer.clone()), None, + None, ); let events = observer.snapshot(); @@ -8008,6 +8230,7 @@ mod error_outcome_emission_tests { &mut respawn_tasks, Some(observer.clone()), None, + None, ); // Batch preserved as a cancelled merge, not dead-lettered — same @@ -8141,6 +8364,7 @@ mod error_outcome_emission_tests { &mut respawn_tasks, Some(observer.clone()), None, + None, ); // No batch to merge — the queue has nothing pending for any channel. @@ -8324,6 +8548,7 @@ mod error_outcome_emission_tests { &mut respawn_tasks, None, None, + None, ); // The batch must not be requeued: pending_channels returns 0. @@ -8410,6 +8635,7 @@ mod error_outcome_emission_tests { &mut respawn_tasks, None, None, + None, ); // Non-auth application error: batch IS requeued (first attempt, retry budget > 0). diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 2efacce2b19..8ce3b8365f2 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -606,6 +606,10 @@ pub struct PromptContext { /// the desktop keys per (agent, relay) pair, e.g. `session_config_captured`, /// mirroring the `managed_agent_runtime_lifecycle` frames. pub relay_url: String, + /// Durable wake-ticket store, if `--wake-ticket-dir` / `BUZZ_ACP_WAKE_TICKET_DIR` + /// is set. `None` (the default) disables the feature entirely — no ticket + /// writes, no claim/done/drop transitions. See `PLANS/WAKE_TICKET_SPEC.md`. + pub wake_tickets: Option>, } impl AgentPool { @@ -7600,6 +7604,7 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" memory_enabled: false, harness_name: "goose".to_string(), relay_url: "ws://127.0.0.1:3000".to_string(), + wake_tickets: None, } } diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index b0f0fa248e3..e13874dcad1 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -628,8 +628,16 @@ impl EventQueue { self.queues.len() } - /// Number of queued events for a specific channel. Test-only. - #[cfg(test)] + /// Whether `channel_id` is currently under an active `retry_after` + /// backoff throttle (a prior attempt was requeued and hasn't reached + /// its next-try deadline yet). + pub fn is_retry_throttled(&self, channel_id: &Uuid) -> bool { + self.retry_after + .get(channel_id) + .is_some_and(|&deadline| deadline > Instant::now()) + } + + /// Number of queued (not-yet-dispatched) events for a specific channel. pub fn queued_event_count(&self, channel_id: &Uuid) -> usize { self.queues.get(channel_id).map_or(0, |q| q.len()) } diff --git a/crates/buzz-acp/src/relay.rs b/crates/buzz-acp/src/relay.rs index 17a818867dd..5e996b3e9e9 100644 --- a/crates/buzz-acp/src/relay.rs +++ b/crates/buzz-acp/src/relay.rs @@ -551,6 +551,11 @@ enum RelayCommand { PublishEvent { event: Box }, /// Floor `since` for membership notification replay; events before startup are never re-delivered. SetStartupWatermark { ts: u64 }, + /// Pre-seed `BgState::seen_ids` with event ids the caller has already + /// consumed out-of-band (wake-ticket boot replay) — so the 5s + /// subscribe-window dedup treats them as already-delivered instead of + /// handing them to the agent a second time. + SeedSeenIds { ids: Vec }, } type WsStream = WebSocketStream>; @@ -912,6 +917,21 @@ impl HarnessRelay { .map_err(|_| RelayError::ConnectionClosed) } + /// Seed `seen_ids` with event ids already consumed via wake-ticket boot + /// replay, before subscribing to any channel. Call before + /// `subscribe_channel` / `subscribe_membership_notifications` — the 5s + /// subscribe window otherwise cannot tell a replayed ticket apart from a + /// genuine live re-delivery of the same event. + pub async fn seed_seen_ids(&self, ids: Vec) -> Result<(), RelayError> { + if ids.is_empty() { + return Ok(()); + } + self.cmd_tx + .send(RelayCommand::SeedSeenIds { ids }) + .await + .map_err(|_| RelayError::ConnectionClosed) + } + /// Reconnect after connection loss. Instructs the background task to /// re-authenticate and resubscribe to all previously active channels. pub async fn reconnect(&mut self) -> Result<(), RelayError> { @@ -1308,6 +1328,11 @@ fn apply_command_to_state(state: &mut BgState, cmd: RelayCommand) { state.membership_last_seen = Some(ts); } } + RelayCommand::SeedSeenIds { ids } => { + for id in ids { + state.seen_ids.insert(id); + } + } // Observer telemetry frames are durable: park them (bounded, visible // overflow) so they are delivered by the post-reconnect drain. Other // ephemeral publishes (typing indicators) are meaningless while @@ -1546,6 +1571,14 @@ async fn execute_connected_command( debug!("startup watermark set to {ts}"); true } + RelayCommand::SeedSeenIds { ids } => { + let count = ids.len(); + for id in ids { + state.seen_ids.insert(id); + } + debug!(count, "seeded seen_ids from wake-ticket replay"); + true + } // Control-flow commands — callers handle these before dispatching. RelayCommand::Shutdown | RelayCommand::Reconnect => { debug_assert!( diff --git a/crates/buzz-acp/src/wake_ticket.rs b/crates/buzz-acp/src/wake_ticket.rs new file mode 100644 index 00000000000..3a9bd5da8b2 --- /dev/null +++ b/crates/buzz-acp/src/wake_ticket.rs @@ -0,0 +1,633 @@ +//! Wake-ticket persistence — today-slice. +//! +//! `EventQueue` (see [`crate::queue`]) is RAM-only: a mention that is queued +//! but not yet consumed does not survive process death. This module gives +//! those in-flight mentions a durable record on disk so a bounced `buzz-acp` +//! process can pick the work back up, even though the connect-time watermark +//! (`lib.rs`) would otherwise treat it as pre-history. +//! +//! Scope is deliberately narrow — see `PLANS/WAKE_TICKET_SPEC.md` §Gate +//! (Oksana, 2026-08-15) for the binding contract this module implements: +//! +//! - One ticket per mention `event_id`, keyed by that id. +//! - `open` → `claimed` → `done` (compacted away) or `drop` (kept for audit). +//! - `mark_complete` is a lock release, not a completion signal — callers in +//! `lib.rs` decide `done` from `PromptOutcome::Ok` plus "no batch to retry". +//! - Persistence is best-effort after construction: a write failure is +//! logged and swallowed so a ticket-store hiccup never fails a live turn. +//! Only [`WakeTicketStore::open`] is fallible — an unavailable or +//! already-locked directory must stop the process before it double-runs. +//! +//! This module has no knowledge of the relay, the author gate, or channel +//! membership — boot-time replay validation lives in `lib.rs`, which already +//! owns those checks for the live event path. + +use std::collections::HashMap; +use std::fs::{self, File, OpenOptions}; +use std::io::{self, BufRead, BufReader, Write}; +use std::path::{Path, PathBuf}; +use std::sync::Mutex; + +use fs2::FileExt; +use serde::{Deserialize, Serialize}; +use uuid::Uuid; + +/// Maximum open+claimed tickets retained per channel. Mirrors +/// `queue::MAX_PENDING_PER_CHANNEL` — a ticket is only ever written for an +/// event the in-memory queue already accepted, so this is a belt-and-suspenders +/// cap, not the primary backpressure mechanism. +const MAX_PENDING_PER_CHANNEL: usize = 500; + +const TICKETS_FILE: &str = "wake-tickets.jsonl"; +const LOCK_FILE: &str = ".wake-tickets.lock"; + +/// Lifecycle state of a wake ticket. See `PLANS/WAKE_TICKET_SPEC.md` §Lifecycle. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum TicketState { + /// Accepted by `EventQueue::push`; not yet handed to an agent. + Open, + /// Drained into a `FlushBatch` an agent is actively working. + Claimed, + /// Turn completed successfully and nothing was requeued. Compacted away + /// on the next [`WakeTicketStore::open`]. + Done, + /// Dead-lettered, auth-failed, or the channel was removed. Kept on disk + /// for audit; never replayed. + Drop, +} + +/// One durable record of an unconsumed (or recently consumed) mention. +/// +/// `event` is the signed `nostr::Event` as accepted — the mention itself, +/// not an assembled LLM prompt and not an ACP transcript. No private keys. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Ticket { + pub event_id: String, + pub channel_id: Uuid, + /// Event's own `created_at`, not local clock. + pub created_at: u64, + /// The mentioned agent (this unit). + pub pubkey: String, + /// When this unit first queued the event. + pub seen_at: u64, + pub state: TicketState, + pub event: nostr::Event, + /// The rule tag that matched this event in `filter::match_event` at + /// write time (e.g. `"@mention"`). Boot replay reconstructs a + /// `QueuedEvent` straight from the ticket without re-running rule + /// matching — subscription rules aren't available that early (channel + /// discovery hasn't happened, there's no live relay yet) and rules may + /// have changed since the ticket was written anyway. Captured, not + /// derived, so replay reproduces what was actually queued. + pub prompt_tag: String, +} + +struct Inner { + file: File, + /// Last-write-wins mirror of every line ever appended this run, minus + /// what boot compaction already dropped. Keyed by `event_id`. + index: HashMap, +} + +/// Durable, single-writer ticket store for one agent's wake-ticket directory. +/// +/// Holds an exclusive `flock` on a lock file inside `dir` for the process +/// lifetime. A second process opening the same directory fails immediately — +/// dual-run of the same identity against the same ticket dir is a crash, not +/// a race, so [`WakeTicketStore::open`] never blocks waiting for the lock. +pub struct WakeTicketStore { + dir: PathBuf, + // Held for the process lifetime; never read after construction. Dropping + // it releases the flock, so it must outlive every other use of `dir`. + _lock_file: File, + inner: Mutex, +} + +#[cfg(unix)] +fn chmod_600(path: &Path) -> io::Result<()> { + use std::os::unix::fs::PermissionsExt; + fs::set_permissions(path, fs::Permissions::from_mode(0o600)) +} + +#[cfg(not(unix))] +fn chmod_600(_path: &Path) -> io::Result<()> { + Ok(()) +} + +impl WakeTicketStore { + /// Open (creating if needed) the ticket store rooted at `dir`. + /// + /// Acquires the exclusive flock, loads any existing `wake-tickets.jsonl`, + /// collapses it last-write-wins by `event_id`, drops `done` entries, and + /// rewrites the compacted result via tmp+rename before returning. Fails + /// if the lock is already held or any filesystem step fails — callers + /// must treat that as fatal, per the gate ("Lock fail → exit"). + pub fn open(dir: &Path) -> io::Result { + fs::create_dir_all(dir)?; + #[cfg(unix)] + chmod_dir_700(dir)?; + + let lock_path = dir.join(LOCK_FILE); + let lock_file = OpenOptions::new() + .create(true) + .truncate(false) + .write(true) + .open(&lock_path)?; + lock_file.try_lock_exclusive().map_err(|_| { + io::Error::other(format!( + "wake-ticket dir {} is already locked by another process \ + — dual-run is a hard stop", + dir.display() + )) + })?; + + let existing = Self::load_raw(dir)?; + let survivors: Vec = existing + .into_iter() + .filter(|t| t.state != TicketState::Done) + .collect(); + Self::write_compacted(dir, &survivors)?; + + let tickets_path = dir.join(TICKETS_FILE); + let file = OpenOptions::new().append(true).open(&tickets_path)?; + + let index = survivors + .into_iter() + .map(|t| (t.event_id.clone(), t)) + .collect(); + + Ok(Self { + dir: dir.to_path_buf(), + _lock_file: lock_file, + inner: Mutex::new(Inner { file, index }), + }) + } + + /// Load `wake-tickets.jsonl` (if present) and collapse to last-write-wins + /// per `event_id`, preserving first-seen order. A line that fails to + /// parse is logged and skipped rather than failing the whole load — a + /// single corrupt line must not block boot. + fn load_raw(dir: &Path) -> io::Result> { + let path = dir.join(TICKETS_FILE); + let file = match File::open(&path) { + Ok(f) => f, + Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(Vec::new()), + Err(e) => return Err(e), + }; + let mut order: Vec = Vec::new(); + let mut by_id: HashMap = HashMap::new(); + for (lineno, line) in BufReader::new(file).lines().enumerate() { + let line = line?; + if line.trim().is_empty() { + continue; + } + match serde_json::from_str::(&line) { + Ok(ticket) => { + if !by_id.contains_key(&ticket.event_id) { + order.push(ticket.event_id.clone()); + } + by_id.insert(ticket.event_id.clone(), ticket); + } + Err(e) => { + tracing::warn!( + line = lineno + 1, + error = %e, + "wake-ticket: skipping unparseable line" + ); + } + } + } + Ok(order + .into_iter() + .filter_map(|id| by_id.remove(&id)) + .collect()) + } + + /// Atomically rewrite `wake-tickets.jsonl` to contain exactly `tickets` + /// (tmp file + fsync + rename), `chmod 600`. + fn write_compacted(dir: &Path, tickets: &[Ticket]) -> io::Result<()> { + let tmp_path = dir.join(format!("{TICKETS_FILE}.tmp")); + let final_path = dir.join(TICKETS_FILE); + { + let mut tmp = File::create(&tmp_path)?; + for ticket in tickets { + let line = serde_json::to_string(ticket) + .map_err(|e| io::Error::other(format!("ticket serialize error: {e}")))?; + tmp.write_all(line.as_bytes())?; + tmp.write_all(b"\n")?; + } + tmp.sync_all()?; + } + chmod_600(&tmp_path)?; + fs::rename(&tmp_path, &final_path)?; + Ok(()) + } + + /// Append one ticket line and fsync. Updates the in-memory index. + fn append(&self, ticket: Ticket) -> io::Result<()> { + let line = serde_json::to_string(&ticket) + .map_err(|e| io::Error::other(format!("ticket serialize error: {e}")))?; + let mut inner = self.inner.lock().unwrap_or_else(|e| e.into_inner()); + inner.file.write_all(line.as_bytes())?; + inner.file.write_all(b"\n")?; + inner.file.sync_data()?; + // First append to a freshly-created file — fix permissions now that + // the inode exists (create_dir_all + append-open above may have + // created it with the process umask). + chmod_600(&self.dir.join(TICKETS_FILE))?; + inner.index.insert(ticket.event_id.clone(), ticket); + Ok(()) + } + + /// Tickets in `open` or `claimed` state — candidates for boot replay. + /// Order is unspecified; callers sort/dedupe as needed. + pub fn pending_for_replay(&self) -> Vec { + let inner = self.inner.lock().unwrap_or_else(|e| e.into_inner()); + inner + .index + .values() + .filter(|t| matches!(t.state, TicketState::Open | TicketState::Claimed)) + .cloned() + .collect() + } + + /// Write `open` for a newly-accepted mention. No-op (logged) if the + /// channel already has `MAX_PENDING_PER_CHANNEL` open+claimed tickets — + /// mirrors the in-memory queue's own per-channel depth cap. + #[allow(clippy::too_many_arguments)] + pub fn write_open( + &self, + event: &nostr::Event, + channel_id: Uuid, + pubkey: &str, + prompt_tag: &str, + seen_at: u64, + ) { + let event_id = event.id.to_hex(); + { + let inner = self.inner.lock().unwrap_or_else(|e| e.into_inner()); + let pending_for_channel = inner + .index + .values() + .filter(|t| { + t.channel_id == channel_id + && matches!(t.state, TicketState::Open | TicketState::Claimed) + }) + .count(); + if pending_for_channel >= MAX_PENDING_PER_CHANNEL { + tracing::warn!( + %channel_id, + cap = MAX_PENDING_PER_CHANNEL, + "wake-ticket: per-channel cap reached — not writing ticket \ + (in-memory queue enforces the same cap independently)" + ); + return; + } + } + let ticket = Ticket { + event_id: event_id.clone(), + channel_id, + created_at: event.created_at.as_secs(), + pubkey: pubkey.to_string(), + seen_at, + state: TicketState::Open, + event: event.clone(), + prompt_tag: prompt_tag.to_string(), + }; + if let Err(e) = self.append(ticket) { + tracing::warn!(event_id = %event_id, error = %e, "wake-ticket: failed to write open"); + } + } + + /// Transition tickets to `claimed` — a batch was handed to an agent. + pub fn mark_claimed(&self, event_ids: &[String]) { + self.transition(event_ids, TicketState::Claimed); + } + + /// Event ids currently `claimed` for `channel_id`. + /// + /// The completion path (`lib.rs::handle_prompt_result`) doesn't carry + /// the dispatched batch's event ids on the success path (`PromptResult` + /// intentionally clears `batch` on `PromptOutcome::Ok` — nothing to + /// requeue), so it looks up "what was claimed for this channel" here + /// instead of threading ids through the pool/queue result plumbing. + pub fn claimed_event_ids_for_channel(&self, channel_id: Uuid) -> Vec { + let inner = self.inner.lock().unwrap_or_else(|e| e.into_inner()); + inner + .index + .values() + .filter(|t| t.channel_id == channel_id && t.state == TicketState::Claimed) + .map(|t| t.event_id.clone()) + .collect() + } + + /// Transition tickets to `done` — `PromptOutcome::Ok` and nothing was + /// requeued for these events. Compacted away on the next boot. + pub fn mark_done(&self, event_ids: &[String]) { + self.transition(event_ids, TicketState::Done); + } + + /// Transition tickets to `drop` — dead-lettered, auth-failed, or the + /// channel was removed. The line survives on disk for audit; it is + /// never a replay candidate again. + pub fn mark_drop(&self, event_ids: &[String]) { + self.transition(event_ids, TicketState::Drop); + } + + /// Drop a replay candidate that failed re-validation (membership, author + /// gate, or rule match) without ever pushing it back into the queue. + pub fn drop_unvalidated(&self, event_id: &str) { + self.transition( + std::slice::from_ref(&event_id.to_string()), + TicketState::Drop, + ); + } + + fn transition(&self, event_ids: &[String], state: TicketState) { + for event_id in event_ids { + let existing = { + let inner = self.inner.lock().unwrap_or_else(|e| e.into_inner()); + inner.index.get(event_id).cloned() + }; + let Some(mut ticket) = existing else { + // No open/claimed/drop record — most commonly a non-mention + // event in the same batch (never had a ticket) or a replay + // survivor already compacted. Not an error. + continue; + }; + if ticket.state == state { + continue; // idempotent — avoid a redundant fsync'd line + } + ticket.state = state; + if let Err(e) = self.append(ticket) { + tracing::warn!( + event_id = %event_id, + ?state, + error = %e, + "wake-ticket: failed to write state transition" + ); + } + } + } +} + +#[cfg(unix)] +fn chmod_dir_700(dir: &Path) -> io::Result<()> { + use std::os::unix::fs::PermissionsExt; + fs::set_permissions(dir, fs::Permissions::from_mode(0o700)) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::time::{SystemTime, UNIX_EPOCH}; + + fn now() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_secs() + } + + fn signed_event(keys: &nostr::Keys, content: &str) -> nostr::Event { + nostr::EventBuilder::new(nostr::Kind::TextNote, content) + .sign_with_keys(keys) + .expect("sign") + } + + #[test] + fn open_write_and_replay_roundtrip() { + let tmp = tempfile::tempdir().unwrap(); + let store = WakeTicketStore::open(tmp.path()).unwrap(); + let keys = nostr::Keys::generate(); + let event = signed_event(&keys, "hello"); + let channel_id = Uuid::new_v4(); + + store.write_open( + &event, + channel_id, + &keys.public_key().to_hex(), + "@mention", + now(), + ); + + let pending = store.pending_for_replay(); + assert_eq!(pending.len(), 1); + assert_eq!(pending[0].event_id, event.id.to_hex()); + assert_eq!(pending[0].state, TicketState::Open); + assert_eq!(pending[0].event.id, event.id); + } + + #[test] + fn claimed_then_done_removes_from_replay_candidates() { + let tmp = tempfile::tempdir().unwrap(); + let store = WakeTicketStore::open(tmp.path()).unwrap(); + let keys = nostr::Keys::generate(); + let event = signed_event(&keys, "hi"); + let channel_id = Uuid::new_v4(); + let id = event.id.to_hex(); + + store.write_open( + &event, + channel_id, + &keys.public_key().to_hex(), + "@mention", + now(), + ); + store.mark_claimed(std::slice::from_ref(&id)); + assert_eq!(store.pending_for_replay()[0].state, TicketState::Claimed); + + store.mark_done(std::slice::from_ref(&id)); + assert!(store.pending_for_replay().is_empty()); + } + + #[test] + fn done_is_compacted_away_on_reopen() { + let tmp = tempfile::tempdir().unwrap(); + let keys = nostr::Keys::generate(); + let event = signed_event(&keys, "hi"); + let channel_id = Uuid::new_v4(); + let id = event.id.to_hex(); + { + let store = WakeTicketStore::open(tmp.path()).unwrap(); + store.write_open( + &event, + channel_id, + &keys.public_key().to_hex(), + "@mention", + now(), + ); + store.mark_done(std::slice::from_ref(&id)); + // Compaction is lazy (see `WakeTicketStore::open` docs) — `mark_done` + // only appends the `done` transition line; the file still carries + // both lines until the next `open()` rewrites it. In-memory state + // is already correct here (covered by + // `claimed_then_done_removes_from_replay_candidates`). + assert!(store.pending_for_replay().is_empty()); + } + + let store = WakeTicketStore::open(tmp.path()).unwrap(); + let contents = fs::read_to_string(tmp.path().join(TICKETS_FILE)).unwrap(); + assert!( + contents.is_empty(), + "done ticket should be compacted away on reopen, got: {contents}" + ); + assert!(store.pending_for_replay().is_empty()); + } + + #[test] + fn drop_survives_compaction_but_is_not_a_replay_candidate() { + let tmp = tempfile::tempdir().unwrap(); + let keys = nostr::Keys::generate(); + let event = signed_event(&keys, "hi"); + let channel_id = Uuid::new_v4(); + let id = event.id.to_hex(); + { + let store = WakeTicketStore::open(tmp.path()).unwrap(); + store.write_open( + &event, + channel_id, + &keys.public_key().to_hex(), + "@mention", + now(), + ); + store.mark_drop(std::slice::from_ref(&id)); + } + let contents = fs::read_to_string(tmp.path().join(TICKETS_FILE)).unwrap(); + assert!( + !contents.trim().is_empty(), + "drop ticket must survive compaction for audit" + ); + + let store = WakeTicketStore::open(tmp.path()).unwrap(); + assert!(store.pending_for_replay().is_empty()); + } + + #[test] + fn open_and_claimed_survive_process_restart() { + let tmp = tempfile::tempdir().unwrap(); + let keys = nostr::Keys::generate(); + let channel_id = Uuid::new_v4(); + let open_event = signed_event(&keys, "one"); + let claimed_event = signed_event(&keys, "two"); + { + let store = WakeTicketStore::open(tmp.path()).unwrap(); + store.write_open( + &open_event, + channel_id, + &keys.public_key().to_hex(), + "@mention", + now(), + ); + store.write_open( + &claimed_event, + channel_id, + &keys.public_key().to_hex(), + "@mention", + now(), + ); + store.mark_claimed(&[claimed_event.id.to_hex()]); + } + + let store = WakeTicketStore::open(tmp.path()).unwrap(); + let mut pending = store.pending_for_replay(); + pending.sort_by_key(|t| t.event_id.clone()); + assert_eq!(pending.len(), 2); + assert!(pending + .iter() + .any(|t| t.event_id == open_event.id.to_hex() && t.state == TicketState::Open)); + assert!(pending + .iter() + .any(|t| t.event_id == claimed_event.id.to_hex() && t.state == TicketState::Claimed)); + } + + #[test] + fn second_open_on_same_dir_fails_lock() { + let tmp = tempfile::tempdir().unwrap(); + let _store = WakeTicketStore::open(tmp.path()).unwrap(); + let second = WakeTicketStore::open(tmp.path()); + assert!(second.is_err(), "dual-run must fail, not block or succeed"); + } + + #[test] + fn lock_releases_on_drop_so_a_later_process_can_open() { + let tmp = tempfile::tempdir().unwrap(); + { + let _store = WakeTicketStore::open(tmp.path()).unwrap(); + } + let reopened = WakeTicketStore::open(tmp.path()); + assert!(reopened.is_ok(), "lock must release when the store drops"); + } + + #[test] + fn unparseable_line_is_skipped_not_fatal() { + let tmp = tempfile::tempdir().unwrap(); + fs::write( + tmp.path().join(TICKETS_FILE), + "not json\n{\"also\": \"not a ticket\"}\n", + ) + .unwrap(); + let store = WakeTicketStore::open(tmp.path()); + assert!(store.is_ok()); + assert!(store.unwrap().pending_for_replay().is_empty()); + } + + #[test] + fn last_write_wins_by_event_id_across_lines() { + let tmp = tempfile::tempdir().unwrap(); + let keys = nostr::Keys::generate(); + let event = signed_event(&keys, "hi"); + let channel_id = Uuid::new_v4(); + let id = event.id.to_hex(); + { + let store = WakeTicketStore::open(tmp.path()).unwrap(); + store.write_open( + &event, + channel_id, + &keys.public_key().to_hex(), + "@mention", + now(), + ); + store.mark_claimed(std::slice::from_ref(&id)); + // Simulate a crash right after claim — no done/drop line yet. + } + let store = WakeTicketStore::open(tmp.path()).unwrap(); + let pending = store.pending_for_replay(); + assert_eq!(pending.len(), 1); + assert_eq!(pending[0].state, TicketState::Claimed); + } + + #[test] + fn per_channel_cap_stops_writing_new_open_tickets() { + let tmp = tempfile::tempdir().unwrap(); + let store = WakeTicketStore::open(tmp.path()).unwrap(); + let keys = nostr::Keys::generate(); + let channel_id = Uuid::new_v4(); + for i in 0..MAX_PENDING_PER_CHANNEL { + let event = signed_event(&keys, &format!("msg {i}")); + store.write_open( + &event, + channel_id, + &keys.public_key().to_hex(), + "@mention", + now(), + ); + } + assert_eq!(store.pending_for_replay().len(), MAX_PENDING_PER_CHANNEL); + + let overflow = signed_event(&keys, "overflow"); + store.write_open( + &overflow, + channel_id, + &keys.public_key().to_hex(), + "@mention", + now(), + ); + assert_eq!( + store.pending_for_replay().len(), + MAX_PENDING_PER_CHANNEL, + "cap must hold — overflow ticket must not be written" + ); + } +} From 0ccf42c943263badfb104a85c90c1981cc850bec Mon Sep 17 00:00:00 2001 From: Robin Leonard Date: Sat, 15 Aug 2026 16:41:48 +0700 Subject: [PATCH 2/3] fix(buzz-acp): fix two wake-ticket bugs found in review (Oksana, 2026-08-15) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two real bugs from Oksana's review of PR #5940: 1. Boot replay treated any REST transport failure (query error, unresolved channel type) as a confirmed deny and permanently dropped the ticket. The bounce this feature exists to survive is often network-adjacent (Railway blip, DNS hiccup) — a transient failure at boot must not delete work a moment's retry could recover. `is_still_member` now returns `Option` (`None` = query failed, not "not a member"). Replay validation is extracted into a testable `validate_ticket_for_replay` -> `ReplayDecision::{Push,Drop,SkipThisBoot}`, resolving channel type directly (not through the live path's `is_dm_channel`, which intentionally fails closed to DM on an unresolved fetch — exactly the behavior that was turning a metadata blip into a false author-gate rejection for a normal channel). `Drop` now only fires on a confirmed deny; `SkipThisBoot` leaves the ticket untouched for the next boot. 2. `done` was gated on the channel's queue being empty. A completed batch (A) whose channel had more traffic queued behind it (B, still open) was never marked done — it stayed `claimed` indefinitely, so a later bounce replayed an already-finished turn a second time. This is the exact `01fa1107` double-delivery class the ticket store exists to prevent. Only one batch can be claimed/in-flight per channel at a time, so `claimed_event_ids_for_channel` already scopes correctly to *this* batch regardless of what else is queued — the fix is to stop gating on queue depth, not to thread ids through a different path. Adds `wake_ticket_wiring_tests`: the state-machine-driving tests through `handle_prompt_result` and the replay-validation helpers that the store-only unit tests couldn't reach, per Oksana's ask — Ok-with-more- queued marks exactly this batch done (the regression above), a requeued failure leaves the ticket claimed (mark_complete alone is not done), a genuine dead-letter still drops, and both `is_still_member` and `validate_ticket_for_replay` treat an unreachable relay as "unknown", not "deny". Signed-off-by: Robin Leonard --- crates/buzz-acp/src/lib.rs | 501 ++++++++++++++++++++++++++++++++--- crates/buzz-acp/src/queue.rs | 12 +- 2 files changed, 462 insertions(+), 51 deletions(-) diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index e9bc50bdb9f..8dcbaca0421 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -295,9 +295,19 @@ pub(crate) async fn is_dm_channel( /// Queries kind:39002 (NIP-29 group members) scoped to both this pubkey and /// the one channel — the same shape as `HarnessRelay::discover_channels`'s /// Step 1, narrowed so replay doesn't refetch the whole membership set per -/// ticket. Fail-closed: a query error means "not a member" rather than -/// risking a stale mention replaying into a channel this unit lost access to. -async fn is_still_member(channel_id: Uuid, pubkey_hex: &str, rest: &relay::RestClient) -> bool { +/// ticket. +/// +/// Returns `None` on a transport/query failure rather than fail-closing to +/// "not a member". `drop` on this path is meant for a *confirmed* deny — the +/// bounce a wake ticket exists to survive is often network-adjacent (Railway +/// blip, DNS hiccup), and a transient REST failure at boot must not +/// permanently delete work that a moment's retry (next boot) could recover. +/// Callers treat `None` as "skip this boot, leave the ticket as-is". +async fn is_still_member( + channel_id: Uuid, + pubkey_hex: &str, + rest: &relay::RestClient, +) -> Option { use nostr::{Alphabet, SingleLetterTag}; let p_tag = SingleLetterTag::lowercase(Alphabet::P); @@ -310,18 +320,82 @@ async fn is_still_member(channel_id: Uuid, pubkey_hex: &str, rest: &relay::RestC .custom_tags(d_tag, [channel_id.to_string()]); match rest.query(&[filter]).await { - Ok(v) => v.as_array().is_some_and(|arr| !arr.is_empty()), + Ok(v) => Some(v.as_array().is_some_and(|arr| !arr.is_empty())), Err(e) => { tracing::warn!( channel_id = %channel_id, error = %e, - "wake-ticket replay: membership check failed — treating as not-a-member (fail closed)" + "wake-ticket replay: membership check failed — skipping this boot (not a confirmed deny)" ); - false + None } } } +/// Outcome of re-validating one wake ticket at boot, before it's pushed back +/// into the queue. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ReplayDecision { + /// Confirmed member + confirmed author-allowed: push into the queue. + Push, + /// Confirmed deny (definite not-a-member, or the author gate rejects a + /// resolved, non-uncertain channel type): drop, never replay again. + Drop, + /// A dependency (channel-type resolution or the membership query) + /// couldn't be confirmed this boot — leave the ticket exactly as it was + /// (`open`/`claimed`) and simply don't push it. It's retried in full on + /// the next boot, same as if this boot never ran. + SkipThisBoot, +} + +/// Re-validate one wake ticket at boot: still a channel member, and the +/// author gate still admits this event's author. See `ReplayDecision` for +/// what each outcome means and `is_still_member` for why a transport failure +/// is `SkipThisBoot`, not `Drop`. +/// +/// Channel-type resolution failure is the same story as the membership +/// query: `ChannelInfoResolver::resolve` returning `None` means "couldn't +/// fetch metadata this boot" (see its own fail-closed-to-DM behavior, which +/// is intentional for the *live* author gate — but here, tying that +/// unresolved state to `is_dm=true` would run `author_allowed` under DM +/// rules against a channel that probably isn't one, and drop a normal +/// mention because of a metadata fetch blip). Resolve it explicitly instead +/// of going through `is_dm_channel`, and treat "unresolved" as uncertain. +async fn validate_ticket_for_replay( + ticket: &wake_ticket::Ticket, + respond_to: &RespondTo, + respond_to_allowlist: &HashSet, + pubkey_hex: &str, + owner_cache: &OwnerCache, + channel_info: &pool::ChannelInfoResolver, + rest: &relay::RestClient, +) -> ReplayDecision { + let is_dm = match channel_info.resolve(ticket.channel_id).await { + Some(info) => info.channel_type == "dm", + None => return ReplayDecision::SkipThisBoot, + }; + + let author = ticket.event.pubkey.to_hex(); + let allowed = author_allowed( + respond_to, + respond_to_allowlist, + &author, + is_dm, + owner_cache, + rest, + ) + .await; + if !allowed { + return ReplayDecision::Drop; + } + + match is_still_member(ticket.channel_id, pubkey_hex, rest).await { + Some(true) => ReplayDecision::Push, + Some(false) => ReplayDecision::Drop, + None => ReplayDecision::SkipThisBoot, + } +} + /// Query an author's kind:0 profile and check if their NIP-OA auth tag /// proves the same owner as us. async fn check_sibling_via_profile( @@ -2065,44 +2139,55 @@ async fn tokio_main() -> Result<()> { let mut replayed = 0usize; let mut dropped = 0usize; + let mut skipped = 0usize; for ticket in pending { - let is_dm = is_dm_channel(ticket.channel_id, &replay_channel_info).await; - let author = ticket.event.pubkey.to_hex(); - let allowed = author_allowed( + match validate_ticket_for_replay( + &ticket, &config.respond_to, &config.respond_to_allowlist, - &author, - is_dm, + &pubkey_hex, &replay_owner_cache, + &replay_channel_info, &replay_rest_client, ) - .await; - let still_member = - is_still_member(ticket.channel_id, &pubkey_hex, &replay_rest_client).await; - - if !allowed || !still_member { - tracing::info!( - event_id = %ticket.event_id, - channel_id = %ticket.channel_id, - allowed, - still_member, - "wake-ticket replay: re-validation failed — dropping" - ); - store.drop_unvalidated(&ticket.event_id); - dropped += 1; - continue; + .await + { + ReplayDecision::Drop => { + tracing::info!( + event_id = %ticket.event_id, + channel_id = %ticket.channel_id, + "wake-ticket replay: confirmed deny — dropping" + ); + store.drop_unvalidated(&ticket.event_id); + dropped += 1; + } + ReplayDecision::SkipThisBoot => { + tracing::info!( + event_id = %ticket.event_id, + channel_id = %ticket.channel_id, + "wake-ticket replay: re-validation inconclusive (transport/query \ + failure) — leaving ticket as-is, retrying next boot" + ); + skipped += 1; + } + ReplayDecision::Push => { + queue.push(QueuedEvent { + channel_id: ticket.channel_id, + event: ticket.event.clone(), + received_at: std::time::Instant::now(), + prompt_tag: ticket.prompt_tag.clone(), + }); + wake_ticket_replay_seen_ids.push(ticket.event_id.clone()); + replayed += 1; + } } - - queue.push(QueuedEvent { - channel_id: ticket.channel_id, - event: ticket.event.clone(), - received_at: std::time::Instant::now(), - prompt_tag: ticket.prompt_tag.clone(), - }); - wake_ticket_replay_seen_ids.push(ticket.event_id.clone()); - replayed += 1; } - tracing::info!(replayed, dropped, "wake-ticket boot replay complete"); + tracing::info!( + replayed, + dropped, + skipped, + "wake-ticket boot replay complete" + ); } } @@ -4191,14 +4276,24 @@ fn handle_prompt_result( // than threaded from the dispatched batch: `result.batch` is // `None` on the `Ok` path (nothing to requeue) and always `None` // in `DedupMode::Drop`, so it cannot carry the completed event - // ids here. + // ids here. This is still exactly *this* batch's ids, not + // "every claimed ticket on the channel ever": only one batch can + // be claimed/in-flight per channel at a time (`flush_next` only + // picks non-in-flight channels), so whatever is `Claimed` for + // `ch` right now is precisely what this turn was dispatched with. + // + // Marking `done` does NOT wait for the channel's queue to drain. + // A later event (B) arriving mid-turn and still being queued when + // this turn (A) completes does not mean A is unfinished — it + // means B is separate, still-`open` work. Gating `done` on an + // empty queue left A `claimed` indefinitely whenever traffic kept + // arriving, so a bounce mid-queue replayed an already-completed + // turn a second time — the exact `01fa1107` class this ticket + // exists to prevent (caught in review, not shipped). if let Some(store) = wake_tickets { if should_drop_tickets { store.mark_drop(&store.claimed_event_ids_for_channel(*ch)); - } else if matches!(result.outcome, PromptOutcome::Ok(_)) - && queue.queued_event_count(ch) == 0 - && !queue.is_retry_throttled(ch) - { + } else if matches!(result.outcome, PromptOutcome::Ok(_)) { store.mark_done(&store.claimed_event_ids_for_channel(*ch)); } // Cancelled/CancelDrainTimeout/requeued-with-budget-remaining/ @@ -8652,6 +8747,330 @@ mod error_outcome_emission_tests { } } +#[cfg(test)] +mod wake_ticket_wiring_tests { + //! Tests the wake-ticket wiring inside `handle_prompt_result` and the + //! boot-replay validation helpers — the state-machine driving logic that + //! `wake_ticket::tests` (store-only) can't reach. Added per Oksana's + //! 2026-08-15 review of PR #5940 (block/buzz), which found two real bugs + //! here: `done` gated on empty queue depth (left an already-completed + //! batch `claimed` forever whenever more traffic kept arriving — a + //! guaranteed duplicate replay on the next bounce), and replay + //! re-validation treating a REST transport failure as a confirmed deny + //! (permanently deleting work over a network blip). See + //! `PLANS/WAKE_TICKET_SPEC.md` §Gate. + + use super::*; + use crate::pool::{AgentPool, PromptOutcome, PromptResult, PromptSource, TimeoutKind}; + use crate::queue::{BatchEvent, FlushBatch}; + use crate::wake_ticket::{TicketState, WakeTicketStore}; + use nostr::{EventBuilder, Keys, Kind}; + use std::collections::HashSet; + + fn test_config() -> Config { + Config { + keys: nostr::Keys::generate(), + relay_url: "ws://localhost:3000".into(), + agent_command: "true".into(), + agent_args: vec![], + mcp_command: "test-mcp-server".into(), + idle_timeout_secs: config::DEFAULT_IDLE_TIMEOUT_SECS, + max_turn_duration_secs: config::DEFAULT_MAX_TURN_DURATION_SECS, + agents: 1, + heartbeat_interval_secs: 0, + turn_liveness_secs: 10, + heartbeat_prompt: None, + system_prompt: None, + team_instructions: None, + initial_message: None, + subscribe_mode: config::SubscribeMode::All, + dedup_mode: config::DedupMode::Queue, + multiple_event_handling: config::MultipleEventHandling::Queue, + ignore_self: true, + kinds_override: None, + channels_override: None, + no_mention_filter: false, + config_path: std::path::PathBuf::from("./buzz-acp.toml"), + context_message_limit: 12, + max_turns_per_session: 0, + presence_enabled: true, + typing_enabled: true, + memory_enabled: false, + model: None, + session_title: None, + permission_mode: config::PermissionMode::BypassPermissions, + respond_to: config::RespondTo::Anyone, + respond_to_allowlist: HashSet::new(), + allowed_respond_to: vec![], + persona_env_vars: vec![], + has_generated_codex_config: false, + relay_observer: false, + exit_after_inactivity_secs: 0, + lazy_pool: false, + idle_pool_sleep_secs: 0, + agent_owner: None, + no_base_prompt: false, + base_prompt_content: None, + wake_ticket_dir: None, + } + } + + async fn dummy_agent(index: usize) -> OwnedAgent { + OwnedAgent { + index, + acp: AcpClient::spawn("cat", &[], &[], false) + .await + .expect("spawn cat as inert agent"), + state: Default::default(), + model_capabilities: None, + desired_model: None, + model_overridden: false, + agent_name: "unknown".into(), + goose_system_prompt_supported: None, + protocol_version: 1, + } + } + + fn signed_event(content: &str) -> nostr::Event { + EventBuilder::new(Kind::Custom(9), content) + .sign_with_keys(&Keys::generate()) + .unwrap() + } + + /// Drive `outcome` (and optionally a dispatched `batch`) through + /// `handle_prompt_result` for `channel_id`, with `store` wired in as the + /// wake-ticket store. + async fn run_result( + store: &WakeTicketStore, + channel_id: Uuid, + outcome: PromptOutcome, + batch: Option, + ) { + let agent = dummy_agent(0).await; + let mut pool = AgentPool::from_slots(vec![None]); + let task_id = pool.join_set.spawn(async {}).id(); + pool.task_map_mut().insert( + task_id, + crate::pool::TaskMeta { + agent_index: 0, + channel_id: Some(channel_id), + turn_id: "test-turn-id".into(), + recoverable_batch: None, + control_tx: None, + steer_tx: None, + successful_steer_deliveries: HashSet::new(), + }, + ); + let mut queue = EventQueue::new(config::DedupMode::Queue); + let config = test_config(); + let mut heartbeat_in_flight = false; + let removed_channels = HashSet::new(); + let mut crash_history = vec![SlotCircuit { + crash_times: Vec::new(), + open_until: None, + respawn_in_flight: false, + }]; + let (respawn_tx, _respawn_rx) = mpsc::channel(8); + let mut respawn_tasks = tokio::task::JoinSet::new(); + let result = PromptResult { + agent, + source: PromptSource::Channel(channel_id), + turn_id: "test-turn-id".into(), + outcome, + batch, + }; + handle_prompt_result( + &mut pool, + &mut queue, + &config, + result, + &mut heartbeat_in_flight, + &removed_channels, + &mut crash_history, + &respawn_tx, + &mut respawn_tasks, + None, + None, + Some(store), + ); + } + + /// Oksana's repro, verbatim: "A in flight, B arrives, A completes Ok, + /// bounce. A is still `claimed`. Replay. Second turn." Gating `done` on + /// an empty queue left A `claimed` forever whenever traffic kept + /// arriving — the fix marks exactly the completed batch's ids `done`, + /// regardless of what else is queued. + #[tokio::test] + async fn ok_with_more_events_queued_still_marks_this_batch_done() { + let tmp = tempfile::tempdir().unwrap(); + let store = WakeTicketStore::open(tmp.path()).unwrap(); + let channel_id = Uuid::new_v4(); + let pubkey = Keys::generate().public_key().to_hex(); + + let event_a = signed_event("a"); + let event_b = signed_event("b"); + store.write_open(&event_a, channel_id, &pubkey, "@mention", 1); + store.write_open(&event_b, channel_id, &pubkey, "@mention", 2); + // A was dispatched and claimed; B arrived afterward and is still open + // — only one batch can be claimed/in-flight per channel at a time. + store.mark_claimed(&[event_a.id.to_hex()]); + + run_result( + &store, + channel_id, + PromptOutcome::Ok(crate::acp::StopReason::EndTurn), + None, + ) + .await; + + let tickets = store.pending_for_replay(); + assert!( + !tickets.iter().any(|t| t.event_id == event_a.id.to_hex()), + "A must be done (compacted away), not left claimed just because B is still queued" + ); + let b = tickets + .iter() + .find(|t| t.event_id == event_b.id.to_hex()) + .expect("B must still be a replay candidate — it was never claimed or completed"); + assert_eq!( + b.state, + TicketState::Open, + "B is separate, unrelated work — completing A must not touch it" + ); + } + + /// `mark_complete` runs on every outcome, including a requeued failure — + /// it is a lock release, not a completion signal. Only `PromptOutcome::Ok` + /// may mark `done`. + #[tokio::test] + async fn requeued_failure_leaves_ticket_claimed_not_done() { + let tmp = tempfile::tempdir().unwrap(); + let store = WakeTicketStore::open(tmp.path()).unwrap(); + let channel_id = Uuid::new_v4(); + let pubkey = Keys::generate().public_key().to_hex(); + let event = signed_event("hi"); + store.write_open(&event, channel_id, &pubkey, "@mention", 1); + store.mark_claimed(&[event.id.to_hex()]); + + let batch = FlushBatch { + channel_id, + events: vec![BatchEvent { + event: event.clone(), + prompt_tag: "@mention".into(), + received_at: std::time::Instant::now(), + }], + cancelled_events: vec![], + cancel_reason: None, + }; + // AgentExited on a fresh channel: queue.requeue() succeeds (well + // under MAX_RETRIES) rather than dead-lettering. mark_complete still + // runs unconditionally — that alone must not mark `done`. + run_result(&store, channel_id, PromptOutcome::AgentExited, Some(batch)).await; + + let tickets = store.pending_for_replay(); + let ticket = tickets + .iter() + .find(|t| t.event_id == event.id.to_hex()) + .expect("must still be a replay candidate — mark_complete alone is not done"); + assert_eq!(ticket.state, TicketState::Claimed); + } + + /// Sanity check that `drop` still fires for a genuine dead-letter now + /// that `done` no longer gates on queue depth. + #[tokio::test] + async fn dead_lettered_failure_drops_ticket() { + let tmp = tempfile::tempdir().unwrap(); + let store = WakeTicketStore::open(tmp.path()).unwrap(); + let channel_id = Uuid::new_v4(); + let pubkey = Keys::generate().public_key().to_hex(); + let event = signed_event("hi"); + store.write_open(&event, channel_id, &pubkey, "@mention", 1); + store.mark_claimed(&[event.id.to_hex()]); + + let batch = FlushBatch { + channel_id, + events: vec![BatchEvent { + event: event.clone(), + prompt_tag: "@mention".into(), + received_at: std::time::Instant::now(), + }], + cancelled_events: vec![], + cancel_reason: None, + }; + run_result( + &store, + channel_id, + PromptOutcome::Timeout(TimeoutKind::Hard { + recently_active: false, + }), + Some(batch), + ) + .await; + + assert!( + store.pending_for_replay().is_empty(), + "dead-lettered ticket must not be a replay candidate" + ); + } + + /// A `RestClient` that always fails to connect (nothing listens on this + /// reserved port) — the transport-failure case, not a confirmed deny. + fn unreachable_rest_client() -> relay::RestClient { + relay::RestClient { + http: reqwest::Client::new(), + base_url: "http://127.0.0.1:1".into(), + keys: Keys::generate(), + auth_tag_json: None, + } + } + + #[tokio::test] + async fn is_still_member_query_failure_is_unknown_not_denied() { + let result = is_still_member(Uuid::new_v4(), "deadbeef", &unreachable_rest_client()).await; + assert_eq!( + result, None, + "a transport failure must not be reported as a confirmed non-member" + ); + } + + #[tokio::test] + async fn validate_ticket_for_replay_skips_on_unreachable_relay() { + let rest = unreachable_rest_client(); + let channel_id = Uuid::new_v4(); + let channel_info = pool::ChannelInfoResolver::new(HashMap::new(), rest.clone()); + let owner_cache = OwnerCache::new(None); + let event = signed_event("hi"); + let ticket = wake_ticket::Ticket { + event_id: event.id.to_hex(), + channel_id, + created_at: event.created_at.as_secs(), + pubkey: event.pubkey.to_hex(), + seen_at: 1, + state: TicketState::Open, + event, + prompt_tag: "@mention".into(), + }; + + let decision = validate_ticket_for_replay( + &ticket, + &RespondTo::Anyone, + &HashSet::new(), + "deadbeef", + &owner_cache, + &channel_info, + &rest, + ) + .await; + + assert_eq!( + decision, + ReplayDecision::SkipThisBoot, + "an unreachable relay must not be treated as a confirmed deny — the ticket must \ + survive to try again next boot, not be permanently deleted over a network blip" + ); + } +} + #[cfg(test)] mod observer_payload_trim_tests { use super::*; diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index e13874dcad1..a81a6483864 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -628,16 +628,8 @@ impl EventQueue { self.queues.len() } - /// Whether `channel_id` is currently under an active `retry_after` - /// backoff throttle (a prior attempt was requeued and hasn't reached - /// its next-try deadline yet). - pub fn is_retry_throttled(&self, channel_id: &Uuid) -> bool { - self.retry_after - .get(channel_id) - .is_some_and(|&deadline| deadline > Instant::now()) - } - - /// Number of queued (not-yet-dispatched) events for a specific channel. + /// Number of queued (not-yet-dispatched) events for a specific channel. Test-only. + #[cfg(test)] pub fn queued_event_count(&self, channel_id: &Uuid) -> usize { self.queues.get(channel_id).map_or(0, |q| q.len()) } From bbeb16963cfd32d47ff3967e9e0979fbc9207be4 Mon Sep 17 00:00:00 2001 From: Robin Leonard Date: Sat, 15 Aug 2026 16:50:32 +0700 Subject: [PATCH 3/3] fix(buzz-acp): close sibling-lookup fail-closed-to-drop residual (Oksana, 2026-08-15) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Second pass: the two named blockers were fixed, but the residual I flagged (check_sibling_via_profile fails closed to false on a query timeout/error) turned out to be blocker 1 on the path this team actually uses — Zar's traffic is other agents, not Robin, so most replayed tickets need a fresh sibling lookup, and a Railway blink landing there was deleting them exactly like the membership/channel-type checks already fixed. Per instruction: do not fork the live author_allowed. Added a parallel replay-only decision tree instead: - extract_verified_sibling: the NIP-OA tag-parsing/verification logic pulled out of check_sibling_via_profile as a pure function, shared by both the live and replay paths so there's exactly one implementation to keep correct. - check_sibling_via_profile_for_replay / is_owner_or_sibling_for_replay / author_allowed_for_replay: tri-state mirrors of the live functions. A query timeout/error is None (unknown, not confirmed). A malformed pubkey, no owner configured, a completed query with no valid tag, or a completed query with a tag that fails signature verification are all Some(false) — confirmed, not a transport problem, no reason to wait for a retry. The live functions are untouched. - validate_ticket_for_replay now calls author_allowed_for_replay and maps None to SkipThisBoot. Cached sibling/owner results skip the network entirely on both paths, so this doesn't cost every replayed ticket a fresh profile query. Tests: unreachable relay on the sibling-lookup path skips both is_owner_or_sibling_for_replay and author_allowed_for_replay (not deny); cached results resolve without touching the network. Signed-off-by: Robin Leonard --- crates/buzz-acp/src/lib.rs | 194 ++++++++++++++++++++++++++++++++++++- 1 file changed, 190 insertions(+), 4 deletions(-) diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 8dcbaca0421..cd9cea065f5 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -376,7 +376,7 @@ async fn validate_ticket_for_replay( }; let author = ticket.event.pubkey.to_hex(); - let allowed = author_allowed( + match author_allowed_for_replay( respond_to, respond_to_allowlist, &author, @@ -384,9 +384,11 @@ async fn validate_ticket_for_replay( owner_cache, rest, ) - .await; - if !allowed { - return ReplayDecision::Drop; + .await + { + Some(true) => {} + Some(false) => return ReplayDecision::Drop, + None => return ReplayDecision::SkipThisBoot, } match is_still_member(ticket.channel_id, pubkey_hex, rest).await { @@ -418,6 +420,18 @@ async fn check_sibling_via_profile( _ => return false, // timeout or error — fail closed }; + extract_verified_sibling(&resp, author, expected_owner) +} + +/// Extract and cryptographically verify a NIP-OA sibling attestation from an +/// already-fetched kind:0 profile query response. Pure — no network, no +/// error-vs-empty distinction to make. Shared by [`check_sibling_via_profile`] +/// (live path, its caller fails closed to `false` on any non-match) and +/// [`check_sibling_via_profile_for_replay`] (wake-ticket boot replay, +/// tri-state) so the tag-parsing / signature-verification logic has exactly +/// one implementation — the two callers differ only in how they got `resp` +/// and what an *absence* of a valid tag means for their caller. +fn extract_verified_sibling(resp: &serde_json::Value, author: &str, expected_owner: &str) -> bool { // Look for an "auth" tag in the profile event. let events = match resp.as_array() { Some(arr) => arr, @@ -471,6 +485,111 @@ async fn check_sibling_via_profile( false } +/// Replay-only counterpart to [`check_sibling_via_profile`]: same query, +/// same NIP-OA verification via [`extract_verified_sibling`], but +/// distinguishes a transport/timeout failure (`None` — couldn't check, not a +/// confirmed answer) from a completed query that simply found no valid +/// sibling attestation (`Some(false)` — a real, confirmed no). +/// +/// Per `PLANS/WAKE_TICKET_SPEC.md` §Gate (Oksana, 2026-08-15, second pass): +/// this is the path Zar's actual traffic hits — sibling agents, not the +/// owner directly — so a Railway blink landing on *this* query was deleting +/// tickets just as surely as the membership/channel-type checks. The live +/// [`check_sibling_via_profile`] intentionally keeps failing closed to +/// `false` for the live turn-firing decision; that tradeoff is untouched. +async fn check_sibling_via_profile_for_replay( + author: &str, + expected_owner: &str, + rest_client: &relay::RestClient, +) -> Option { + let author_pk = match nostr::PublicKey::from_hex(author) { + Ok(pk) => pk, + // Malformed pubkey isn't a transport problem — it can never resolve + // to a valid sibling attestation no matter how many times we ask. + Err(_) => return Some(false), + }; + let filter = nostr::Filter::new() + .kind(nostr::Kind::Metadata) + .author(author_pk) + .limit(1); + + let resp = match tokio::time::timeout(Duration::from_millis(2000), rest_client.query(&[filter])) + .await + { + Ok(Ok(v)) => v, + Ok(Err(_)) | Err(_) => return None, // timeout or query error — unknown, not confirmed + }; + + Some(extract_verified_sibling(&resp, author, expected_owner)) +} + +/// Replay-only counterpart to [`is_owner_or_sibling`]. See +/// [`check_sibling_via_profile_for_replay`] for why this distinguishes +/// "couldn't check" from "confirmed no". A confirmed result (owner, cache +/// hit, or a completed profile query) is cached the same way the live path +/// caches it — `None` results are never cached, so the next ticket (or the +/// next boot) gets a fresh attempt instead of a pinned unknown. +async fn is_owner_or_sibling_for_replay( + author: &str, + owner_cache: &OwnerCache, + rest_client: &relay::RestClient, +) -> Option { + let my_owner = match owner_cache.get() { + Some(o) => o, + // No owner configured is a confirmed, static fact — not a transport + // problem — so this fails closed exactly like the live path. + None => return Some(false), + }; + + if author == my_owner { + return Some(true); + } + + if let Some(cached) = owner_cache.is_known_sibling(author) { + return Some(cached); + } + + let is_sibling = check_sibling_via_profile_for_replay(author, my_owner, rest_client).await; + if let Some(confirmed) = is_sibling { + owner_cache.cache_sibling(author.to_string(), confirmed); + } + is_sibling +} + +/// Replay-only counterpart to [`author_allowed`] — same decision tree +/// (owner ∪ allowlist ∪ siblings, DM hardening), but every network-backed +/// leaf can return "unknown" instead of being forced to a boolean. See +/// [`ReplayDecision`] for how callers use `None` here. +async fn author_allowed_for_replay( + respond_to: &RespondTo, + allowlist: &HashSet, + author: &str, + is_dm: bool, + owner_cache: &OwnerCache, + rest_client: &relay::RestClient, +) -> Option { + if is_dm { + return match respond_to { + RespondTo::Nobody => Some(false), + _ => is_owner_or_sibling_for_replay(author, owner_cache, rest_client).await, + }; + } + match respond_to { + RespondTo::Anyone => Some(true), + RespondTo::Nobody => Some(false), + RespondTo::OwnerOnly => { + is_owner_or_sibling_for_replay(author, owner_cache, rest_client).await + } + RespondTo::Allowlist => { + if allowlist.contains(author) { + Some(true) + } else { + is_owner_or_sibling_for_replay(author, owner_cache, rest_client).await + } + } + } +} + /// Observer frames are published at a global rate of AT MOST ONE relay frame /// per tick — not one per channel, and not one per drain. Everything that /// accumulates between ticks waits in [`ObserverPublishQueue`] as events and @@ -9033,6 +9152,73 @@ mod wake_ticket_wiring_tests { ); } + /// Oksana's second-pass finding: Zar's actual traffic is other agents + /// (siblings), not the owner directly, so a blink on the sibling + /// profile-lookup query — not just the membership/channel-type checks — + /// was silently deleting tickets. `is_owner_or_sibling_for_replay` must + /// distinguish "couldn't check" from "confirmed not a sibling", the same + /// way `is_still_member` already does for membership. + #[tokio::test] + async fn owner_only_author_skips_on_unreachable_relay_not_denied() { + // Must be syntactically valid hex — a malformed pubkey is a + // *confirmed* non-sibling (nothing to retry), not a transport + // failure, and would short-circuit before ever reaching the network. + let owner = Keys::generate().public_key().to_hex(); + let other_agent = Keys::generate().public_key().to_hex(); + let owner_cache = OwnerCache::new(Some(owner)); + let result = + is_owner_or_sibling_for_replay(&other_agent, &owner_cache, &unreachable_rest_client()) + .await; + assert_eq!( + result, None, + "a transport failure on the sibling-profile query must not be reported as a \ + confirmed non-sibling — this is the exact path Zar's real traffic hits" + ); + + let decision = author_allowed_for_replay( + &RespondTo::OwnerOnly, + &HashSet::new(), + &other_agent, + false, + &owner_cache, + &unreachable_rest_client(), + ) + .await; + assert_eq!( + decision, None, + "author_allowed_for_replay must propagate the sibling-lookup uncertainty, not \ + collapse it to a deny" + ); + } + + /// Cached siblinghood never needs the network — pinning this documents + /// why the fix above doesn't cost every replayed ticket a fresh query. + #[tokio::test] + async fn cached_sibling_result_is_confirmed_without_network() { + let owner_cache = OwnerCache::new(Some("owner-pubkey".into())); + owner_cache.cache_sibling("known-sibling".into(), true); + owner_cache.cache_sibling("known-stranger".into(), false); + + assert_eq!( + is_owner_or_sibling_for_replay( + "known-sibling", + &owner_cache, + &unreachable_rest_client() + ) + .await, + Some(true) + ); + assert_eq!( + is_owner_or_sibling_for_replay( + "known-stranger", + &owner_cache, + &unreachable_rest_client() + ) + .await, + Some(false) + ); + } + #[tokio::test] async fn validate_ticket_for_replay_skips_on_unreachable_relay() { let rest = unreachable_rest_client();