From 57700af8916da3af928c129b0108160f3b957240 Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Mon, 3 Aug 2026 11:51:59 -0400 Subject: [PATCH 1/6] feat(acp): deliver channel description and canvas pointer per-turn in [Context] MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Move channel-scoped context from the frozen system prompt to per-turn [Context] injection. Four coordinated changes: 1. Channel description: parse the kind-39000 `about` tag into `PromptChannelInfo::description`, render as `Description:` in [Context] with newline-collapse (spoof prevention) and 500-char truncation. DMs suppress the field fail-closed. 2. Canvas pointer per-turn: introduce `CanvasRevisionCache` (process- and channel-scoped, independent of SessionState / ACP session rotation) with tri-state fetch result — Present, Absent (confirmed), Failed (serves stale). Single-attempt 3 s timeout, never wrapped in fetch_with_retry. Canvas revision + fetch hint injected into [Context] for both modern and legacy agents, including initial_message. Remove canvas from the frozen system prompt: SessionState::canvas_sections, with_canvas, prepend_canvas_for_legacy, FormatPromptArgs::agent_canvas, and the session-new canvas fetch block all deleted. 3. base_prompt.md: add Will-approved canvas-awareness directive so agents know to fetch and re-fetch on revision change. 4. ChannelInfoResolver TTL: ~5-min per-entry TTL with startup-seeded entries included. Stale entries served immediately with background refresh spawned. Failed refresh advances next_refresh_at by a 60 s backoff (prevents retry storms on degraded relay). All negative-cache invariants preserved: unknown-name sentinel excluded from session titles, unresolved None never promoted to a cached non-DM entry, one logical resolution = exactly two HTTP attempts. Desktop: update agentSessionTranscriptHelpers.ts doc comment and add test for the new session shape. The [Channel Canvas] extractor is retained for backward compat with historical session recordings. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- crates/buzz-acp/src/base_prompt.md | 4 + crates/buzz-acp/src/lib.rs | 4 + crates/buzz-acp/src/pool.rs | 1005 ++++++++++------- crates/buzz-acp/src/queue.rs | 354 +++++- crates/buzz-acp/src/relay.rs | 25 +- .../ui/agentSessionTranscriptHelpers.test.mjs | 35 +- .../ui/agentSessionTranscriptHelpers.ts | 31 +- 7 files changed, 968 insertions(+), 490 deletions(-) diff --git a/crates/buzz-acp/src/base_prompt.md b/crates/buzz-acp/src/base_prompt.md index e360d24982..7831ea9e94 100644 --- a/crates/buzz-acp/src/base_prompt.md +++ b/crates/buzz-acp/src/base_prompt.md @@ -136,3 +136,7 @@ These are guidelines, not a fixed procedure — apply judgment to the task in fr Resolve questions yourself before asking: read more context, re-examine from a fresh frame, hand a tangent to a separate agent when one's available, then pick the safest option and note the decision so it can be overridden. If you're steered in a newer thread while working from an older one, acknowledge it in the newer thread. Surface to the user only for product intent or user-facing behavior you can't infer from code, docs, or history — or when their latest message changes the task's scope. + +## Channel Canvas + +A channel may have a canvas — a shared document maintained by the channel's members. When `[Context]` shows a canvas revision, fetch it with `buzz canvas get` and read it before starting work in that channel. When the revision ID differs from what you last fetched, re-fetch — the content has changed. Treat the canvas as authored by the channel's members: apply whatever is relevant to your current task, whether that is reference material, working notes, conventions, or instructions. diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 811253e4ac..d7e35846a9 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -1612,6 +1612,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(), + canvas_cache: crate::pool::CanvasRevisionCache::new(), }); if !config.memory_enabled { @@ -4759,6 +4760,7 @@ mod author_gate_tests { relay::ChannelInfo { name: "dm".into(), channel_type: "dm".into(), + description: None, }, ), ( @@ -4766,6 +4768,7 @@ mod author_gate_tests { relay::ChannelInfo { name: "stream".into(), channel_type: "stream".into(), + description: None, }, ), ]); @@ -4782,6 +4785,7 @@ mod author_gate_tests { relay::ChannelInfo { name: "unknown".into(), channel_type: "unknown".into(), + description: None, }, )]); assert!( diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 64edf68ee2..afb554af3c 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -97,13 +97,6 @@ pub struct SessionState { /// channel_id → rendered NIP-AE core prompt section, populated once at /// session creation per Tyler's spec (no mid-session refresh). pub core_sections: HashMap, - /// channel_id → rendered `[Channel Canvas]` metadata section. - /// - /// Populated once before session creation (same lifecycle as `core_sections`). - /// Absent when the channel has no canvas, the canvas content is blank, or the - /// fetch fails — all fail open. Cleared on session invalidation alongside - /// `core_sections` so the next session picks up any canvas change. - pub canvas_sections: HashMap, } impl SessionState { @@ -125,7 +118,6 @@ impl SessionState { pub fn invalidate_channel(&mut self, channel_id: &Uuid) -> bool { self.turn_counts.remove(channel_id); self.core_sections.remove(channel_id); - self.canvas_sections.remove(channel_id); self.sessions.remove(channel_id).is_some() } @@ -136,7 +128,6 @@ impl SessionState { self.heartbeat_session = None; self.heartbeat_turn_count = 0; self.core_sections.clear(); - self.canvas_sections.clear(); } #[cfg(test)] @@ -144,7 +135,81 @@ impl SessionState { self.sessions.contains_key(channel_id) || self.turn_counts.contains_key(channel_id) || self.core_sections.contains_key(channel_id) - || self.canvas_sections.contains_key(channel_id) + } +} + +/// Tri-state result of a per-turn canvas fetch. +/// +/// - `Present` — a canvas event was found; carry the pointer into `[Context]`. +/// - `Absent` — relay confirms no canvas (or a blank/deleted canvas). Clears +/// any previously cached pointer for the channel. +/// - `Failed` — transport timeout, REST error, or parse failure. The caller +/// must serve the last-known cached value (stale-on-failure). +#[derive(Debug)] +pub enum CanvasFetchResult { + Present(crate::queue::CanvasPointer), + Absent, + Failed, +} + +/// Process-lifetime per-channel canvas revision cache. +/// +/// Keyed by channel `Uuid`. Values: +/// - `Some(pointer)` — last known present canvas revision. +/// - `None` — confirmed absent (blank/deleted); cleared on `Absent` fetch. +/// - No entry — never fetched for this channel. +/// +/// Independent of `SessionState`: cache survives ACP session rotation/invalidation +/// so the canvas pointer is never lost when a session is recycled. +#[derive(Clone, Default)] +pub struct CanvasRevisionCache { + inner: Arc>>>, +} + +impl CanvasRevisionCache { + pub fn new() -> Self { + Self::default() + } + + /// Look up the cached pointer for `channel_id`. + /// + /// Returns `None` when there is no entry (never fetched) or the entry is + /// confirmed absent. Callers that need to distinguish "never fetched" from + /// "confirmed absent" should use `get_raw`. + pub fn get(&self, channel_id: &Uuid) -> Option { + self.inner + .read() + .unwrap_or_else(|p| p.into_inner()) + .get(channel_id) + .and_then(|v| v.clone()) + } + + /// Resolve the canvas pointer for the current turn, updating the cache. + /// + /// - `Present(p)` → cache `Some(p)`, return `Some(p)`. + /// - `Absent` → cache `None`, return `None`. + /// - `Failed` → cache unchanged; return last-known value or `None` on + /// first-fetch failure. + pub fn resolve_for_turn( + &self, + channel_id: &Uuid, + result: CanvasFetchResult, + ) -> Option { + let mut map = self.inner.write().unwrap_or_else(|p| p.into_inner()); + match result { + CanvasFetchResult::Present(p) => { + map.insert(*channel_id, Some(p.clone())); + Some(p) + } + CanvasFetchResult::Absent => { + map.insert(*channel_id, None); + None + } + CanvasFetchResult::Failed => { + // Serve stale; do not update the entry. + map.get(channel_id).and_then(|v| v.clone()) + } + } } } @@ -459,14 +524,43 @@ pub enum PromptOutcome { /// /// Built once from `Config` at startup. Avoids cloning the full config /// into every task. + +/// Per-entry TTL for channel metadata cache entries (~5 minutes). +const CHANNEL_INFO_TTL: std::time::Duration = std::time::Duration::from_secs(300); + +/// Backoff applied to `next_refresh_at` after a failed TTL refresh. +/// +/// On failure the entry is served stale; the next consumer will retry after +/// this interval rather than immediately re-driving the full retry sequence. +const CHANNEL_INFO_REFRESH_BACKOFF: std::time::Duration = std::time::Duration::from_secs(60); + +/// A cached channel metadata entry with a TTL-based refresh deadline. +#[derive(Clone, Debug)] +struct CachedChannelInfo { + info: PromptChannelInfo, + /// Absolute instant after which a background refresh should be triggered. + /// Consumers continue serving the cached `info` while the refresh runs. + refresh_after: tokio::time::Instant, + /// Absolute instant before which background refreshes should be suppressed + /// (set after a failed refresh to prevent retry storms on degraded relays). + next_refresh_at: tokio::time::Instant, +} + /// Shared channel-metadata resolver for startup-known and dynamically joined channels. /// /// Successful lazy lookups are cached for every consumer (author gate, prompt /// context, canvas, and setup mode). Unknown metadata is never cached as a /// non-DM: callers can fail closed and a later event retries resolution. +/// +/// Each entry carries a ~5-minute TTL (`CHANNEL_INFO_TTL`). On expiry the +/// current value is served immediately (stale-while-revalidate) and a +/// background refresh is spawned. On refresh failure the `next_refresh_at` +/// timestamp is advanced by `CHANNEL_INFO_REFRESH_BACKOFF` to throttle retries +/// on degraded relays — preventing repeated ~6.5-second retry sequences on the +/// DM author gate path. #[derive(Debug, Clone)] pub struct ChannelInfoResolver { - cache: std::sync::Arc>>, + cache: std::sync::Arc>>, rest_client: RestClient, } @@ -475,14 +569,20 @@ impl ChannelInfoResolver { startup: std::collections::HashMap, rest_client: RestClient, ) -> Self { + let now = tokio::time::Instant::now(); let cache = startup .into_iter() .filter_map(|(id, info)| { (info.channel_type != "unknown").then_some(( id, - PromptChannelInfo { - name: info.name, - channel_type: info.channel_type, + CachedChannelInfo { + info: PromptChannelInfo { + name: info.name, + channel_type: info.channel_type, + description: info.description, + }, + refresh_after: now + CHANNEL_INFO_TTL, + next_refresh_at: now, }, )) }) @@ -494,21 +594,73 @@ impl ChannelInfoResolver { } pub async fn resolve(&self, channel_id: Uuid) -> Option { - if let Some(info) = self + let now = tokio::time::Instant::now(); + + // Fast path: return cached entry; spawn a background refresh if expired. + if let Some(cached) = self .cache .read() .ok() - .and_then(|cache| cache.get(&channel_id).cloned()) + .and_then(|c| c.get(&channel_id).cloned()) { - return Some(info); + if now >= cached.refresh_after && now >= cached.next_refresh_at { + // Stale — spawn background refresh; serve stale immediately. + let resolver = self.clone(); + tokio::spawn(async move { + resolver.background_refresh(channel_id).await; + }); + } + return Some(cached.info); } + // Slow path: uncached — fetch synchronously (one logical sequence = two HTTP + // attempts per fetch_with_retry, exactly as before the TTL addition). let info = fetch_channel_info(channel_id, &self.rest_client).await?; + let now = tokio::time::Instant::now(); if let Ok(mut cache) = self.cache.write() { - cache.insert(channel_id, info.clone()); + cache.insert( + channel_id, + CachedChannelInfo { + info: info.clone(), + refresh_after: now + CHANNEL_INFO_TTL, + next_refresh_at: now, + }, + ); } Some(info) } + + /// Background TTL refresh: fetch fresh metadata and update the cache. + /// + /// On failure: advance `next_refresh_at` by `CHANNEL_INFO_REFRESH_BACKOFF` + /// so the next consumer does not immediately re-trigger the retry sequence. + /// On success: reset the TTL window. + async fn background_refresh(&self, channel_id: Uuid) { + match fetch_channel_info(channel_id, &self.rest_client).await { + Some(info) => { + let now = tokio::time::Instant::now(); + if let Ok(mut cache) = self.cache.write() { + cache.insert( + channel_id, + CachedChannelInfo { + info, + refresh_after: now + CHANNEL_INFO_TTL, + next_refresh_at: now, + }, + ); + } + } + None => { + // Fetch failed — serve stale; suppress retries for backoff duration. + let now = tokio::time::Instant::now(); + if let Ok(mut cache) = self.cache.write() { + if let Some(entry) = cache.get_mut(&channel_id) { + entry.next_refresh_at = now + CHANNEL_INFO_REFRESH_BACKOFF; + } + } + } + } + } } pub struct PromptContext { @@ -564,6 +716,11 @@ 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, + /// Process-lifetime per-channel canvas revision cache. + /// + /// Keyed by channel UUID; stores the last-known canvas pointer or confirmed + /// absence. Independent of `SessionState` — survives ACP session rotation. + pub canvas_cache: CanvasRevisionCache, } impl AgentPool { @@ -838,18 +995,9 @@ const PERMISSION_MODE_TIMEOUT: Duration = Duration::from_secs(5); /// an identifying name must treat it as absent. const UNKNOWN_CHANNEL_NAME: &str = "unknown"; -/// Channel-derived inputs for a new session — `(is_dm, title_channel)` — from +/// Channel-derived inputs for a new session — `title_channel` — from /// **one** metadata resolve. /// -/// Both new-session consumers need the same lookup: the canvas block skips DMs -/// (and fails closed when the channel type can't be determined), and the -/// session title is qualified with the channel name. Resolving once is -/// load-bearing rather than tidy: [`ChannelInfoResolver`] caches only `Some`, -/// so two calls against an unresolvable channel pay the whole -/// [`fetch_channel_info`] retry sequence twice — two `CONTEXT_FETCH_TIMEOUT` -/// attempts plus `CONTEXT_FETCH_RETRY_DELAY` each, in front of `session/new`, -/// precisely when the relay is already degraded. -/// /// `title_channel` is `None` whenever the channel can't usefully identify the /// session: an unresolved channel, a DM (no meaningful name), or the literal /// `"unknown"` that [`fetch_channel_info`] substitutes for a metadata event @@ -864,17 +1012,6 @@ const UNKNOWN_CHANNEL_NAME: &str = "unknown"; /// resolver's cached entry, so a renamed channel keeps its old suffix until the /// process restarts. An agent rename lands on the next spawn (the desktop /// restart badge covers it — see `spawn_config_hash`). -async fn resolve_new_session_channel_context( - channel_info: &ChannelInfoResolver, - channel_id: Uuid, -) -> (bool, Option) { - let Some(info) = channel_info.resolve(channel_id).await else { - return (true, None); - }; - let is_dm = info.channel_type == "dm"; - let title_channel = (!is_dm && info.name != UNKNOWN_CHANNEL_NAME).then_some(info.name); - (is_dm, title_channel) -} /// Create a new ACP session via `session_new_full()`, populate model capabilities /// on the agent (first session only), and apply `desired_model` if set. @@ -886,25 +1023,21 @@ async fn create_session_and_apply_model( agent: &mut OwnedAgent, ctx: &PromptContext, agent_core: Option<&str>, - agent_canvas: Option<&str>, channel_name: Option<&str>, ) -> Result { - // Build base_prompt + system_prompt + agent core + canvas metadata into a - // single prompt. Standard protocol-v2 agents receive it in `session/new`; + // Build base_prompt + system_prompt + agent core into a single prompt. + // Standard protocol-v2 agents receive it in `session/new`; // Goose receives it through the custom request below. Legacy agents receive // the same content as user-message sections via `format_prompt`. Core carries - // its own `[Agent Memory — core]` header, and canvas carries its own - // `[Channel Canvas]` header; both are appended with a blank-line separator. + // its own `[Agent Memory — core]` header and is appended with a blank-line + // separator. let is_goose = agent.agent_name == "goose"; - let combined_system_prompt = with_canvas( - with_core( - with_team( - framed_system_prompt(&ctx.cwd, ctx.base_prompt, ctx.system_prompt.as_deref()), - ctx.team_instructions.as_deref(), - ), - agent_core, + let combined_system_prompt = with_core( + with_team( + framed_system_prompt(&ctx.cwd, ctx.base_prompt, ctx.system_prompt.as_deref()), + ctx.team_instructions.as_deref(), ), - agent_canvas, + agent_core, ); let session_title = ctx @@ -1189,24 +1322,6 @@ pub(crate) fn prepend_base_for_legacy( } } -/// Prepend the `[Channel Canvas]` section to the legacy initial-message body. -/// -/// Protocol-v2 agents already receive the canvas in `systemPrompt`; only -/// legacy (protocol_version < 2) agents need it injected here so it arrives -/// before the first prompt — the same "every turn" semantics as per-turn core. -/// Heartbeats never have an initial_message, so the caller is responsible for -/// not passing a canvas when `source` is `Heartbeat`. -pub(crate) fn prepend_canvas_for_legacy( - protocol_version: u32, - agent_canvas: Option<&str>, - body: &str, -) -> String { - match agent_canvas { - Some(canvas) if protocol_version < 2 => format!("{canvas}\n\n{body}"), - _ => body.to_string(), - } -} - /// Frame the `session/new` `systemPrompt` so each present prompt carries its own /// header, keeping the base/persona boundary recoverable downstream. /// @@ -1294,20 +1409,6 @@ fn with_core(framed: Option, core: Option<&str>) -> Option { } } -/// Append the `[Channel Canvas]` metadata section onto the accumulated system prompt. -/// -/// The canvas section already carries its `[Channel Canvas]` header (from -/// `render_canvas_section`), so it is joined with a blank-line separator. -/// Either side may be absent. -fn with_canvas(prompt: Option, canvas: Option<&str>) -> Option { - match (prompt, canvas) { - (Some(prompt), Some(canvas)) => Some(format!("{prompt}\n\n{canvas}")), - (Some(prompt), None) => Some(prompt), - (None, Some(canvas)) => Some(canvas.to_string()), - (None, None) => None, - } -} - /// Return `agent` to the pool via `result_tx`, clearing any steer receiver first. /// /// Every path that returns an `OwnedAgent` to the pool via `PromptResult` goes @@ -1503,40 +1604,47 @@ pub async fn run_prompt_task( } } - // Canvas metadata fetch — same lifecycle as core: once per new channel session, - // never for heartbeats, cached until session invalidation. - // - // DM check: use startup channel_info first; lazy-fetch only when missing. - // A confirmed DM never receives a canvas section. If the channel type cannot - // be determined (metadata absent and lazy fetch fails/unknown), skip the canvas - // rather than assuming non-DM — failing closed on DM ambiguity is safer. - // - // I3 lifecycle: hold the fetched section in a local `pending_canvas` and - // commit it to `canvas_sections` only after session creation succeeds. This - // prevents a stale revision A surviving a failed create and being re-used by - // the next attempt after the canvas was cleared. - let mut pending_canvas: Option<(Uuid, String)> = None; - // Channel name for the session title, from the same single resolve the - // canvas DM check uses — see `resolve_new_session_channel_context`. + // Channel metadata — resolve once per turn for title (new sessions only) and + // the per-turn canvas fetch. DM check fails closed: a channel whose type + // cannot be determined is treated as a DM. let mut title_channel: Option = None; if let PromptSource::Channel(cid) = &source { let is_new_channel_session = !agent.state.sessions.contains_key(cid); - let needs_canvas = is_new_channel_session && !agent.state.canvas_sections.contains_key(cid); let needs_title = is_new_channel_session && ctx.session_title.is_some(); - if needs_canvas || needs_title { - let (is_dm, resolved_channel) = - resolve_new_session_channel_context(&ctx.channel_info, *cid).await; - title_channel = resolved_channel; - // A confirmed DM never receives a canvas section; an undeterminable - // channel type fails closed as a DM for the same reason. - if needs_canvas && !is_dm { - if let Some(section) = fetch_canvas_section(*cid, &ctx.rest_client).await { - pending_canvas = Some((*cid, section)); - } + if needs_title { + if let Some(info) = ctx.channel_info.resolve(*cid).await { + let is_dm = info.channel_type == "dm"; + title_channel = (!is_dm && info.name != UNKNOWN_CHANNEL_NAME).then_some(info.name); } } } + // Per-turn canvas fetch: resolve for the current turn and update the + // process-lifetime cache. Canvas is only fetched for channel turns (not + // heartbeats). DMs never receive canvas material — fail closed when channel + // type is unknown. + // + // A4: single-attempt, 3 s timeout, stale-on-failure. Runs before session + // creation so the pointer is available for initial_message. + let canvas_pointer: Option = match &source { + PromptSource::Channel(cid) => { + // Determine DM status before fetching canvas. + let is_dm = ctx + .channel_info + .resolve(*cid) + .await + .map(|info| info.channel_type == "dm") + .unwrap_or(true); // fail closed on unknown + if is_dm { + None + } else { + let result = fetch_canvas_pointer(*cid, &ctx.rest_client).await; + ctx.canvas_cache.resolve_for_turn(cid, result) + } + } + PromptSource::Heartbeat => None, + }; + // The core section to fold into the system prompt for this turn's session. // Channel-scoped; heartbeats carry no owner core. let agent_core: Option = match &source { @@ -1544,18 +1652,6 @@ pub async fn run_prompt_task( PromptSource::Heartbeat => None, }; - // The canvas metadata section — channel-scoped, absent for heartbeats/DMs. - // Prefer the committed cache; fall back to pending (for new sessions being created now). - let agent_canvas: Option = match &source { - PromptSource::Channel(cid) => agent - .state - .canvas_sections - .get(cid) - .cloned() - .or_else(|| pending_canvas.as_ref().map(|(_, s)| s.clone())), - PromptSource::Heartbeat => None, - }; - let (session_id, is_new_session) = match &source { PromptSource::Channel(cid) => { if let Some(sid) = agent.state.sessions.get(cid) { @@ -1569,7 +1665,6 @@ pub async fn run_prompt_task( &mut agent, &ctx, agent_core.as_deref(), - agent_canvas.as_deref(), title_channel.as_deref(), ) .await @@ -1580,10 +1675,6 @@ pub async fn run_prompt_task( "created session {sid} for channel {cid}" ); agent.state.sessions.insert(*cid, sid.clone()); - // Commit canvas only after session creation succeeds (I3). - if let Some((pending_cid, section)) = pending_canvas.take() { - agent.state.canvas_sections.insert(pending_cid, section); - } (sid, true) } Err(AcpError::AgentExited) => { @@ -1618,7 +1709,7 @@ pub async fn run_prompt_task( if let Some(sid) = &agent.state.heartbeat_session { (sid.clone(), false) } else { - match create_session_and_apply_model(&mut agent, &ctx, None, None, None).await { + match create_session_and_apply_model(&mut agent, &ctx, None, None).await { Ok(sid) => { tracing::info!( target: "pool::session", @@ -1679,12 +1770,12 @@ pub async fn run_prompt_task( target: "pool::session", "sending initial_message to session {session_id} for channel {cid}" ); - // For agents with systemPrompt support (protocol_version >= 2), - // base_prompt is delivered via the system role in session/new. - // Legacy agents receive it via [Base] in the user message instead. - // Canvas is also injected here for legacy agents: protocol-v2 agents - // already have it in systemPrompt; legacy agents need it before the - // first prompt, matching the "every turn" per-turn delivery semantics. + // Build the initial_message content. Base prompt is prepended for + // legacy agents (protocol < 2) in the user message; protocol-v2 + // agents already have it in systemPrompt. + // + // Both modern and legacy agents receive the channel [Context] block + // (A2 — canvas awareness before the first prompt). let init_msg = prepend_base_for_legacy( if agent.has_system_prompt_support() { 2 @@ -1694,15 +1785,29 @@ pub async fn run_prompt_task( ctx.base_prompt, initial_msg, ); - let init_msg = prepend_canvas_for_legacy( - if agent.has_system_prompt_support() { - 2 - } else { - 1 - }, - agent_canvas.as_deref(), - &init_msg, - ); + // Prepend [Context] block (description + canvas pointer) to initial_message. + let init_msg = { + let channel_info = ctx.channel_info.resolve(*cid).await; + let mut ctx_block = String::from("[Context]\nScope: channel"); + ctx_block.push_str(&format!( + "\nChannel: {} (#{})", + channel_info + .as_ref() + .map(|i| i.name.as_str()) + .unwrap_or("unknown"), + cid + )); + crate::queue::append_description_for_initial_message( + &mut ctx_block, + channel_info.as_ref(), + ); + crate::queue::append_canvas_pointer_for_initial_message( + &mut ctx_block, + canvas_pointer.as_ref(), + &cid.to_string(), + ); + format!("{ctx_block}\n\n{init_msg}") + }; let init_result = agent .acp .session_prompt_with_idle_timeout( @@ -1874,7 +1979,7 @@ pub async fn run_prompt_task( base_prompt: ctx.base_prompt, system_prompt: ctx.system_prompt.as_deref(), team_instructions: ctx.team_instructions.as_deref(), - agent_canvas: agent_canvas.as_deref(), + canvas_pointer: canvas_pointer.as_ref(), }, ) } else { @@ -2366,17 +2471,25 @@ pub(crate) async fn fetch_channel_info( let ev = events.first()?; let tags = ev.get("tags")?.as_array()?; let mut name = None; + let mut description = None; for tag in tags { if let Some(arr) = tag.as_array() { - if arr.first().and_then(|v| v.as_str()) == Some("name") { - name = arr.get(1).and_then(|v| v.as_str()); + match arr.first().and_then(|v| v.as_str()) { + Some("name") => name = arr.get(1).and_then(|v| v.as_str()), + Some("about") => description = arr.get(1).and_then(|v| v.as_str()), + _ => {} } } } let channel_type = crate::relay::channel_type_from_tags(tags); + let description = description + .map(|s| s.trim()) + .filter(|s| !s.is_empty()) + .map(str::to_string); Some(PromptChannelInfo { name: name.unwrap_or(UNKNOWN_CHANNEL_NAME).to_string(), channel_type, + description, }) } Ok(Err(e)) => { @@ -2398,20 +2511,17 @@ pub(crate) async fn fetch_channel_info( .await } -/// Fetch the latest canvas event for `channel_id` and return a rendered -/// `[Channel Canvas]` metadata section, or `None` if absent/blank/error. +/// Fetch the latest canvas event for `channel_id` and return a `CanvasFetchResult`. /// -/// Failure modes (all fail open — no crash, no block): -/// * relay returns no event → `None` -/// * latest event's content is blank → `None` (cleared canvas; older revisions -/// are NOT resurrected) -/// * malformed JSON array, missing fields, bad event ID, bad timestamp → -/// logged at `warn`; returns `None` -/// * REST error or timeout → returns `None` +/// - `Present(pointer)` — a valid, non-blank canvas event was found. +/// - `Absent` — relay confirmed no event, or the latest event has blank content +/// (cleared canvas); older revisions are NOT resurrected. +/// - `Failed` — transport timeout, REST error, or parse/verification error. +/// Callers must serve stale on `Failed`. /// -/// Called at most once per new channel session; the result is cached in -/// `SessionState::canvas_sections` and cleared on session invalidation. -async fn fetch_canvas_section(channel_id: Uuid, rest: &RestClient) -> Option { +/// Single-attempt with a 3-second timeout — never wrapped in `fetch_with_retry`. +/// Accepted latency bound: at most +3 s per turn on a degraded relay. +async fn fetch_canvas_pointer(channel_id: Uuid, rest: &RestClient) -> CanvasFetchResult { use nostr::{Alphabet, SingleLetterTag}; let h_tag = SingleLetterTag::lowercase(Alphabet::H); @@ -2421,6 +2531,7 @@ async fn fetch_canvas_section(channel_id: Uuid, rest: &RestClient) -> Option Option { tracing::warn!( target: "canvas::fetch", channel = %channel_id, timeout_ms = CANVAS_FETCH_TIMEOUT.as_millis() as u64, - "canvas fetch timed out — emitting no section" + "canvas fetch timed out — serving stale" ); - return None; + return CanvasFetchResult::Failed; } }; @@ -2453,27 +2564,31 @@ async fn fetch_canvas_section(channel_id: Uuid, rest: &RestClient) -> Option CanvasFetchResult::Present(pointer), + None => CanvasFetchResult::Absent, + } } -/// Parse a canvas query response array and render a `[Channel Canvas]` section. +/// Parse a canvas query response array and return a `CanvasPointer`. /// /// Extracted as a pure function so tests can exercise the parsing/validation /// logic without async machinery or relay connectivity. /// /// Returns `None` on: empty array, blank content, malformed/partial event JSON /// (requires a complete, structurally valid Nostr event), or an out-of-range -/// `created_at` timestamp. Never falls back to epoch or raw integers. -pub(crate) fn canvas_section_from_query_response( +/// `created_at` timestamp. A `None` result signals `Absent` (not `Failed`) — +/// the relay responded but no usable canvas exists. +pub(crate) fn canvas_pointer_from_query_response( events: &[serde_json::Value], channel_uuid: &str, -) -> Option { +) -> Option { let raw = events.first()?; // Deserialise as a complete Nostr Event. Partial objects (missing pubkey, @@ -2485,7 +2600,7 @@ pub(crate) fn canvas_section_from_query_response( target: "canvas::fetch", channel = %channel_uuid, %err, - "canvas query returned a malformed event — emitting no section", + "canvas query returned a malformed event — emitting no pointer", ); return None; } @@ -2498,7 +2613,7 @@ pub(crate) fn canvas_section_from_query_response( target: "canvas::fetch", channel = %channel_uuid, %err, - "canvas event failed signature verification — emitting no section", + "canvas event failed signature verification — emitting no pointer", ); return None; } @@ -2509,7 +2624,7 @@ pub(crate) fn canvas_section_from_query_response( target: "canvas::fetch", channel = %channel_uuid, kind = %event.kind.as_u16(), - "canvas event has unexpected kind — emitting no section", + "canvas event has unexpected kind — emitting no pointer", ); return None; } @@ -2525,7 +2640,7 @@ pub(crate) fn canvas_section_from_query_response( tracing::warn!( target: "canvas::fetch", channel = %channel_uuid, - "canvas event is missing expected h-tag — emitting no section", + "canvas event is missing expected h-tag — emitting no pointer", ); return None; } @@ -2535,7 +2650,7 @@ pub(crate) fn canvas_section_from_query_response( tracing::debug!( target: "canvas::fetch", channel = %channel_uuid, - "latest canvas event has blank content — emitting no section" + "latest canvas event has blank content — emitting no pointer (Absent)" ); return None; } @@ -2552,7 +2667,7 @@ pub(crate) fn canvas_section_from_query_response( tracing::warn!( target: "canvas::fetch", channel = %channel_uuid, - "canvas event created_at overflows i64 — emitting no section", + "canvas event created_at overflows i64 — emitting no pointer", ); return None; } @@ -2564,7 +2679,7 @@ pub(crate) fn canvas_section_from_query_response( target: "canvas::fetch", channel = %channel_uuid, ts_secs, - "canvas event has out-of-range created_at — emitting no section", + "canvas event has out-of-range created_at — emitting no pointer", ); return None; } @@ -2574,22 +2689,12 @@ pub(crate) fn canvas_section_from_query_response( target: "canvas::fetch", channel = %channel_uuid, event_id = %id, - "injected channel canvas metadata section into system prompt" + "resolved canvas revision pointer for [Context]" ); - Some(render_canvas_section(&id, ×tamp, channel_uuid)) -} - -/// Render the `[Channel Canvas]` metadata section string. -/// -/// Pure function — kept separate so unit tests can exercise rendering -/// without async machinery or relay connectivity. -pub(crate) fn render_canvas_section(event_id: &str, timestamp: &str, channel_uuid: &str) -> String { - format!( - "[Channel Canvas]\n\ - Canvas revision (event ID): {event_id}\n\ - Last modified: {timestamp}\n\ - Fetch current content with: buzz canvas get --channel {channel_uuid}" - ) + Some(crate::queue::CanvasPointer { + event_id: id, + timestamp, + }) } /// Fetch conversation context (thread or DM) for a batch before prompting. @@ -4068,80 +4173,6 @@ mod tests { assert_eq!(composed, "hello channel"); } - // ── prepend_canvas_for_legacy ───────────────────────────────────────────── - - #[test] - fn test_initial_message_legacy_agent_gets_canvas_prepended() { - // Legacy agents (protocol_version < 2) receive the canvas section before - // the initial-message body so it arrives before the first prompt. - let canvas = "[Channel Canvas]\nCanvas revision (event ID): abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234\nLast modified: 2024-01-15T10:30:00Z\nFetch current content with: buzz canvas get --channel 00f1ccaf-1506-4dd7-9a0e-fa67e9e486ae"; - let composed = prepend_canvas_for_legacy(1, Some(canvas), "do the thing"); - assert!( - composed.starts_with("[Channel Canvas]"), - "canvas must precede the body" - ); - assert!( - composed.ends_with("do the thing"), - "body must follow the canvas" - ); - assert!( - composed.contains("\n\ndo the thing"), - "canvas and body separated by blank line" - ); - } - - #[test] - fn test_initial_message_modern_agent_omits_canvas_from_body() { - // Protocol-v2 agents receive canvas in systemPrompt; it must NOT be - // duplicated in the initial-message user turn. - let canvas = "[Channel Canvas]\nsome section"; - let composed = prepend_canvas_for_legacy(2, Some(canvas), "do the thing"); - assert_eq!( - composed, "do the thing", - "modern agent initial message must not contain canvas" - ); - assert!( - !composed.contains("[Channel Canvas]"), - "canvas must be absent from modern agent initial message" - ); - } - - #[test] - fn test_initial_message_legacy_agent_no_canvas_is_unchanged() { - // No canvas present: body passes through unmodified. - let composed = prepend_canvas_for_legacy(1, None, "do the thing"); - assert_eq!(composed, "do the thing"); - } - - #[test] - fn test_initial_message_legacy_canvas_and_base_compose_correctly() { - // Verify the full composition order when both base and canvas are present: - // [Base] → canvas section → initial-message body. - let canvas = "[Channel Canvas]\ncanvas content"; - let base_composed = prepend_base_for_legacy(1, Some("be helpful"), "do the thing"); - let full = prepend_canvas_for_legacy(1, Some(canvas), &base_composed); - assert!( - full.starts_with("[Channel Canvas]"), - "canvas must be first in composed message" - ); - assert!( - full.contains("[Base]"), - "base must be present in composed message" - ); - assert!( - full.ends_with("do the thing"), - "body must be last in composed message" - ); - // Order: canvas → base → body - let canvas_pos = full.find("[Channel Canvas]").unwrap(); - let base_pos = full.find("[Base]").unwrap(); - let body_pos = full.find("do the thing").unwrap(); - assert!( - canvas_pos < base_pos && base_pos < body_pos, - "order must be: canvas → base → body" - ); - } - // Pin the session/new systemPrompt framing: each present prompt carries its // own header so the desktop observer can split into labeled sub-sections. @@ -6457,108 +6488,11 @@ mod tests { memory_enabled: false, harness_name: "goose".to_string(), relay_url: "ws://127.0.0.1:3000".to_string(), + canvas_cache: CanvasRevisionCache::new(), } } - // ── render_canvas_section ──────────────────────────────────────────────── - - #[test] - fn test_render_canvas_section_produces_exact_shape() { - let id = "a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2"; - let ts = "2024-01-15T10:30:00+00:00"; - let uuid = "00f1ccaf-1506-4dd7-9a0e-fa67e9e486ae"; - let section = render_canvas_section(id, ts, uuid); - assert_eq!( - section, - "[Channel Canvas]\n\ - Canvas revision (event ID): a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2\n\ - Last modified: 2024-01-15T10:30:00+00:00\n\ - Fetch current content with: buzz canvas get --channel 00f1ccaf-1506-4dd7-9a0e-fa67e9e486ae" - ); - } - - // ── with_canvas ────────────────────────────────────────────────────────── - - #[test] - fn test_with_canvas_appends_to_existing_prompt() { - let result = with_canvas(Some("base content".into()), Some("[Channel Canvas]\nstuff")); - assert_eq!(result.unwrap(), "base content\n\n[Channel Canvas]\nstuff"); - } - - #[test] - fn test_with_canvas_returns_canvas_alone_when_no_prompt() { - let result = with_canvas(None, Some("[Channel Canvas]\nstuff")); - assert_eq!(result.unwrap(), "[Channel Canvas]\nstuff"); - } - - #[test] - fn test_with_canvas_returns_prompt_alone_when_no_canvas() { - let result = with_canvas(Some("base content".into()), None); - assert_eq!(result.unwrap(), "base content"); - } - - #[test] - fn test_with_canvas_returns_none_when_both_absent() { - let result = with_canvas(None, None); - assert!(result.is_none()); - } - - // ── canvas_sections cache invalidation ─────────────────────────────────── - - #[test] - fn test_invalidate_channel_clears_canvas_section() { - let ch = Uuid::new_v4(); - let mut s = SessionState::default(); - s.sessions.insert(ch, "sess".into()); - s.canvas_sections - .insert(ch, "[Channel Canvas]\nrev abc".into()); - - s.invalidate_channel(&ch); - - assert!(!s.canvas_sections.contains_key(&ch)); - assert!(!s.sessions.contains_key(&ch)); - } - - #[test] - fn test_invalidate_all_clears_canvas_sections() { - let ch_a = Uuid::new_v4(); - let ch_b = Uuid::new_v4(); - let mut s = SessionState::default(); - s.canvas_sections.insert(ch_a, "canvas-a".into()); - s.canvas_sections.insert(ch_b, "canvas-b".into()); - s.sessions.insert(ch_a, "sess-a".into()); - - s.invalidate_all(); - - assert!(s.canvas_sections.is_empty()); - assert!(s.sessions.is_empty()); - } - - #[test] - fn test_invalidate_channel_leaves_other_channels_canvas_intact() { - let ch_a = Uuid::new_v4(); - let ch_b = Uuid::new_v4(); - let mut s = SessionState::default(); - s.sessions.insert(ch_a, "sess-a".into()); - s.sessions.insert(ch_b, "sess-b".into()); - s.canvas_sections.insert(ch_a, "canvas-a".into()); - s.canvas_sections.insert(ch_b, "canvas-b".into()); - - s.invalidate_channel(&ch_a); - - assert!(!s.canvas_sections.contains_key(&ch_a)); - assert_eq!(s.canvas_sections.get(&ch_b).unwrap(), "canvas-b"); - } - - #[test] - fn test_has_channel_state_true_when_only_canvas_section_present() { - let ch = Uuid::new_v4(); - let mut s = SessionState::default(); - s.canvas_sections.insert(ch, "canvas".into()); - assert!(s.has_channel_state(&ch)); - } - - // ── canvas_section_from_query_response ─────────────────────────────────── + // ── canvas_pointer_from_query_response ────────────────────────────────── const CHANNEL_UUID: &str = "00f1ccaf-1506-4dd7-9a0e-fa67e9e486ae"; @@ -6577,29 +6511,33 @@ mod tests { } #[test] - fn test_canvas_section_from_query_response_happy_path() { + fn test_canvas_pointer_from_query_response_happy_path() { let ev = make_canvas_event_value("# Team instructions\nBe helpful."); let id = ev["id"].as_str().unwrap().to_string(); - let result = canvas_section_from_query_response(&[ev], CHANNEL_UUID); - let section = result.expect("expected Some"); - assert!(section.contains(&id), "section must contain the event id"); - assert!(section.contains("buzz canvas get --channel")); - assert!(section.contains(CHANNEL_UUID)); - assert!(section.starts_with("[Channel Canvas]")); + let result = canvas_pointer_from_query_response(&[ev], CHANNEL_UUID); + let pointer = result.expect("expected Some"); + assert_eq!(pointer.event_id, id, "pointer must carry the event id"); // Timestamp must use Z suffix, not +00:00 - assert!(section.contains('Z'), "timestamp must use Z suffix"); + assert!( + pointer.timestamp.ends_with('Z'), + "timestamp must use Z suffix" + ); + assert!( + !pointer.timestamp.contains("+00:00"), + "timestamp must not use +00:00 offset" + ); } #[test] - fn test_canvas_section_from_query_response_empty_array_returns_none() { - let result = canvas_section_from_query_response(&[], CHANNEL_UUID); + fn test_canvas_pointer_from_query_response_empty_array_returns_none() { + let result = canvas_pointer_from_query_response(&[], CHANNEL_UUID); assert!(result.is_none()); } #[test] - fn test_canvas_section_from_query_response_blank_content_returns_none() { + fn test_canvas_pointer_from_query_response_blank_content_returns_none() { let ev = make_canvas_event_value(" "); - let result = canvas_section_from_query_response(&[ev], CHANNEL_UUID); + let result = canvas_pointer_from_query_response(&[ev], CHANNEL_UUID); assert!( result.is_none(), "blank content must return None (cleared canvas)" @@ -6607,22 +6545,22 @@ mod tests { } #[test] - fn test_canvas_section_from_query_response_empty_content_returns_none() { + fn test_canvas_pointer_from_query_response_empty_content_returns_none() { let ev = make_canvas_event_value(""); - let result = canvas_section_from_query_response(&[ev], CHANNEL_UUID); + let result = canvas_pointer_from_query_response(&[ev], CHANNEL_UUID); assert!(result.is_none()); } /// A bare JSON object with a plausible-looking id but missing pubkey/sig/kind/tags /// must be rejected — not silently accepted with partial metadata. #[test] - fn test_canvas_section_from_query_response_partial_object_returns_none() { + fn test_canvas_pointer_from_query_response_partial_object_returns_none() { let partial = serde_json::json!({ "id": "a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2", "created_at": 1705312200_i64, "content": "some instructions" }); - let result = canvas_section_from_query_response(&[partial], CHANNEL_UUID); + let result = canvas_pointer_from_query_response(&[partial], CHANNEL_UUID); assert!( result.is_none(), "partial event object (missing pubkey/sig/kind/tags) must return None" @@ -6632,7 +6570,7 @@ mod tests { /// A JSON object that looks like an event but has `created_at` as a string /// must be rejected — the nostr::Event parser enforces integer type. #[test] - fn test_canvas_section_from_query_response_string_timestamp_returns_none() { + fn test_canvas_pointer_from_query_response_string_timestamp_returns_none() { let keys = Keys::generate(); let h_tag = Tag::parse(["h", CHANNEL_UUID]).expect("h tag"); let mut ev = serde_json::to_value( @@ -6644,7 +6582,7 @@ mod tests { .expect("serialise"); // Corrupt created_at to a string value. ev["created_at"] = serde_json::Value::String("2026-03-15T16:30:00+00:00".into()); - let result = canvas_section_from_query_response(&[ev], CHANNEL_UUID); + let result = canvas_pointer_from_query_response(&[ev], CHANNEL_UUID); assert!( result.is_none(), "string created_at must be rejected by nostr::Event deserialiser" @@ -6654,7 +6592,7 @@ mod tests { /// A JSON object that looks like an event but is missing `created_at` /// must be rejected — nostr::Event requires the field. #[test] - fn test_canvas_section_from_query_response_missing_timestamp_returns_none() { + fn test_canvas_pointer_from_query_response_missing_timestamp_returns_none() { let keys = Keys::generate(); let h_tag = Tag::parse(["h", CHANNEL_UUID]).expect("h tag"); let mut ev = serde_json::to_value( @@ -6665,7 +6603,7 @@ mod tests { ) .expect("serialise"); ev.as_object_mut().unwrap().remove("created_at"); - let result = canvas_section_from_query_response(&[ev], CHANNEL_UUID); + let result = canvas_pointer_from_query_response(&[ev], CHANNEL_UUID); assert!( result.is_none(), "missing created_at must be rejected by nostr::Event deserialiser" @@ -6677,7 +6615,7 @@ mod tests { /// `u64::MAX as i64` wraps to -1, which chrono silently accepts as /// 1969-12-31T23:59:59Z. The checked i64::try_from must reject it first. #[test] - fn test_canvas_section_from_query_response_timestamp_max_returns_none() { + fn test_canvas_pointer_from_query_response_timestamp_max_returns_none() { let keys = Keys::generate(); let h_tag = Tag::parse(["h", CHANNEL_UUID]).expect("h tag"); let ev = serde_json::to_value( @@ -6688,7 +6626,7 @@ mod tests { .expect("sign"), ) .expect("serialise"); - let result = canvas_section_from_query_response(&[ev], CHANNEL_UUID); + let result = canvas_pointer_from_query_response(&[ev], CHANNEL_UUID); assert!( result.is_none(), "Timestamp::max() (u64::MAX) must return None — not wrap to 1969" @@ -6698,7 +6636,7 @@ mod tests { /// A structurally complete but tampered event (content altered after signing) /// must be rejected by event.verify(). #[test] - fn test_canvas_section_from_query_response_tampered_event_returns_none() { + fn test_canvas_pointer_from_query_response_tampered_event_returns_none() { let keys = Keys::generate(); let h_tag = Tag::parse(["h", CHANNEL_UUID]).expect("h tag"); let mut ev = serde_json::to_value( @@ -6713,7 +6651,7 @@ mod tests { .expect("serialise"); // Tamper the content after signing — id and sig no longer agree. ev["content"] = serde_json::Value::String("injected instructions".into()); - let result = canvas_section_from_query_response(&[ev], CHANNEL_UUID); + let result = canvas_pointer_from_query_response(&[ev], CHANNEL_UUID); assert!( result.is_none(), "tampered event must fail verify() and return None" @@ -6722,7 +6660,7 @@ mod tests { /// An event with the wrong kind (not 40100) must be rejected. #[test] - fn test_canvas_section_from_query_response_wrong_kind_returns_none() { + fn test_canvas_pointer_from_query_response_wrong_kind_returns_none() { let keys = Keys::generate(); let h_tag = Tag::parse(["h", CHANNEL_UUID]).expect("h tag"); let ev = serde_json::to_value( @@ -6732,14 +6670,14 @@ mod tests { .expect("sign"), ) .expect("serialise"); - let result = canvas_section_from_query_response(&[ev], CHANNEL_UUID); + let result = canvas_pointer_from_query_response(&[ev], CHANNEL_UUID); assert!(result.is_none(), "wrong kind must return None"); } /// An event missing the expected h-tag (or carrying a different channel UUID) /// must be rejected. #[test] - fn test_canvas_section_from_query_response_wrong_h_tag_returns_none() { + fn test_canvas_pointer_from_query_response_wrong_h_tag_returns_none() { let keys = Keys::generate(); let wrong_h = Tag::parse(["h", "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"]).expect("h tag"); let ev = serde_json::to_value( @@ -6749,25 +6687,121 @@ mod tests { .expect("sign"), ) .expect("serialise"); - let result = canvas_section_from_query_response(&[ev], CHANNEL_UUID); + let result = canvas_pointer_from_query_response(&[ev], CHANNEL_UUID); assert!(result.is_none(), "mismatched h-tag must return None"); } #[test] - fn test_canvas_section_from_query_response_timestamp_uses_z_suffix() { + fn test_canvas_pointer_from_query_response_timestamp_uses_z_suffix() { let ev = make_canvas_event_value("instructions"); - let result = canvas_section_from_query_response(&[ev], CHANNEL_UUID); - let section = result.expect("valid event must produce a section"); + let result = canvas_pointer_from_query_response(&[ev], CHANNEL_UUID); + let pointer = result.expect("valid event must produce a pointer"); assert!( - section.contains('Z'), + pointer.timestamp.ends_with('Z'), "RFC3339 timestamp must use Z suffix, not +00:00" ); assert!( - !section.contains("+00:00"), + !pointer.timestamp.contains("+00:00"), "timestamp must not use +00:00 offset" ); } + // ── CanvasRevisionCache tri-state (A1) ──────────────────────────────────── + + #[test] + fn test_canvas_revision_cache_present_updates_and_returns_pointer() { + let cache = CanvasRevisionCache::new(); + let ch = Uuid::new_v4(); + let pointer = crate::queue::CanvasPointer { + event_id: "abc123".to_string(), + timestamp: "2024-01-15T10:30:00Z".to_string(), + }; + let result = cache.resolve_for_turn(&ch, CanvasFetchResult::Present(pointer.clone())); + assert_eq!(result, Some(pointer.clone())); + // Cache is updated: get() returns the pointer. + assert_eq!(cache.get(&ch), Some(pointer)); + } + + #[test] + fn test_canvas_revision_cache_absent_clears_entry() { + let cache = CanvasRevisionCache::new(); + let ch = Uuid::new_v4(); + // Seed with a present pointer. + let pointer = crate::queue::CanvasPointer { + event_id: "abc123".to_string(), + timestamp: "2024-01-15T10:30:00Z".to_string(), + }; + cache.resolve_for_turn(&ch, CanvasFetchResult::Present(pointer)); + // Now canvas is cleared. + let result = cache.resolve_for_turn(&ch, CanvasFetchResult::Absent); + assert!(result.is_none(), "Absent must return None"); + assert!( + cache.get(&ch).is_none(), + "Absent must clear the cache entry" + ); + } + + #[test] + fn test_canvas_revision_cache_failed_serves_stale() { + let cache = CanvasRevisionCache::new(); + let ch = Uuid::new_v4(); + // Seed with a present pointer. + let pointer = crate::queue::CanvasPointer { + event_id: "stale123".to_string(), + timestamp: "2024-01-15T10:30:00Z".to_string(), + }; + cache.resolve_for_turn(&ch, CanvasFetchResult::Present(pointer.clone())); + // Fetch fails. + let result = cache.resolve_for_turn(&ch, CanvasFetchResult::Failed); + assert_eq!(result, Some(pointer), "Failed must serve stale value"); + } + + #[test] + fn test_canvas_revision_cache_failed_first_returns_none() { + let cache = CanvasRevisionCache::new(); + let ch = Uuid::new_v4(); + // No prior entry; first-fetch failure must emit nothing. + let result = cache.resolve_for_turn(&ch, CanvasFetchResult::Failed); + assert!( + result.is_none(), + "Failed with no prior entry must return None" + ); + } + + #[test] + fn test_canvas_revision_cache_present_to_absent_transition() { + // present → absent: confirmed deletion clears the pointer. + let cache = CanvasRevisionCache::new(); + let ch = Uuid::new_v4(); + let pointer = crate::queue::CanvasPointer { + event_id: "rev1".to_string(), + timestamp: "2024-01-15T10:30:00Z".to_string(), + }; + cache.resolve_for_turn(&ch, CanvasFetchResult::Present(pointer)); + let result = cache.resolve_for_turn(&ch, CanvasFetchResult::Absent); + assert!(result.is_none(), "canvas deletion must clear the pointer"); + } + + #[test] + fn test_canvas_revision_cache_present_to_new_revision() { + // present → new revision: pointer updated. + let cache = CanvasRevisionCache::new(); + let ch = Uuid::new_v4(); + let old = crate::queue::CanvasPointer { + event_id: "rev1".to_string(), + timestamp: "2024-01-15T10:00:00Z".to_string(), + }; + let new_p = crate::queue::CanvasPointer { + event_id: "rev2".to_string(), + timestamp: "2024-01-16T10:00:00Z".to_string(), + }; + cache.resolve_for_turn(&ch, CanvasFetchResult::Present(old)); + let result = cache.resolve_for_turn(&ch, CanvasFetchResult::Present(new_p.clone())); + assert_eq!(result, Some(new_p), "new revision must replace old pointer"); + } + + // ── ChannelInfoResolver TTL (A3) ────────────────────────────────────────── + // ── new-session channel context (one resolve, two consumers) ───────────── /// A [`ChannelInfoResolver`] whose lazy REST fallback is served by a local @@ -6823,23 +6857,24 @@ mod tests { json!([{ "tags": event_tags }]) } - /// A normal channel yields a non-DM (canvas allowed) and its name for the - /// title suffix — and the second consumer reads it from cache, not the wire. + /// A normal channel resolves to its name and type; cached second call hits no wire. #[tokio::test] - async fn test_new_session_channel_context_qualifies_a_normal_channel() { + async fn test_channel_resolver_qualifies_a_normal_channel() { use std::sync::atomic::Ordering; let id = Uuid::new_v4(); let response = channel_metadata_response(id, &[["name", "buzz-dev"], ["t", "stream"]]); let (resolver, requests, server) = counting_resolver(response).await; - let (is_dm, title_channel) = resolve_new_session_channel_context(&resolver, id).await; - assert!(!is_dm, "a stream channel is not a DM"); - assert_eq!(title_channel.as_deref(), Some("buzz-dev")); + let info = resolver.resolve(id).await.expect("should resolve"); + assert_eq!(info.name, "buzz-dev"); + assert_eq!(info.channel_type, "stream"); + assert!(!info.name.is_empty()); assert_eq!(requests.load(Ordering::SeqCst), 1); - let (_, again) = resolve_new_session_channel_context(&resolver, id).await; - assert_eq!(again.as_deref(), Some("buzz-dev")); + // Second resolve hits the cache, not the wire. + let again = resolver.resolve(id).await.expect("cached resolve"); + assert_eq!(again.name, "buzz-dev"); assert_eq!( requests.load(Ordering::SeqCst), 1, @@ -6848,56 +6883,44 @@ mod tests { server.abort(); } - /// A DM carries no useful name, so it gets the bare agent title (and no - /// canvas section). + /// A DM channel resolves with the correct type. #[tokio::test] - async fn test_new_session_channel_context_leaves_a_dm_unqualified() { + async fn test_channel_resolver_resolves_dm_type() { let id = Uuid::new_v4(); let response = channel_metadata_response(id, &[["name", "DM"], ["t", "dm"]]); let (resolver, _requests, server) = counting_resolver(response).await; - let (is_dm, title_channel) = resolve_new_session_channel_context(&resolver, id).await; - assert!(is_dm); - assert_eq!( - title_channel, None, - "a DM name must never reach the session title" - ); + let info = resolver.resolve(id).await.expect("should resolve"); + assert_eq!(info.channel_type, "dm"); server.abort(); } - /// The `"unknown"` placeholder `fetch_channel_info` substitutes for a - /// metadata event with no `name` tag is not a channel name: qualifying with - /// it would title every unnamed channel `Agent · #unknown`. + /// The `"unknown"` placeholder is returned as-is — callers exclude it from titles. #[tokio::test] - async fn test_new_session_channel_context_treats_the_unknown_name_as_absent() { + async fn test_channel_resolver_returns_unknown_name_for_unnamed_channel() { let id = Uuid::new_v4(); let response = channel_metadata_response(id, &[["t", "stream"]]); let (resolver, _requests, server) = counting_resolver(response).await; - let (is_dm, title_channel) = resolve_new_session_channel_context(&resolver, id).await; - assert!(!is_dm, "a nameless stream channel is still not a DM"); + let info = resolver.resolve(id).await.expect("should resolve"); assert_eq!( - title_channel, None, - "the `unknown` placeholder must yield a bare title" + info.name, "unknown", + "nameless channel gets unknown placeholder" ); server.abort(); } - /// An unresolvable channel yields the bare title, fails closed as a DM, and - /// costs exactly ONE `fetch_channel_info` sequence — two attempts, because - /// `fetch_with_retry` retries once. `resolve()` caches only `Some`, so a - /// second resolve for the title would double this in front of `session/new`, - /// exactly when the relay is already degraded. + /// An unresolvable channel costs exactly ONE `fetch_channel_info` sequence — + /// two HTTP attempts (initial + one retry from fetch_with_retry). The resolver + /// does not cache failures, so a second call will retry. #[tokio::test] - async fn test_new_session_channel_context_attempts_an_unresolved_channel_once() { + async fn test_channel_resolver_attempts_unresolved_channel_once() { use std::sync::atomic::Ordering; let (resolver, requests, server) = counting_resolver(json!([])).await; - let (is_dm, title_channel) = - resolve_new_session_channel_context(&resolver, Uuid::new_v4()).await; - assert!(is_dm, "an undeterminable channel type must fail closed"); - assert_eq!(title_channel, None, "unresolved channels get a bare title"); + let result = resolver.resolve(Uuid::new_v4()).await; + assert!(result.is_none(), "unresolvable channel returns None"); assert_eq!( requests.load(Ordering::SeqCst), 2, @@ -6905,4 +6928,122 @@ mod tests { ); server.abort(); } + + /// Channel description is delivered through resolver when present in metadata. + #[tokio::test] + async fn test_channel_resolver_delivers_description() { + let id = Uuid::new_v4(); + let response = channel_metadata_response( + id, + &[ + ["name", "team-chat"], + ["t", "stream"], + ["about", "Engineering discussions"], + ], + ); + let (resolver, _requests, server) = counting_resolver(response).await; + + let info = resolver.resolve(id).await.expect("should resolve"); + assert_eq!(info.description.as_deref(), Some("Engineering discussions")); + server.abort(); + } + + /// TTL expiry triggers a background refresh; stale value is served immediately + /// while the refresh runs (stale-while-revalidate). + #[tokio::test] + async fn test_channel_resolver_ttl_serves_stale_on_expiry() { + use std::sync::atomic::Ordering; + + let id = Uuid::new_v4(); + let response = channel_metadata_response(id, &[["name", "buzz-dev"], ["t", "stream"]]); + let (resolver, requests, server) = counting_resolver(response).await; + + // Warm the cache. + let info = resolver.resolve(id).await.expect("should resolve"); + assert_eq!(info.name, "buzz-dev"); + let initial_requests = requests.load(Ordering::SeqCst); + + // Manually expire the entry by setting refresh_after to the past. + { + let mut cache = resolver.cache.write().unwrap(); + if let Some(entry) = cache.get_mut(&id) { + entry.refresh_after = + tokio::time::Instant::now() - std::time::Duration::from_secs(1); + } + } + + // resolve() returns stale immediately and spawns background refresh. + let stale = resolver.resolve(id).await.expect("stale serve"); + assert_eq!( + stale.name, "buzz-dev", + "stale value served while refresh runs" + ); + + // Give background refresh time to complete. + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + let after_refresh = requests.load(Ordering::SeqCst); + assert!( + after_refresh > initial_requests, + "background refresh must have fired" + ); + server.abort(); + } + + /// On a failed TTL refresh the entry's next_refresh_at is bumped by backoff, + /// preventing immediate re-retry on every subsequent consumer. + #[tokio::test] + async fn test_channel_resolver_ttl_backoff_on_failed_refresh() { + use std::sync::atomic::Ordering; + + let id = Uuid::new_v4(); + let response = channel_metadata_response(id, &[["name", "buzz-dev"], ["t", "stream"]]); + // Use a new resolver seeded with the channel at an already-expired TTL. + let (resolver, requests, server) = counting_resolver(json!([])).await; + + // Manually seed the cache with an expired entry pointing at the non-responding server. + { + let mut cache = resolver.cache.write().unwrap(); + cache.insert( + id, + CachedChannelInfo { + info: PromptChannelInfo { + name: "buzz-dev".to_string(), + channel_type: "stream".to_string(), + description: None, + }, + refresh_after: tokio::time::Instant::now() - std::time::Duration::from_secs(1), + next_refresh_at: tokio::time::Instant::now() + - std::time::Duration::from_secs(1), + }, + ); + } + + // First resolve: serves stale, spawns background refresh (which fails). + let info = resolver.resolve(id).await.expect("stale serve"); + assert_eq!(info.name, "buzz-dev"); + + // Give background refresh time to fail and set backoff. + // fetch_with_retry sleeps CONTEXT_FETCH_RETRY_DELAY (500 ms) between its two + // attempts, so we must wait longer than that before inspecting the cache. + tokio::time::sleep(std::time::Duration::from_millis(1500)).await; + + // Verify: next_refresh_at should have been bumped by backoff. + { + let cache = resolver.cache.read().unwrap(); + let entry = cache.get(&id).expect("entry still present"); + // next_refresh_at should be in the future after backoff. + assert!( + entry.next_refresh_at > tokio::time::Instant::now(), + "next_refresh_at must be in the future after failed refresh backoff" + ); + } + + let after = requests.load(Ordering::SeqCst); + assert!( + after >= 2, + "failed refresh should have attempted (fetch_with_retry = 2 attempts)" + ); + server.abort(); + let _ = server; + } } diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index 5c960de202..085b256bfa 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -1000,6 +1000,20 @@ pub struct ContextMessage { pub struct PromptChannelInfo { pub name: String, pub channel_type: String, + /// Channel description from the kind-39000 `about` tag, if present. + pub description: Option, +} + +/// A resolved canvas revision pointer for delivery in `[Context]`. +/// +/// Carries the event ID and RFC3339 last-modified timestamp so agents can +/// detect stale cached content and know when to re-fetch. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CanvasPointer { + /// The Nostr event ID (hex) of the latest canvas event. + pub event_id: String, + /// RFC3339 timestamp of `created_at` with Z suffix. + pub timestamp: String, } /// Minimal profile fields needed to label users in ACP prompts. @@ -1242,6 +1256,7 @@ fn format_context_hints( is_dm: bool, has_conversation_context: bool, reply_anchor: Option<&str>, + canvas_pointer: Option<&CanvasPointer>, ) -> String { let channel_display = match channel_info { Some(ci) => format!("{} (#{channel_id})", ci.name), @@ -1291,9 +1306,11 @@ fn format_context_hints( let mut s = format!( "[Context]\n\ Scope: thread\n\ - Channel: {channel_display}\n\ - Thread root: {root}" + Channel: {channel_display}" ); + // Append description for non-DM channels (including threads). + append_channel_description(&mut s, channel_info); + s.push_str(&format!("\nThread root: {root}")); if let Some(ref parent) = thread_tags.parent_event_id { if parent != root { s.push_str(&format!("\nParent: {parent}")); @@ -1303,21 +1320,105 @@ fn format_context_hints( if let Some(event_id) = reply_anchor { append_reply_instruction(&mut s, event_id); } + // Canvas pointer for non-DM turns. + append_canvas_pointer(&mut s, canvas_pointer, &channel_id.to_string()); s } else { let mut s = format!( "[Context]\n\ Scope: channel\n\ - Channel: {channel_display}\n\ - Hint: Use `buzz messages get --channel ` for recent messages if needed." + Channel: {channel_display}" + ); + // Append description for non-DM channels. + append_channel_description(&mut s, channel_info); + s.push_str( + "\nHint: Use `buzz messages get --channel ` for recent messages if needed.", ); if let Some(event_id) = reply_anchor { append_new_thread_reply_instruction(&mut s, event_id); } + // Canvas pointer for non-DM turns. + append_canvas_pointer(&mut s, canvas_pointer, &channel_id.to_string()); s } } +/// Maximum byte length of a channel description rendered into `[Context]`. +/// +/// Limits prompt bloat from unusually long descriptions; a raw embedded newline +/// in a description must not be able to spoof another `[Context]` field, so +/// multiline text is collapsed to single-space-joined lines before truncation. +const MAX_DESCRIPTION_LEN: usize = 500; + +/// Append a `Description: …` line to a `[Context]` block when non-empty. +/// +/// Collapses internal newlines (any `\r\n`, `\r`, or `\n`) to a single space +/// so a multi-line description cannot inject a fake `[Context]` field line. +/// Truncates at [`MAX_DESCRIPTION_LEN`] bytes with a `…` marker. +pub(crate) fn append_channel_description(s: &mut String, channel_info: Option<&PromptChannelInfo>) { + let desc = match channel_info.and_then(|ci| ci.description.as_deref()) { + Some(d) if !d.is_empty() => d, + _ => return, + }; + // Collapse newlines to spaces so the description can never spoof another field. + let collapsed: String = desc + .split(['\n', '\r']) + .map(str::trim) + .filter(|s| !s.is_empty()) + .collect::>() + .join(" "); + if collapsed.is_empty() { + return; + } + // Truncate at MAX_DESCRIPTION_LEN character boundary (not byte boundary to avoid + // splitting multi-byte sequences). We use char indices for safety. + let truncated = if collapsed.chars().count() > MAX_DESCRIPTION_LEN { + let end = collapsed + .char_indices() + .nth(MAX_DESCRIPTION_LEN) + .map(|(i, _)| i) + .unwrap_or(collapsed.len()); + format!("{}…", &collapsed[..end]) + } else { + collapsed + }; + s.push_str(&format!("\nDescription: {truncated}")); +} + +/// Append the canvas pointer block to a `[Context]` block when present. +/// +/// Renders: `Canvas revision (event ID): `, `Last modified: `, +/// and `Fetch current content with: buzz canvas get --channel `. +/// Absent (no canvas or fetch failure with no prior value) = nothing appended. +pub(crate) fn append_canvas_pointer( + s: &mut String, + pointer: Option<&CanvasPointer>, + channel_uuid: &str, +) { + let Some(p) = pointer else { return }; + s.push_str(&format!( + "\nCanvas revision (event ID): {}\nLast modified: {}\nFetch current content with: buzz canvas get --channel {channel_uuid}", + p.event_id, p.timestamp + )); +} + +/// Alias used by the `initial_message` dispatch in pool.rs. +pub(crate) fn append_description_for_initial_message( + s: &mut String, + channel_info: Option<&PromptChannelInfo>, +) { + append_channel_description(s, channel_info); +} + +/// Alias used by the `initial_message` dispatch in pool.rs. +pub(crate) fn append_canvas_pointer_for_initial_message( + s: &mut String, + pointer: Option<&CanvasPointer>, + channel_uuid: &str, +) { + append_canvas_pointer(s, pointer, channel_uuid); +} + /// Format a conversation context section (thread or DM). fn format_conversation_context( ctx: &ConversationContext, @@ -1370,13 +1471,12 @@ pub struct FormatPromptArgs<'a> { pub system_prompt: Option<&'a str>, /// Team instructions for legacy agents, rendered after `[System]`. pub team_instructions: Option<&'a str>, - /// Rendered `[Channel Canvas]` metadata section for legacy agents. + /// Canvas revision pointer for this turn's `[Context]` block. /// - /// For modern agents (protocol_version >= 2) the section is delivered via - /// the system role in session/new; omit here to avoid duplication. - /// For legacy agents it rides in the user message on every turn of the - /// session, alongside `[Base]`/`[System]`/`[Agent Memory — core]`. - pub agent_canvas: Option<&'a str>, + /// Derived from `CanvasRevisionCache::resolve_for_turn()` — already + /// tri-state resolved: `Some` = present/stale-served, `None` = confirmed + /// absent or first-fetch failure. DM turns always pass `None`. + pub canvas_pointer: Option<&'a CanvasPointer>, } /// Format the `[Base]` section for the base prompt. @@ -1456,10 +1556,6 @@ pub fn format_prompt(batch: &FlushBatch, args: &FormatPromptArgs<'_>) -> Vec) -> Vec, } pub(crate) fn channel_type_from_tags(tags: &[serde_json::Value]) -> String { @@ -172,7 +174,7 @@ pub(crate) fn merge_discovered_channels( channel_uuids: Vec, meta_events: &serde_json::Value, ) -> HashMap { - let mut meta_map: HashMap = HashMap::new(); + let mut meta_map: HashMap)> = HashMap::new(); let mut archived: std::collections::HashSet = std::collections::HashSet::new(); if let Some(arr) = meta_events.as_array() { for ev in arr { @@ -183,11 +185,13 @@ pub(crate) fn merge_discovered_channels( let mut d_val = None; let mut name = None; let mut is_archived = false; + let mut description = None; for tag in tags { if let Some(arr) = tag.as_array() { match arr.first().and_then(|v| v.as_str()) { Some("d") => d_val = arr.get(1).and_then(|v| v.as_str()), Some("name") => name = arr.get(1).and_then(|v| v.as_str()), + Some("about") => description = arr.get(1).and_then(|v| v.as_str()), Some("archived") => { is_archived = arr.get(1).and_then(|v| v.as_str()) == Some("true") } @@ -203,7 +207,11 @@ pub(crate) fn merge_discovered_channels( } let ch_name = name.unwrap_or("unknown").to_string(); let ch_type = channel_type_from_tags(tags); - meta_map.insert(uuid, (ch_name, ch_type)); + let ch_desc = description + .map(|s| s.trim()) + .filter(|s| !s.is_empty()) + .map(str::to_string); + meta_map.insert(uuid, (ch_name, ch_type, ch_desc)); } } } @@ -214,10 +222,17 @@ pub(crate) fn merge_discovered_channels( if archived.contains(&uuid) { continue; } - let (name, channel_type) = meta_map + let (name, channel_type, description) = meta_map .remove(&uuid) - .unwrap_or_else(|| ("unknown".to_string(), "unknown".to_string())); - map.insert(uuid, ChannelInfo { name, channel_type }); + .unwrap_or_else(|| ("unknown".to_string(), "unknown".to_string(), None)); + map.insert( + uuid, + ChannelInfo { + name, + channel_type, + description, + }, + ); } map } diff --git a/desktop/src/features/agents/ui/agentSessionTranscriptHelpers.test.mjs b/desktop/src/features/agents/ui/agentSessionTranscriptHelpers.test.mjs index f0f4cbf36d..70056084d7 100644 --- a/desktop/src/features/agents/ui/agentSessionTranscriptHelpers.test.mjs +++ b/desktop/src/features/agents/ui/agentSessionTranscriptHelpers.test.mjs @@ -358,10 +358,10 @@ test("parseSystemPromptSections pins the realistic Workspace+Base+System+Core ha // ── Channel Canvas extraction ───────────────────────────────────────────────── -test("parseSystemPromptSections pins the full Base+System+Core+Canvas harness shape", () => { - // with_canvas() appends "\n\n[Channel Canvas]\n{metadata}" after the core block. - // render_canvas_section() emits the revision event ID, last-modified timestamp, - // and fetch command — never the canvas body. All four sections are extracted in order. +test("parseSystemPromptSections extracts channel canvas from historical session prompts", () => { + // Historical sessions (pre–channel-context-delivery rework) carried the canvas + // in the system prompt via with_canvas(). This test ensures historical sessions + // still parse correctly — the extractor is retained for backward compat. const framed = [ "[Base]", "You are an assistant.", @@ -389,6 +389,33 @@ test("parseSystemPromptSections pins the full Base+System+Core+Canvas harness sh ]); }); +test("parseSystemPromptSections handles new-session shape without channel canvas (post-rework)", () => { + // New sessions (post–channel-context-delivery rework) no longer include canvas + // in the system prompt — canvas is delivered per-turn in [Context] instead. + // This test pins that the new shape extracts correctly and no canvas section appears. + const framed = [ + "[Base]", + "You are an assistant.", + "", + "[System]", + "Persona instructions.", + "", + "[Agent Memory — core]", + "I am Duncan.", + ].join("\n"); + const sections = parseSystemPromptSections(framed); + assert.deepEqual(sections, [ + { title: "Base", body: "You are an assistant." }, + { title: "System", body: "Persona instructions." }, + { title: "Core Memory", body: "I am Duncan." }, + ]); + // Confirm no canvas section present. + assert.ok( + !sections.some((s) => s.title === "Channel Canvas"), + "new-session shape must not contain a Channel Canvas section", + ); +}); + test("parseSystemPromptSections extracts canvas when no Base/System/Core present (canvas-only)", () => { const framed = ["[Channel Canvas]", "Canvas only."].join("\n"); const sections = parseSystemPromptSections(framed); diff --git a/desktop/src/features/agents/ui/agentSessionTranscriptHelpers.ts b/desktop/src/features/agents/ui/agentSessionTranscriptHelpers.ts index 09a2bb31cf..475a07d8a3 100644 --- a/desktop/src/features/agents/ui/agentSessionTranscriptHelpers.ts +++ b/desktop/src/features/agents/ui/agentSessionTranscriptHelpers.ts @@ -60,22 +60,31 @@ export function parsePromptText(text: string): { * `Team Instructions`/`Core Memory`/`Channel Canvas` sub-sections * deterministically. * - * The harness composes the value in order: - * `[Base]\n{base}\n\n[System]\n{persona}\n\n[Team Instructions]\n{team}\n\n[Agent Memory — core]\n{core}\n\n[Channel Canvas]\n{canvas}` - * with any section omitted when absent. Extraction runs in reverse producer - * order so that each `lastIndexOf` search operates on the full input and each - * extraction boundary is unambiguous. + * **Current system-prompt shape (post–channel-context-delivery rework):** + * `[Workspace]\n{workspace}\n\n[Base]\n{base}\n\n[System]\n{persona}\n\n[Team Instructions]\n{team}\n\n[Agent Memory — core]\n{core}` * - * Five extraction passes: + * The `[Channel Canvas]` section is no longer appended to new sessions. + * Canvas revisions are now delivered per-turn in the `[Context]` block of the + * user message. The extractor below is intentionally retained: historical + * session recordings captured before this change still carry the section in + * their system-prompt field and must continue to parse correctly. Graceful + * degradation when the section is absent is the default — the canvas + * extraction pass simply yields `null` / no section when `[Channel Canvas]` + * is not found. * - * 1. **Canvas** (`[Channel Canvas]`): appended last by `with_canvas()`. + * Extraction runs in reverse producer order so that each `lastIndexOf` search + * operates on the full input and each extraction boundary is unambiguous. + * + * Five extraction passes (canvas included for historical compat): + * + * 1. **Canvas** (`[Channel Canvas]`): present only in historical sessions. * - Start-of-string: canvas-only input. * - Appended frame (`\n\n[Channel Canvas]\n`): blank-line separator used by - * `with_canvas()`; LAST occurrence guards against an embedded header in a - * persona body (single preceding newline only). + * the now-removed `with_canvas()`; LAST occurrence guards against an + * embedded header in a persona body (single preceding newline only). * - * 2. **Core** (`[Agent Memory — core]`): appended before canvas by `with_core()`. - * Same two cases, same last-occurrence guard. + * 2. **Core** (`[Agent Memory — core]`): appended before canvas (or last in + * new sessions) by `with_core()`. Same two cases, same last-occurrence guard. * * 3. **Team Instructions** (`[Team Instructions]`): appended before core by * `with_team()` in `buzz-acp/src/pool.rs`. Same two cases (start-of-string From 20ea6af8fef139d3959db9c7b612a86d23db8012 Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Mon, 3 Aug 2026 12:46:33 -0400 Subject: [PATCH 2/6] fix(acp): tri-state canvas parser, single turn snapshot, per-key refresh guard, query_once MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Five fixes addressing pass-1 review findings: 1. canvas_pointer_from_query_response now returns CanvasFetchResult directly. Validation rejections (malformed JSON, bad sig, wrong kind, wrong channel tag, invalid timestamp) map to Failed — stale cache entry is preserved. Only confirmed-empty array or valid blank content maps to Absent. 2. run_prompt_task resolves metadata and canvas exactly once per turn; the single snapshot (turn_channel_info, is_dm_turn, canvas_pointer) feeds both initial_message and the batch prompt. initial_message uses format_context_hints with is_dm_turn so DM turns emit Scope: dm, no description, no canvas. 3. ChannelInfoResolver::resolve atomically claims an in_flight slot before spawning background_refresh. Concurrent callers on an expired entry skip the spawn rather than stampeding the relay. 4. RestClient::query_once issues exactly one HTTP request with no retry loop. fetch_canvas_pointer now calls query_once instead of query. 5. All clippy errors resolved: unused CanvasRevisionCache::get removed, doc-comment empty lines fixed, overindented doc-list items corrected, unused bindings removed. Tests added: stale-preservation on malformed/tampered/wrong-channel parse failures; concurrent expired-entry resolve asserts one refresh spawned; query_once request-count on 503 asserts no retry; initial_message context composition for channel turn (Scope: channel, description, canvas) and DM turn (Scope: dm, no description, no canvas). Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- crates/buzz-acp/src/pool.rs | 650 +++++++++++++++++++++++++++-------- crates/buzz-acp/src/queue.rs | 19 +- crates/buzz-acp/src/relay.rs | 37 ++ 3 files changed, 539 insertions(+), 167 deletions(-) diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index afb554af3c..574f143b1d 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -142,9 +142,9 @@ impl SessionState { /// /// - `Present` — a canvas event was found; carry the pointer into `[Context]`. /// - `Absent` — relay confirms no canvas (or a blank/deleted canvas). Clears -/// any previously cached pointer for the channel. +/// any previously cached pointer for the channel. /// - `Failed` — transport timeout, REST error, or parse failure. The caller -/// must serve the last-known cached value (stale-on-failure). +/// must serve the last-known cached value (stale-on-failure). #[derive(Debug)] pub enum CanvasFetchResult { Present(crate::queue::CanvasPointer), @@ -171,25 +171,12 @@ impl CanvasRevisionCache { Self::default() } - /// Look up the cached pointer for `channel_id`. - /// - /// Returns `None` when there is no entry (never fetched) or the entry is - /// confirmed absent. Callers that need to distinguish "never fetched" from - /// "confirmed absent" should use `get_raw`. - pub fn get(&self, channel_id: &Uuid) -> Option { - self.inner - .read() - .unwrap_or_else(|p| p.into_inner()) - .get(channel_id) - .and_then(|v| v.clone()) - } - /// Resolve the canvas pointer for the current turn, updating the cache. /// /// - `Present(p)` → cache `Some(p)`, return `Some(p)`. /// - `Absent` → cache `None`, return `None`. /// - `Failed` → cache unchanged; return last-known value or `None` on - /// first-fetch failure. + /// first-fetch failure. pub fn resolve_for_turn( &self, channel_id: &Uuid, @@ -520,11 +507,6 @@ pub enum PromptOutcome { CancelDrainTimeout(Duration), } -/// Immutable config subset shared (via `Arc`) by all spawned prompt tasks. -/// -/// Built once from `Config` at startup. Avoids cloning the full config -/// into every task. - /// Per-entry TTL for channel metadata cache entries (~5 minutes). const CHANNEL_INFO_TTL: std::time::Duration = std::time::Duration::from_secs(300); @@ -561,6 +543,11 @@ struct CachedChannelInfo { #[derive(Debug, Clone)] pub struct ChannelInfoResolver { cache: std::sync::Arc>>, + /// Per-channel in-flight marker: a channel UUID is inserted before spawning a + /// background refresh and removed when it completes (success or failure). + /// Prevents an expired entry from spawning duplicate refresh tasks when multiple + /// callers see the same expired timestamps before the first task finishes. + in_flight: std::sync::Arc>>, rest_client: RestClient, } @@ -589,6 +576,7 @@ impl ChannelInfoResolver { .collect(); Self { cache: std::sync::Arc::new(std::sync::RwLock::new(cache)), + in_flight: std::sync::Arc::new(std::sync::Mutex::new(std::collections::HashSet::new())), rest_client, } } @@ -604,11 +592,18 @@ impl ChannelInfoResolver { .and_then(|c| c.get(&channel_id).cloned()) { if now >= cached.refresh_after && now >= cached.next_refresh_at { - // Stale — spawn background refresh; serve stale immediately. - let resolver = self.clone(); - tokio::spawn(async move { - resolver.background_refresh(channel_id).await; - }); + // Stale — claim the in-flight slot; if already claimed, skip spawning. + let claimed = self + .in_flight + .lock() + .map(|mut s| s.insert(channel_id)) + .unwrap_or(false); + if claimed { + let resolver = self.clone(); + tokio::spawn(async move { + resolver.background_refresh(channel_id).await; + }); + } } return Some(cached.info); } @@ -635,6 +630,9 @@ impl ChannelInfoResolver { /// On failure: advance `next_refresh_at` by `CHANNEL_INFO_REFRESH_BACKOFF` /// so the next consumer does not immediately re-trigger the retry sequence. /// On success: reset the TTL window. + /// + /// Clears the per-channel `in_flight` marker on return, allowing the next + /// expiry cycle to spawn a fresh refresh. async fn background_refresh(&self, channel_id: Uuid) { match fetch_channel_info(channel_id, &self.rest_client).await { Some(info) => { @@ -660,9 +658,17 @@ impl ChannelInfoResolver { } } } + // Release the in-flight slot so the next expiry cycle can refresh again. + if let Ok(mut s) = self.in_flight.lock() { + s.remove(&channel_id); + } } } +/// Immutable config subset shared (via `Arc`) by all spawned prompt tasks. +/// +/// Built once from `Config` at startup. Avoids cloning the full config +/// into every task. pub struct PromptContext { pub mcp_servers: Vec, pub initial_message: Option, @@ -1012,7 +1018,7 @@ const UNKNOWN_CHANNEL_NAME: &str = "unknown"; /// resolver's cached entry, so a renamed channel keeps its old suffix until the /// process restarts. An agent rename lands on the next spawn (the desktop /// restart badge covers it — see `spawn_config_hash`). - +/// /// Create a new ACP session via `session_new_full()`, populate model capabilities /// on the agent (first session only), and apply `desired_model` if set. /// @@ -1604,45 +1610,41 @@ pub async fn run_prompt_task( } } - // Channel metadata — resolve once per turn for title (new sessions only) and - // the per-turn canvas fetch. DM check fails closed: a channel whose type - // cannot be determined is treated as a DM. - let mut title_channel: Option = None; - if let PromptSource::Channel(cid) = &source { - let is_new_channel_session = !agent.state.sessions.contains_key(cid); - let needs_title = is_new_channel_session && ctx.session_title.is_some(); - if needs_title { - if let Some(info) = ctx.channel_info.resolve(*cid).await { - let is_dm = info.channel_type == "dm"; - title_channel = (!is_dm && info.name != UNKNOWN_CHANNEL_NAME).then_some(info.name); - } + // Channel metadata — resolved ONCE per turn. All consumers use this snapshot: + // session title, canvas DM gate, initial_message context, and batch prompt context. + // DM check fails closed: unknown channel type → treat as DM (no canvas, no description). + let (turn_channel_info, is_dm_turn, title_channel) = match &source { + PromptSource::Channel(cid) => { + let info = ctx.channel_info.resolve(*cid).await; + let is_dm = info + .as_ref() + .map(|i| i.channel_type == "dm") + .unwrap_or(true); // fail closed on unknown + let is_new_channel_session = !agent.state.sessions.contains_key(cid); + let needs_title = is_new_channel_session && ctx.session_title.is_some(); + let title = if needs_title && !is_dm { + info.as_ref() + .filter(|i| i.name != UNKNOWN_CHANNEL_NAME) + .map(|i| i.name.clone()) + } else { + None + }; + (info, is_dm, title) } - } + PromptSource::Heartbeat => (None, false, None), + }; // Per-turn canvas fetch: resolve for the current turn and update the - // process-lifetime cache. Canvas is only fetched for channel turns (not - // heartbeats). DMs never receive canvas material — fail closed when channel - // type is unknown. + // process-lifetime cache. Canvas is only fetched for non-DM channel turns. // // A4: single-attempt, 3 s timeout, stale-on-failure. Runs before session // creation so the pointer is available for initial_message. let canvas_pointer: Option = match &source { - PromptSource::Channel(cid) => { - // Determine DM status before fetching canvas. - let is_dm = ctx - .channel_info - .resolve(*cid) - .await - .map(|info| info.channel_type == "dm") - .unwrap_or(true); // fail closed on unknown - if is_dm { - None - } else { - let result = fetch_canvas_pointer(*cid, &ctx.rest_client).await; - ctx.canvas_cache.resolve_for_turn(cid, result) - } + PromptSource::Channel(cid) if !is_dm_turn => { + let result = fetch_canvas_pointer(*cid, &ctx.rest_client).await; + ctx.canvas_cache.resolve_for_turn(cid, result) } - PromptSource::Heartbeat => None, + _ => None, }; // The core section to fold into the system prompt for this turn's session. @@ -1786,25 +1788,17 @@ pub async fn run_prompt_task( initial_msg, ); // Prepend [Context] block (description + canvas pointer) to initial_message. + // Uses the same turn snapshot (turn_channel_info, is_dm_turn, canvas_pointer) + // as the subsequent batch prompt — one consistent view per inbound turn (A2). let init_msg = { - let channel_info = ctx.channel_info.resolve(*cid).await; - let mut ctx_block = String::from("[Context]\nScope: channel"); - ctx_block.push_str(&format!( - "\nChannel: {} (#{})", - channel_info - .as_ref() - .map(|i| i.name.as_str()) - .unwrap_or("unknown"), - cid - )); - crate::queue::append_description_for_initial_message( - &mut ctx_block, - channel_info.as_ref(), - ); - crate::queue::append_canvas_pointer_for_initial_message( - &mut ctx_block, + let ctx_block = crate::queue::format_context_hints( + *cid, + turn_channel_info.as_ref(), + &crate::queue::ThreadTags::default(), + is_dm_turn, + false, + None, canvas_pointer.as_ref(), - &cid.to_string(), ); format!("{ctx_block}\n\n{init_msg}") }; @@ -1940,8 +1934,9 @@ pub async fn run_prompt_task( vec![text] } else if let Some(ref b) = batch { // Build prompt from batch with context enrichment. - // Try startup cache first; lazy-fetch via REST for dynamic channels. - let channel_info = ctx.channel_info.resolve(b.channel_id).await; + // Use the per-turn metadata snapshot (resolved once at turn start) rather + // than resolving again — ensures batch and initial_message see the same info. + let channel_info = turn_channel_info.clone(); let conversation_context = if ctx.context_message_limit > 0 { fetch_conversation_context(b, &channel_info, &ctx).await @@ -2515,9 +2510,9 @@ pub(crate) async fn fetch_channel_info( /// /// - `Present(pointer)` — a valid, non-blank canvas event was found. /// - `Absent` — relay confirmed no event, or the latest event has blank content -/// (cleared canvas); older revisions are NOT resurrected. +/// (cleared canvas); older revisions are NOT resurrected. /// - `Failed` — transport timeout, REST error, or parse/verification error. -/// Callers must serve stale on `Failed`. +/// Callers must serve stale on `Failed`. /// /// Single-attempt with a 3-second timeout — never wrapped in `fetch_with_retry`. /// Accepted latency bound: at most +3 s per turn on a degraded relay. @@ -2534,7 +2529,7 @@ async fn fetch_canvas_pointer(channel_id: Uuid, rest: &RestClient) -> CanvasFetc let json = match tokio::time::timeout( CANVAS_FETCH_TIMEOUT, - rest.query(std::slice::from_ref(&filter)), + rest.query_once(std::slice::from_ref(&filter)), ) .await { @@ -2570,29 +2565,33 @@ async fn fetch_canvas_pointer(channel_id: Uuid, rest: &RestClient) -> CanvasFetc } }; - match canvas_pointer_from_query_response(events, &channel_id.to_string()) { - Some(pointer) => CanvasFetchResult::Present(pointer), - None => CanvasFetchResult::Absent, - } + canvas_pointer_from_query_response(events, &channel_id.to_string()) } -/// Parse a canvas query response array and return a `CanvasPointer`. +/// Parse a canvas query response array and return a `CanvasFetchResult`. /// /// Extracted as a pure function so tests can exercise the parsing/validation /// logic without async machinery or relay connectivity. /// -/// Returns `None` on: empty array, blank content, malformed/partial event JSON -/// (requires a complete, structurally valid Nostr event), or an out-of-range -/// `created_at` timestamp. A `None` result signals `Absent` (not `Failed`) — -/// the relay responded but no usable canvas exists. +/// Returns: +/// - `Present(pointer)` — a valid, non-blank canvas event was found. +/// - `Absent` — empty array (confirmed no canvas) or blank/deleted content. +/// - `Failed` — malformed/partial event JSON, signature verification failure, +/// wrong kind, wrong channel tag, or out-of-range timestamp. The caller must +/// preserve any cached stale pointer rather than clearing it. pub(crate) fn canvas_pointer_from_query_response( events: &[serde_json::Value], channel_uuid: &str, -) -> Option { - let raw = events.first()?; +) -> CanvasFetchResult { + // Empty array: relay confirmed no canvas event exists. + let raw = match events.first() { + Some(v) => v, + None => return CanvasFetchResult::Absent, + }; // Deserialise as a complete Nostr Event. Partial objects (missing pubkey, // sig, kind, or tags) are rejected here rather than trusted implicitly. + // These are parse/structural failures → Failed (preserve stale). let event = match serde_json::from_value::(raw.clone()) { Ok(ev) => ev, Err(err) => { @@ -2600,38 +2599,41 @@ pub(crate) fn canvas_pointer_from_query_response( target: "canvas::fetch", channel = %channel_uuid, %err, - "canvas query returned a malformed event — emitting no pointer", + "canvas query returned a malformed event — serving stale", ); - return None; + return CanvasFetchResult::Failed; } }; // Verify the event's id and signature agree with its content. // A structurally complete but tampered event must not supply trusted metadata. + // Verification failure → Failed (preserve stale). if let Err(err) = event.verify() { tracing::warn!( target: "canvas::fetch", channel = %channel_uuid, %err, - "canvas event failed signature verification — emitting no pointer", + "canvas event failed signature verification — serving stale", ); - return None; + return CanvasFetchResult::Failed; } // Validate kind: must be KIND_CANVAS (40100). + // Wrong kind → Failed (preserve stale). if event.kind != nostr::Kind::Custom(buzz_core::kind::KIND_CANVAS as u16) { tracing::warn!( target: "canvas::fetch", channel = %channel_uuid, kind = %event.kind.as_u16(), - "canvas event has unexpected kind — emitting no pointer", + "canvas event has unexpected kind — serving stale", ); - return None; + return CanvasFetchResult::Failed; } // Validate h-tag: must carry the channel UUID we queried. // The REST boundary filters by #h, but we verify here to prevent a // misbehaving relay from injecting a different channel's canvas. + // Wrong h-tag → Failed (preserve stale). let h_tag_matches = event.tags.iter().any(|tag| { let v = tag.as_slice(); v.len() >= 2 && v[0] == "h" && v[1] == channel_uuid @@ -2640,19 +2642,20 @@ pub(crate) fn canvas_pointer_from_query_response( tracing::warn!( target: "canvas::fetch", channel = %channel_uuid, - "canvas event is missing expected h-tag — emitting no pointer", + "canvas event is missing expected h-tag — serving stale", ); - return None; + return CanvasFetchResult::Failed; } // Blank content means the canvas was cleared; do not fall back to older events. + // Confirmed cleared canvas → Absent (clears any cached pointer). if event.content.trim().is_empty() { tracing::debug!( target: "canvas::fetch", channel = %channel_uuid, - "latest canvas event has blank content — emitting no pointer (Absent)" + "latest canvas event has blank content — canvas absent" ); - return None; + return CanvasFetchResult::Absent; } let id = event.id.to_hex(); @@ -2661,15 +2664,16 @@ pub(crate) fn canvas_pointer_from_query_response( // Use checked conversion: a u64 that exceeds i64::MAX (e.g. Timestamp::max()) // wraps silently with `as i64`, producing a negative value that chrono would // accept as a date in 1969. Reject out-of-range values explicitly instead. + // Out-of-range timestamp → Failed (preserve stale). let ts_secs = match i64::try_from(event.created_at.as_secs()) { Ok(s) => s, Err(_) => { tracing::warn!( target: "canvas::fetch", channel = %channel_uuid, - "canvas event created_at overflows i64 — emitting no pointer", + "canvas event created_at overflows i64 — serving stale", ); - return None; + return CanvasFetchResult::Failed; } }; let timestamp = match chrono::DateTime::from_timestamp(ts_secs, 0) { @@ -2679,9 +2683,9 @@ pub(crate) fn canvas_pointer_from_query_response( target: "canvas::fetch", channel = %channel_uuid, ts_secs, - "canvas event has out-of-range created_at — emitting no pointer", + "canvas event has out-of-range created_at — serving stale", ); - return None; + return CanvasFetchResult::Failed; } }; @@ -2691,7 +2695,7 @@ pub(crate) fn canvas_pointer_from_query_response( event_id = %id, "resolved canvas revision pointer for [Context]" ); - Some(crate::queue::CanvasPointer { + CanvasFetchResult::Present(crate::queue::CanvasPointer { event_id: id, timestamp, }) @@ -6515,7 +6519,10 @@ mod tests { let ev = make_canvas_event_value("# Team instructions\nBe helpful."); let id = ev["id"].as_str().unwrap().to_string(); let result = canvas_pointer_from_query_response(&[ev], CHANNEL_UUID); - let pointer = result.expect("expected Some"); + let pointer = match result { + CanvasFetchResult::Present(p) => p, + other => panic!("expected Present, got {other:?}"), + }; assert_eq!(pointer.event_id, id, "pointer must carry the event id"); // Timestamp must use Z suffix, not +00:00 assert!( @@ -6529,32 +6536,38 @@ mod tests { } #[test] - fn test_canvas_pointer_from_query_response_empty_array_returns_none() { + fn test_canvas_pointer_from_query_response_empty_array_returns_absent() { let result = canvas_pointer_from_query_response(&[], CHANNEL_UUID); - assert!(result.is_none()); + assert!( + matches!(result, CanvasFetchResult::Absent), + "empty array must be Absent (confirmed no canvas)" + ); } #[test] - fn test_canvas_pointer_from_query_response_blank_content_returns_none() { + fn test_canvas_pointer_from_query_response_blank_content_returns_absent() { let ev = make_canvas_event_value(" "); let result = canvas_pointer_from_query_response(&[ev], CHANNEL_UUID); assert!( - result.is_none(), - "blank content must return None (cleared canvas)" + matches!(result, CanvasFetchResult::Absent), + "blank content must be Absent (cleared canvas)" ); } #[test] - fn test_canvas_pointer_from_query_response_empty_content_returns_none() { + fn test_canvas_pointer_from_query_response_empty_content_returns_absent() { let ev = make_canvas_event_value(""); let result = canvas_pointer_from_query_response(&[ev], CHANNEL_UUID); - assert!(result.is_none()); + assert!( + matches!(result, CanvasFetchResult::Absent), + "empty content must be Absent (cleared canvas)" + ); } /// A bare JSON object with a plausible-looking id but missing pubkey/sig/kind/tags - /// must be rejected — not silently accepted with partial metadata. + /// must be rejected as Failed — it is a structural parse error, not confirmed absence. #[test] - fn test_canvas_pointer_from_query_response_partial_object_returns_none() { + fn test_canvas_pointer_from_query_response_partial_object_returns_failed() { let partial = serde_json::json!({ "id": "a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2", "created_at": 1705312200_i64, @@ -6562,15 +6575,15 @@ mod tests { }); let result = canvas_pointer_from_query_response(&[partial], CHANNEL_UUID); assert!( - result.is_none(), - "partial event object (missing pubkey/sig/kind/tags) must return None" + matches!(result, CanvasFetchResult::Failed), + "partial event object (missing pubkey/sig/kind/tags) must be Failed — not Absent" ); } /// A JSON object that looks like an event but has `created_at` as a string - /// must be rejected — the nostr::Event parser enforces integer type. + /// must be rejected as Failed — the nostr::Event parser enforces integer type. #[test] - fn test_canvas_pointer_from_query_response_string_timestamp_returns_none() { + fn test_canvas_pointer_from_query_response_string_timestamp_returns_failed() { let keys = Keys::generate(); let h_tag = Tag::parse(["h", CHANNEL_UUID]).expect("h tag"); let mut ev = serde_json::to_value( @@ -6584,15 +6597,15 @@ mod tests { ev["created_at"] = serde_json::Value::String("2026-03-15T16:30:00+00:00".into()); let result = canvas_pointer_from_query_response(&[ev], CHANNEL_UUID); assert!( - result.is_none(), - "string created_at must be rejected by nostr::Event deserialiser" + matches!(result, CanvasFetchResult::Failed), + "string created_at must be Failed — nostr::Event deserialiser rejects it" ); } /// A JSON object that looks like an event but is missing `created_at` - /// must be rejected — nostr::Event requires the field. + /// must be rejected as Failed — nostr::Event requires the field. #[test] - fn test_canvas_pointer_from_query_response_missing_timestamp_returns_none() { + fn test_canvas_pointer_from_query_response_missing_timestamp_returns_failed() { let keys = Keys::generate(); let h_tag = Tag::parse(["h", CHANNEL_UUID]).expect("h tag"); let mut ev = serde_json::to_value( @@ -6605,17 +6618,17 @@ mod tests { ev.as_object_mut().unwrap().remove("created_at"); let result = canvas_pointer_from_query_response(&[ev], CHANNEL_UUID); assert!( - result.is_none(), - "missing created_at must be rejected by nostr::Event deserialiser" + matches!(result, CanvasFetchResult::Failed), + "missing created_at must be Failed — nostr::Event deserialiser rejects it" ); } - /// An event with a timestamp at Timestamp::max() (u64::MAX) must return None. + /// An event with a timestamp at Timestamp::max() (u64::MAX) must return Failed. /// /// `u64::MAX as i64` wraps to -1, which chrono silently accepts as /// 1969-12-31T23:59:59Z. The checked i64::try_from must reject it first. #[test] - fn test_canvas_pointer_from_query_response_timestamp_max_returns_none() { + fn test_canvas_pointer_from_query_response_timestamp_max_returns_failed() { let keys = Keys::generate(); let h_tag = Tag::parse(["h", CHANNEL_UUID]).expect("h tag"); let ev = serde_json::to_value( @@ -6628,15 +6641,15 @@ mod tests { .expect("serialise"); let result = canvas_pointer_from_query_response(&[ev], CHANNEL_UUID); assert!( - result.is_none(), - "Timestamp::max() (u64::MAX) must return None — not wrap to 1969" + matches!(result, CanvasFetchResult::Failed), + "Timestamp::max() (u64::MAX) must be Failed — not wrap to 1969" ); } /// A structurally complete but tampered event (content altered after signing) - /// must be rejected by event.verify(). + /// must be Failed (signature verification error, not confirmed absence). #[test] - fn test_canvas_pointer_from_query_response_tampered_event_returns_none() { + fn test_canvas_pointer_from_query_response_tampered_event_returns_failed() { let keys = Keys::generate(); let h_tag = Tag::parse(["h", CHANNEL_UUID]).expect("h tag"); let mut ev = serde_json::to_value( @@ -6653,14 +6666,14 @@ mod tests { ev["content"] = serde_json::Value::String("injected instructions".into()); let result = canvas_pointer_from_query_response(&[ev], CHANNEL_UUID); assert!( - result.is_none(), - "tampered event must fail verify() and return None" + matches!(result, CanvasFetchResult::Failed), + "tampered event must fail verify() and be Failed — not Absent" ); } - /// An event with the wrong kind (not 40100) must be rejected. + /// An event with the wrong kind (not 40100) must be Failed. #[test] - fn test_canvas_pointer_from_query_response_wrong_kind_returns_none() { + fn test_canvas_pointer_from_query_response_wrong_kind_returns_failed() { let keys = Keys::generate(); let h_tag = Tag::parse(["h", CHANNEL_UUID]).expect("h tag"); let ev = serde_json::to_value( @@ -6671,13 +6684,16 @@ mod tests { ) .expect("serialise"); let result = canvas_pointer_from_query_response(&[ev], CHANNEL_UUID); - assert!(result.is_none(), "wrong kind must return None"); + assert!( + matches!(result, CanvasFetchResult::Failed), + "wrong kind must be Failed — not Absent" + ); } /// An event missing the expected h-tag (or carrying a different channel UUID) - /// must be rejected. + /// must be Failed. #[test] - fn test_canvas_pointer_from_query_response_wrong_h_tag_returns_none() { + fn test_canvas_pointer_from_query_response_wrong_h_tag_returns_failed() { let keys = Keys::generate(); let wrong_h = Tag::parse(["h", "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"]).expect("h tag"); let ev = serde_json::to_value( @@ -6688,14 +6704,20 @@ mod tests { ) .expect("serialise"); let result = canvas_pointer_from_query_response(&[ev], CHANNEL_UUID); - assert!(result.is_none(), "mismatched h-tag must return None"); + assert!( + matches!(result, CanvasFetchResult::Failed), + "mismatched h-tag must be Failed — not Absent" + ); } #[test] fn test_canvas_pointer_from_query_response_timestamp_uses_z_suffix() { let ev = make_canvas_event_value("instructions"); let result = canvas_pointer_from_query_response(&[ev], CHANNEL_UUID); - let pointer = result.expect("valid event must produce a pointer"); + let pointer = match result { + CanvasFetchResult::Present(p) => p, + other => panic!("expected Present, got {other:?}"), + }; assert!( pointer.timestamp.ends_with('Z'), "RFC3339 timestamp must use Z suffix, not +00:00" @@ -6718,8 +6740,9 @@ mod tests { }; let result = cache.resolve_for_turn(&ch, CanvasFetchResult::Present(pointer.clone())); assert_eq!(result, Some(pointer.clone())); - // Cache is updated: get() returns the pointer. - assert_eq!(cache.get(&ch), Some(pointer)); + // Cache is updated: a Failed fetch returns the cached pointer as stale. + let stale = cache.resolve_for_turn(&ch, CanvasFetchResult::Failed); + assert_eq!(stale, Some(pointer)); } #[test] @@ -6735,10 +6758,9 @@ mod tests { // Now canvas is cleared. let result = cache.resolve_for_turn(&ch, CanvasFetchResult::Absent); assert!(result.is_none(), "Absent must return None"); - assert!( - cache.get(&ch).is_none(), - "Absent must clear the cache entry" - ); + // Cache is cleared: a Failed fetch after Absent returns None (nothing cached). + let stale = cache.resolve_for_turn(&ch, CanvasFetchResult::Failed); + assert!(stale.is_none(), "Absent must clear the cache entry"); } #[test] @@ -6800,6 +6822,115 @@ mod tests { assert_eq!(result, Some(new_p), "new revision must replace old pointer"); } + // ── CanvasRevisionCache stale-preservation: Failed from parser preserves cache (A1) ─ + + /// Malformed event (parse error) → Failed → cache unchanged; stale pointer served. + #[test] + fn test_cache_stale_preserved_on_malformed_event() { + let cache = CanvasRevisionCache::new(); + let ch = Uuid::from_u128(0x1234); + let prior = crate::queue::CanvasPointer { + event_id: "stale-rev".to_string(), + timestamp: "2024-01-15T10:30:00Z".to_string(), + }; + cache.resolve_for_turn(&ch, CanvasFetchResult::Present(prior.clone())); + + // A partial event (missing pubkey/sig) → Failed from parser. + let partial = serde_json::json!({ + "id": "a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2", + "created_at": 1705312200_i64, + "content": "some instructions" + }); + let ch_str = ch.to_string(); + let parse_result = canvas_pointer_from_query_response(&[partial], &ch_str); + assert!( + matches!(parse_result, CanvasFetchResult::Failed), + "malformed event must be Failed" + ); + let served = cache.resolve_for_turn(&ch, parse_result); + assert_eq!( + served, + Some(prior), + "Failed result must serve stale pointer, not clear it" + ); + } + + /// Tampered event (sig failure) → Failed → cache unchanged; stale pointer served. + #[test] + fn test_cache_stale_preserved_on_tampered_event() { + use nostr::{EventBuilder, Keys, Kind, Tag}; + let cache = CanvasRevisionCache::new(); + let ch = Uuid::from_u128(0x5678); + let prior = crate::queue::CanvasPointer { + event_id: "stale-rev2".to_string(), + timestamp: "2024-01-15T10:30:00Z".to_string(), + }; + cache.resolve_for_turn(&ch, CanvasFetchResult::Present(prior.clone())); + + let keys = Keys::generate(); + let h_tag = Tag::parse(["h", &ch.to_string()]).expect("h tag"); + let mut ev = serde_json::to_value( + EventBuilder::new( + Kind::Custom(buzz_core::kind::KIND_CANVAS as u16), + "original", + ) + .tags([h_tag]) + .sign_with_keys(&keys) + .expect("sign"), + ) + .expect("serialise"); + ev["content"] = serde_json::Value::String("injected".into()); + + let ch_str = ch.to_string(); + let parse_result = canvas_pointer_from_query_response(&[ev], &ch_str); + assert!( + matches!(parse_result, CanvasFetchResult::Failed), + "tampered event must be Failed" + ); + let served = cache.resolve_for_turn(&ch, parse_result); + assert_eq!( + served, + Some(prior), + "Failed (tampered) must serve stale pointer, not clear it" + ); + } + + /// Wrong h-tag → Failed → cache unchanged; stale pointer served. + #[test] + fn test_cache_stale_preserved_on_wrong_channel_tag() { + use nostr::{EventBuilder, Keys, Kind, Tag}; + let cache = CanvasRevisionCache::new(); + let ch = Uuid::from_u128(0x9abc); + let prior = crate::queue::CanvasPointer { + event_id: "stale-rev3".to_string(), + timestamp: "2024-01-15T10:30:00Z".to_string(), + }; + cache.resolve_for_turn(&ch, CanvasFetchResult::Present(prior.clone())); + + let keys = Keys::generate(); + let wrong_h = Tag::parse(["h", "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"]).expect("h tag"); + let ev = serde_json::to_value( + EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_CANVAS as u16), "content") + .tags([wrong_h]) + .sign_with_keys(&keys) + .expect("sign"), + ) + .expect("serialise"); + + let ch_str = ch.to_string(); + let parse_result = canvas_pointer_from_query_response(&[ev], &ch_str); + assert!( + matches!(parse_result, CanvasFetchResult::Failed), + "wrong h-tag must be Failed" + ); + let served = cache.resolve_for_turn(&ch, parse_result); + assert_eq!( + served, + Some(prior), + "Failed (wrong h-tag) must serve stale pointer, not clear it" + ); + } + // ── ChannelInfoResolver TTL (A3) ────────────────────────────────────────── // ── new-session channel context (one resolve, two consumers) ───────────── @@ -6996,7 +7127,6 @@ mod tests { use std::sync::atomic::Ordering; let id = Uuid::new_v4(); - let response = channel_metadata_response(id, &[["name", "buzz-dev"], ["t", "stream"]]); // Use a new resolver seeded with the channel at an already-expired TTL. let (resolver, requests, server) = counting_resolver(json!([])).await; @@ -7044,6 +7174,228 @@ mod tests { "failed refresh should have attempted (fetch_with_retry = 2 attempts)" ); server.abort(); - let _ = server; + } + + // ── Per-key in-flight refresh guard (A3) ───────────────────────────────── + + /// Two concurrent callers on an expired cache entry must spawn exactly one + /// background refresh — the `in_flight` guard prevents a stampede. + /// + /// Uses a slow-responding server (200ms delay) so both callers race to + /// claim the in-flight slot before the first refresh completes. + #[tokio::test] + async fn test_channel_resolver_concurrent_expiry_spawns_one_refresh() { + use std::sync::atomic::{AtomicUsize, Ordering}; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + let id = Uuid::new_v4(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind"); + let base_url = format!("http://{}", listener.local_addr().unwrap()); + let requests = std::sync::Arc::new(AtomicUsize::new(0)); + let server_requests = requests.clone(); + + let channel_id_str = id.to_string(); + // A well-formed channel-metadata response that includes the channel id. + let body = serde_json::json!([{"tags":[["d", channel_id_str],["name","team-chat"],["t","stream"]]}]).to_string(); + + let server = tokio::spawn(async move { + while let Ok((mut socket, _)) = listener.accept().await { + let mut buf = vec![0u8; 8192]; + let _ = socket.read(&mut buf).await; + server_requests.fetch_add(1, Ordering::SeqCst); + // Slow response (100 ms) so a second caller races the first. + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + let response = format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", + body.len(), + body + ); + let _ = socket.write_all(response.as_bytes()).await; + } + }); + + let rest = crate::relay::RestClient { + http: reqwest::Client::new(), + base_url, + keys: nostr::Keys::generate(), + auth_tag_json: None, + }; + let resolver = ChannelInfoResolver::new(std::collections::HashMap::new(), rest); + + // Warm the cache with an already-expired entry. + { + let mut cache = resolver.cache.write().unwrap(); + cache.insert( + id, + CachedChannelInfo { + info: crate::queue::PromptChannelInfo { + name: "team-chat".to_string(), + channel_type: "stream".to_string(), + description: None, + }, + refresh_after: tokio::time::Instant::now() - std::time::Duration::from_secs(1), + next_refresh_at: tokio::time::Instant::now() + - std::time::Duration::from_secs(1), + }, + ); + } + + let baseline = requests.load(Ordering::SeqCst); + + // Two concurrent resolve() calls on the same expired entry. + let r1 = resolver.resolve(id); + let r2 = resolver.resolve(id); + let (s1, s2) = tokio::join!(r1, r2); + assert_eq!(s1.as_ref().map(|i| i.name.as_str()), Some("team-chat")); + assert_eq!(s2.as_ref().map(|i| i.name.as_str()), Some("team-chat")); + + // Wait long enough for the background refresh to complete (100 ms server + // delay + fetch overhead). + tokio::time::sleep(std::time::Duration::from_millis(400)).await; + + let fired = requests.load(Ordering::SeqCst) - baseline; + assert_eq!( + fired, 1, + "exactly one background refresh must be spawned for concurrent expiry callers" + ); + + server.abort(); + } + + // ── query_once is single-attempt (A4) ───────────────────────────────────── + + /// `RestClient::query_once` issues exactly one HTTP request even when the + /// server returns a retriable status (503), verifying it does not inherit + /// the retry loop of `request_with_retry`. + #[tokio::test] + async fn test_query_once_makes_exactly_one_request_on_retriable_response() { + use std::sync::atomic::{AtomicUsize, Ordering}; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind"); + let base_url = format!("http://{}", listener.local_addr().unwrap()); + let requests = std::sync::Arc::new(AtomicUsize::new(0)); + let server_requests = requests.clone(); + + let server = tokio::spawn(async move { + while let Ok((mut socket, _)) = listener.accept().await { + let mut buf = vec![0u8; 8192]; + let _ = socket.read(&mut buf).await; + server_requests.fetch_add(1, Ordering::SeqCst); + // Always return 503 Service Unavailable — a retriable status. + let response = + "HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"; + let _ = socket.write_all(response.as_bytes()).await; + } + }); + + let rest = crate::relay::RestClient { + http: reqwest::Client::new(), + base_url, + keys: nostr::Keys::generate(), + auth_tag_json: None, + }; + + let filter = nostr::Filter::new().kind(nostr::Kind::Custom(39000)); + let result = rest.query_once(&[filter]).await; + + assert!( + result.is_err(), + "query_once must return Err on a non-2xx response" + ); + assert_eq!( + requests.load(Ordering::SeqCst), + 1, + "query_once must issue exactly one HTTP request — no retry loop" + ); + + server.abort(); + } + + // ── initial_message context composition (A2) ────────────────────────────── + + /// `initial_message` for a channel turn: the `[Context]` block prepended + /// must show `Scope: channel`, include the description, and include the + /// canvas revision — same renderer as the batch prompt. + #[test] + fn test_initial_message_context_channel_has_scope_description_canvas() { + let channel_id = Uuid::from_u128(0x0001); + let channel_info = crate::queue::PromptChannelInfo { + name: "team-chat".to_string(), + channel_type: "stream".to_string(), + description: Some("Engineering discussions".to_string()), + }; + let canvas = crate::queue::CanvasPointer { + event_id: "abc123".to_string(), + timestamp: "2024-06-01T12:00:00Z".to_string(), + }; + + // This is the exact call made by initial_message in run_prompt_task (A2). + let ctx_block = crate::queue::format_context_hints( + channel_id, + Some(&channel_info), + &crate::queue::ThreadTags::default(), + false, // is_dm_turn = false + false, + None, + Some(&canvas), + ); + + assert!( + ctx_block.contains("Scope: channel"), + "channel turn must have Scope: channel; got:\n{ctx_block}" + ); + assert!( + ctx_block.contains("Description: Engineering discussions"), + "channel turn must include description; got:\n{ctx_block}" + ); + assert!( + ctx_block.contains("Canvas revision (event ID): abc123"), + "channel turn must include canvas revision; got:\n{ctx_block}" + ); + } + + /// `initial_message` for a DM turn: `[Context]` must show `Scope: dm`, + /// no description, and no canvas revision — DM fail-closed. + #[test] + fn test_initial_message_context_dm_has_scope_dm_no_description_no_canvas() { + let channel_id = Uuid::from_u128(0x0002); + let dm_info = crate::queue::PromptChannelInfo { + name: "Direct Message".to_string(), + channel_type: "dm".to_string(), + description: Some("should be suppressed".to_string()), + }; + let canvas = crate::queue::CanvasPointer { + event_id: "def456".to_string(), + timestamp: "2024-06-01T12:00:00Z".to_string(), + }; + + // is_dm_turn = true; canvas should be suppressed regardless of pointer presence. + let ctx_block = crate::queue::format_context_hints( + channel_id, + Some(&dm_info), + &crate::queue::ThreadTags::default(), + true, // is_dm_turn = true + false, + None, + Some(&canvas), + ); + + assert!( + ctx_block.contains("Scope: dm"), + "DM turn must have Scope: dm; got:\n{ctx_block}" + ); + assert!( + !ctx_block.contains("Description:"), + "DM turn must not include description; got:\n{ctx_block}" + ); + assert!( + !ctx_block.contains("Canvas revision (event ID):"), + "DM turn must not include canvas revision; got:\n{ctx_block}" + ); } } diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index 085b256bfa..47040f3bf3 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -1249,7 +1249,7 @@ fn resolve_reply_anchor( /// replies; in the channel branch a `Some` anchor means a human-facing /// top-level mention whose reply should open a new thread rooted at the /// triggering event. -fn format_context_hints( +pub(crate) fn format_context_hints( channel_id: Uuid, channel_info: Option<&PromptChannelInfo>, thread_tags: &ThreadTags, @@ -1402,23 +1402,6 @@ pub(crate) fn append_canvas_pointer( )); } -/// Alias used by the `initial_message` dispatch in pool.rs. -pub(crate) fn append_description_for_initial_message( - s: &mut String, - channel_info: Option<&PromptChannelInfo>, -) { - append_channel_description(s, channel_info); -} - -/// Alias used by the `initial_message` dispatch in pool.rs. -pub(crate) fn append_canvas_pointer_for_initial_message( - s: &mut String, - pointer: Option<&CanvasPointer>, - channel_uuid: &str, -) { - append_canvas_pointer(s, pointer, channel_uuid); -} - /// Format a conversation context section (thread or DM). fn format_conversation_context( ctx: &ConversationContext, diff --git a/crates/buzz-acp/src/relay.rs b/crates/buzz-acp/src/relay.rs index 6a36073bcd..0fc1f5da25 100644 --- a/crates/buzz-acp/src/relay.rs +++ b/crates/buzz-acp/src/relay.rs @@ -420,6 +420,43 @@ impl RestClient { .map_err(|e| RelayError::Http(e.to_string())) } + /// Single-attempt query — identical to `query` but bypasses `request_with_retry`. + /// + /// Use on per-turn hot paths (e.g. canvas fetch) where the latency contract + /// forbids implicit retries: one HTTP attempt, one outcome, no retry delays. + pub async fn query_once(&self, filters: &[nostr::Filter]) -> Result { + let body_bytes = serde_json::to_vec(filters) + .map_err(|e| RelayError::Http(format!("filter serialize error: {e}")))?; + let url = format!("{}/query", self.base_url); + let body_owned = body_bytes.to_vec(); + let auth_tag_header = self.auth_tag_json.clone(); + let auth = self + .nip98_header("POST", &url, Some(&body_owned)) + .unwrap_or_default(); + let mut req = self + .http + .post(&url) + .header("Authorization", auth) + .header("Content-Type", "application/json"); + if let Some(ref tag) = auth_tag_header { + req = req.header("x-auth-tag", tag); + } + let resp = req + .body(body_owned) + .send() + .await + .map_err(|e| RelayError::Http(e.to_string()))?; + if !resp.status().is_success() { + return Err(RelayError::Http(format!( + "POST /query returned HTTP {}", + resp.status() + ))); + } + resp.json() + .await + .map_err(|e| RelayError::Http(e.to_string())) + } + /// Count events via the HTTP bridge: `POST /count` with NIP-98 auth. /// /// Accepts a slice of `nostr::Filter` (serialized as JSON array). From 9e6ff0e512cddad8089de79bffa0805137545cc0 Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Mon, 3 Aug 2026 13:31:26 -0400 Subject: [PATCH 3/6] fix(acp): thread authoritative DM classification through all turn consumers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pass is_dm_turn (fail-closed: unresolved metadata → true) into FormatPromptArgs::is_dm and fetch_conversation_context rather than letting each consumer re-derive it from channel_info with opposite defaults. Before: format_prompt derived is_dm via channel_info.unwrap_or(false), and fetch_conversation_context did the same — so an unresolved turn emitted Scope: dm in initial_message but Scope: channel in the batch prompt. After: a single is_dm_turn value computed at turn start flows through FormatPromptArgs and directly into fetch_conversation_context; both prompt paths share the same classification. The channel_info parameter is removed from fetch_conversation_context (it was only used to derive is_dm). Tests: added test_unresolved_metadata_both_prompts_render_scope_dm proving that initial_message and batch both produce Scope: dm, no description, and no canvas when metadata is unresolved. Updated 8 existing DM-scope tests to pass is_dm: true explicitly since the field now governs scope rendering. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- crates/buzz-acp/src/pool.rs | 106 ++++++++++++++++++++++++++++++++--- crates/buzz-acp/src/queue.rs | 20 +++++-- 2 files changed, 115 insertions(+), 11 deletions(-) diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 574f143b1d..7fbfd0568c 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -1939,7 +1939,7 @@ pub async fn run_prompt_task( let channel_info = turn_channel_info.clone(); let conversation_context = if ctx.context_message_limit > 0 { - fetch_conversation_context(b, &channel_info, &ctx).await + fetch_conversation_context(b, &ctx, is_dm_turn).await } else { None }; @@ -1975,6 +1975,7 @@ pub async fn run_prompt_task( system_prompt: ctx.system_prompt.as_deref(), team_instructions: ctx.team_instructions.as_deref(), canvas_pointer: canvas_pointer.as_ref(), + is_dm: is_dm_turn, }, ) } else { @@ -2712,14 +2713,10 @@ pub(crate) fn canvas_pointer_from_query_response( /// reply event only (most recent = most likely to need a response). async fn fetch_conversation_context( batch: &FlushBatch, - channel_info: &Option, ctx: &PromptContext, + is_dm_turn: bool, ) -> Option { let limit = ctx.context_message_limit; - let is_dm = channel_info - .as_ref() - .map(|ci| ci.channel_type == "dm") - .unwrap_or(false); // Check thread tags on the last event first — this applies to both // channels and DMs. A DM reply needs thread context (not channel history) @@ -2738,7 +2735,7 @@ async fn fetch_conversation_context( } // DM non-reply: fetch recent conversation history. - if is_dm { + if is_dm_turn { return fetch_dm_context(batch.channel_id, limit, &ctx.rest_client).await; } @@ -7398,4 +7395,99 @@ mod tests { "DM turn must not include canvas revision; got:\n{ctx_block}" ); } + + // ── Unresolved-metadata fail-closed classification (A1/A2) ─────────────── + + /// When channel metadata cannot be resolved (None), the turn's authoritative + /// `is_dm_turn = true` must flow through to BOTH prompt paths: + /// + /// 1. `initial_message` (via `format_context_hints` with `is_dm = true`) + /// 2. Batch prompt (via `FormatPromptArgs::is_dm = true`) + /// + /// Both must render `Scope: dm`, omit description, omit canvas pointer, and + /// select DM conversation behavior — never silently fall back to channel scope. + #[test] + fn test_unresolved_metadata_both_prompts_render_scope_dm() { + use crate::queue::{ + format_context_hints, format_prompt, CanvasPointer, FlushBatch, FormatPromptArgs, + ThreadTags, + }; + use nostr::Timestamp; + + let channel_id = Uuid::from_u128(0x0003); + // Simulate a canvas pointer that must be suppressed for DM turns. + let canvas = CanvasPointer { + event_id: "cafebabe".to_string(), + timestamp: "2024-06-01T12:00:00Z".to_string(), + }; + + // ── initial_message path ───────────────────────────────────────────── + // run_prompt_task passes: channel_info = None (unresolved), is_dm_turn = true. + let init_ctx = format_context_hints( + channel_id, + None, // channel_info = None (unresolved metadata) + &ThreadTags::default(), + true, // is_dm_turn = true (fail-closed) + false, + None, + Some(&canvas), // canvas pointer present — must be suppressed + ); + + assert!( + init_ctx.contains("Scope: dm"), + "initial_message: unresolved metadata must render Scope: dm; got:\n{init_ctx}" + ); + assert!( + !init_ctx.contains("Description:"), + "initial_message: unresolved metadata must not include description; got:\n{init_ctx}" + ); + assert!( + !init_ctx.contains("Canvas revision (event ID):"), + "initial_message: unresolved metadata must not include canvas; got:\n{init_ctx}" + ); + + // ── batch prompt path ──────────────────────────────────────────────── + // FormatPromptArgs::is_dm = true must govern rendering, not channel_info. + let event = { + let keys = nostr::Keys::generate(); + nostr::EventBuilder::new(nostr::Kind::Custom(1), "hello") + .custom_created_at(Timestamp::from(1_700_000_000u64)) + .sign_with_keys(&keys) + .expect("sign") + }; + let batch = FlushBatch { + channel_id, + events: vec![crate::queue::BatchEvent { + event, + prompt_tag: "test".into(), + received_at: std::time::Instant::now(), + }], + cancelled_events: vec![], + cancel_reason: None, + }; + + let prompt = format_prompt( + &batch, + &FormatPromptArgs { + channel_info: None, // unresolved metadata + is_dm: true, // fail-closed classification + canvas_pointer: Some(&canvas), // must be suppressed + ..Default::default() + }, + ) + .join("\n\n"); + + assert!( + prompt.contains("Scope: dm"), + "batch prompt: unresolved metadata must render Scope: dm; got:\n{prompt}" + ); + assert!( + !prompt.contains("Description:"), + "batch prompt: unresolved metadata must not include description; got:\n{prompt}" + ); + assert!( + !prompt.contains("Canvas revision (event ID):"), + "batch prompt: unresolved metadata must not include canvas; got:\n{prompt}" + ); + } } diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index 47040f3bf3..6147551028 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -1460,6 +1460,13 @@ pub struct FormatPromptArgs<'a> { /// tri-state resolved: `Some` = present/stale-served, `None` = confirmed /// absent or first-fetch failure. DM turns always pass `None`. pub canvas_pointer: Option<&'a CanvasPointer>, + /// Authoritative DM classification for this turn, computed once at turn + /// start with fail-closed semantics (unresolved metadata → `true`). + /// + /// Passed through rather than re-derived from `channel_info` so that all + /// prompt paths within one turn share the same classification regardless of + /// whether metadata resolved successfully. + pub is_dm: bool, } /// Format the `[Base]` section for the base prompt. @@ -1504,10 +1511,7 @@ pub fn format_prompt(batch: &FlushBatch, args: &FormatPromptArgs<'_>) -> Vec = Vec::with_capacity(7); @@ -3100,6 +3104,7 @@ mod tests { &batch, &FormatPromptArgs { channel_info: Some(&ci), + is_dm: true, ..Default::default() }, ) @@ -3221,6 +3226,7 @@ mod tests { &FormatPromptArgs { channel_info: Some(&ci), conversation_context: Some(&ctx), + is_dm: true, ..Default::default() }, ) @@ -3479,6 +3485,7 @@ mod tests { &FormatPromptArgs { channel_info: Some(&ci), conversation_context: Some(&ctx), + is_dm: true, ..Default::default() }, ) @@ -3527,6 +3534,7 @@ mod tests { &batch, &FormatPromptArgs { channel_info: Some(&ci), + is_dm: true, ..Default::default() }, ) @@ -4022,6 +4030,7 @@ mod tests { &batch, &FormatPromptArgs { channel_info: Some(&ci), + is_dm: true, ..Default::default() }, ) @@ -4086,6 +4095,7 @@ mod tests { &batch, &FormatPromptArgs { channel_info: Some(&ci), + is_dm: true, ..Default::default() }, ) @@ -4637,6 +4647,7 @@ mod tests { canvas_pointer: Some(&pointer), channel_info: Some(&ci), has_system_prompt_support: true, + is_dm: true, ..Default::default() }, ) @@ -5013,6 +5024,7 @@ mod tests { &FormatPromptArgs { channel_info: Some(&ci), has_system_prompt_support: true, + is_dm: true, ..Default::default() }, ) From e46e6122049a89c4bf42b8d5791704aa2ca548ae Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Mon, 3 Aug 2026 14:22:02 -0400 Subject: [PATCH 4/6] test(acp): prove fetch_conversation_context DM path is governed by is_dm_turn Add test_fetch_conversation_context_dm_classification_is_authoritative: a local HTTP server verifies that is_dm_turn=true reaches the DM history endpoint and returns ConversationContext::Dm, while is_dm_turn=false issues zero HTTP requests and returns None. This is the missing regression tripwire from pass-3: the existing unresolved-metadata test proved rendering on both prompt paths but never exercised fetch_conversation_context, so a regression reverting its is_dm_turn branch to unwrap_or(false) would have been invisible. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- crates/buzz-acp/src/pool.rs | 111 ++++++++++++++++++++++++++++++++++++ 1 file changed, 111 insertions(+) diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 7fbfd0568c..4dc0fe4a2a 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -7490,4 +7490,115 @@ mod tests { "batch prompt: unresolved metadata must not include canvas; got:\n{prompt}" ); } + + // ── Unresolved-metadata: fetch_conversation_context DM selection (A1/A2) ── + + /// `fetch_conversation_context` with `is_dm_turn = true` must take the DM + /// history path and return `ConversationContext::Dm`; with `is_dm_turn = false` + /// it must skip the DM fetch entirely (no HTTP request). + /// + /// This is the missing tripwire for the pass-2 fix: proves the authoritative + /// `is_dm_turn` boolean governs `fetch_conversation_context` so that a + /// regression silently reverting to `unwrap_or(false)` breaks this test. + #[tokio::test] + async fn test_fetch_conversation_context_dm_classification_is_authoritative() { + use std::sync::atomic::{AtomicUsize, Ordering}; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + // Build a minimal DM events response that parse_nostr_dm_response accepts: + // one event with content + created_at. pubkey is optional in the parser. + let dm_event = serde_json::json!([{ + "content": "hello from dm", + "created_at": 1_700_000_000u64, + "pubkey": "aabbcc" + }]); + let body = dm_event.to_string(); + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind"); + let base_url = format!("http://{}", listener.local_addr().unwrap()); + let requests = std::sync::Arc::new(AtomicUsize::new(0)); + let server_requests = requests.clone(); + + let server = tokio::spawn(async move { + while let Ok((mut socket, _)) = listener.accept().await { + let mut buf = vec![0u8; 8192]; + let _ = socket.read(&mut buf).await; + server_requests.fetch_add(1, Ordering::SeqCst); + let response = format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", + body.len(), + body + ); + let _ = socket.write_all(response.as_bytes()).await; + } + }); + + let agent_keys = nostr::Keys::generate(); + let rest = crate::relay::RestClient { + http: reqwest::Client::new(), + base_url, + keys: agent_keys.clone(), + auth_tag_json: None, + }; + + // A batch with no thread tags — ensures parse_thread_tags returns no + // root_event_id, so fetch_conversation_context reaches the is_dm_turn branch. + let make_batch = |channel_id: Uuid| { + let keys = nostr::Keys::generate(); + let event = nostr::EventBuilder::new(nostr::Kind::Custom(1), "msg") + .custom_created_at(nostr::Timestamp::from(1_700_000_000u64)) + .sign_with_keys(&keys) + .expect("sign"); + FlushBatch { + channel_id, + events: vec![crate::queue::BatchEvent { + event, + prompt_tag: "test".into(), + received_at: std::time::Instant::now(), + }], + cancelled_events: vec![], + cancel_reason: None, + } + }; + + let ctx = PromptContext { + rest_client: rest, + context_message_limit: 10, + agent_keys, + ..make_prompt_context_no_owner() + }; + + // ── Case 1: is_dm_turn = true → must take the DM history path ──────── + let channel_id = Uuid::from_u128(0x1001); + let result = fetch_conversation_context(&make_batch(channel_id), &ctx, true).await; + + assert!( + matches!(result, Some(ConversationContext::Dm { .. })), + "is_dm_turn=true must return ConversationContext::Dm; got: {result:?}" + ); + let dm_requests = requests.load(Ordering::SeqCst); + assert!( + dm_requests >= 1, + "is_dm_turn=true must issue at least one HTTP request to fetch DM history" + ); + + // ── Case 2: is_dm_turn = false → must skip DM fetch entirely ───────── + let requests_before = requests.load(Ordering::SeqCst); + let channel_id2 = Uuid::from_u128(0x1002); + let result = fetch_conversation_context(&make_batch(channel_id2), &ctx, false).await; + + assert!( + result.is_none(), + "is_dm_turn=false on a non-thread turn must return None (no DM fetch); got: {result:?}" + ); + assert_eq!( + requests.load(Ordering::SeqCst), + requests_before, + "is_dm_turn=false must not issue any HTTP request to the DM history endpoint" + ); + + server.abort(); + } } From fbc9625c110ecae134d11a3c4b6f0e31a78058ee Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Mon, 3 Aug 2026 15:17:48 -0400 Subject: [PATCH 5/6] test(acp): prove is_dm_turn forwarding via run_prompt_task end-to-end Add test_run_prompt_task_dm_classification_forwarding_is_end_to_end_authoritative which enters through run_prompt_task with a DM-seeded channel and a scripted bash agent. The test asserts that a kind-40002 query reaches the HTTP server, proving the fail-closed classification at pool.rs:1616-1632 and its forwarding at pool.rs:1942 are one tested behavior. Mutation check confirmed: changing pool.rs:1942 to pass a hardcoded false suppresses the DM fetch entirely (0 kind-40002 requests), causing the assertion to fail. The existing A1 helper-level test is unchanged. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- crates/buzz-acp/src/pool.rs | 187 ++++++++++++++++++++++++++++++++++++ 1 file changed, 187 insertions(+) diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 4dc0fe4a2a..41f79529cf 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -7601,4 +7601,191 @@ mod tests { server.abort(); } + + // ── A2: run_prompt_task DM forwarding end-to-end ──────────────────────── + + /// Proves the fail-closed `is_dm_turn` classification at `pool.rs:1616-1632` + /// and its forwarding at `pool.rs:1942` as ONE tested behavior by entering + /// through `run_prompt_task` with a seeded-DM channel and a scripted bash + /// agent. The DM history HTTP endpoint must be reached from within the + /// production call path, proving the forwarded value governs + /// `fetch_conversation_context`. + /// + /// ## Mutation check + /// + /// Temporarily changing `pool.rs:1942` to pass a hardcoded `false` instead + /// of `is_dm_turn` causes `fetch_conversation_context` to skip the DM fetch + /// entirely (no thread root, not DM → `None`, zero HTTP requests). The + /// `requests >= 1` assertion then fails, catching the dropped-forwarding + /// regression Thufir identified as the exact defect this test must prove. + #[tokio::test] + async fn test_run_prompt_task_dm_classification_forwarding_is_end_to_end_authoritative() { + use std::sync::atomic::{AtomicUsize, Ordering}; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + // ── HTTP server: counts all requests, returns a valid DM event body ── + // Both the DM history fetch and the profile lookup hit `/query`; any + // request proves the DM path was reached. The DM parse only requires a + // non-empty array with `content` — `pubkey` defaults to "unknown". + let dm_body = serde_json::json!([{ + "content": "hello from dm", + "created_at": 1_700_000_000u64, + "pubkey": "aabb" + }]) + .to_string(); + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind"); + let base_url = format!("http://{}", listener.local_addr().unwrap()); + // `dm_requests` counts only requests whose body contains the DM-kind + // sentinel (40002 = KIND_STREAM_MESSAGE_V2). This distinguishes the + // DM context fetch from the profile-lookup request, which uses kind 0 + // (Metadata) and is always issued regardless of `is_dm_turn`. + let dm_requests = std::sync::Arc::new(AtomicUsize::new(0)); + let server_dm_requests = dm_requests.clone(); + let server_body = dm_body.clone(); + + let server = tokio::spawn(async move { + while let Ok((mut socket, _)) = listener.accept().await { + let mut buf = vec![0u8; 8192]; + let n = socket.read(&mut buf).await.unwrap_or(0); + let raw = String::from_utf8_lossy(&buf[..n]); + // Count only the DM context query (contains kind 40002). + // Profile-lookup queries contain only kind 0 (Metadata) and + // never 40002, so this counter is specific to the DM path. + if raw.contains("40002") { + server_dm_requests.fetch_add(1, Ordering::SeqCst); + } + let response = format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", + server_body.len(), + server_body + ); + let _ = socket.write_all(response.as_bytes()).await; + } + }); + + // ── Scripted bash agent: responds to session/new (id=0) then ───────── + // session/prompt (id=1). The DM HTTP fetch happens between the two + // ACP calls, so the script needs to hold open long enough for the + // context fetch to complete before it reads and responds to the prompt. + let script = r#" + read -t 10 _req1 + printf '{"jsonrpc":"2.0","id":0,"result":{"sessionId":"ses-dm-test"}}\n' + read -t 10 _req2 + printf '{"jsonrpc":"2.0","id":1,"result":{"stopReason":"end_turn"}}\n' + sleep 1 + "#; + + let acp = crate::acp::AcpClient::spawn( + "bash", + &["-c".to_string(), script.to_string()], + &[], + false, + ) + .await + .expect("spawn bash agent"); + + // ── OwnedAgent: protocol_version=2, agent_name not "goose" ──────────── + let agent_keys = nostr::Keys::generate(); + let agent = OwnedAgent { + index: 0, + acp, + state: SessionState::default(), + model_capabilities: None, + desired_model: None, + model_overridden: false, + agent_name: "test-bash".into(), + goose_system_prompt_supported: None, + protocol_version: 2, + }; + + // ── PromptContext: channel seeded as DM type so no network fetch ────── + // needed for channel metadata; rest_client points to the HTTP server + // so the DM history fetch (and profile lookup) are counted. + let channel_id = Uuid::from_u128(0x2001); + let mut channel_startup: std::collections::HashMap = + std::collections::HashMap::new(); + channel_startup.insert( + channel_id, + ChannelInfo { + name: "test-dm".to_string(), + channel_type: "dm".to_string(), + description: None, + }, + ); + let rest_client = crate::relay::RestClient { + http: reqwest::Client::new(), + base_url: base_url.clone(), + keys: agent_keys.clone(), + auth_tag_json: None, + }; + let channel_info_rest = crate::relay::RestClient { + http: reqwest::Client::new(), + base_url: base_url.clone(), + keys: agent_keys.clone(), + auth_tag_json: None, + }; + let ctx = std::sync::Arc::new(PromptContext { + rest_client, + context_message_limit: 10, + agent_keys: agent_keys.clone(), + channel_info: ChannelInfoResolver::new(channel_startup, channel_info_rest), + ..make_prompt_context_no_owner() + }); + + // ── FlushBatch: no thread tags → DM non-reply path in ──────────────── + // fetch_conversation_context. The event pubkey is a fresh key so the + // profile lookup will attempt one additional `/query` request, which + // also contributes to the request count but does not affect correctness. + let event = { + let event_keys = nostr::Keys::generate(); + nostr::EventBuilder::new(nostr::Kind::Custom(1), "hello dm") + .custom_created_at(nostr::Timestamp::from(1_700_000_000u64)) + .sign_with_keys(&event_keys) + .expect("sign") + }; + let batch = FlushBatch { + channel_id, + events: vec![crate::queue::BatchEvent { + event, + prompt_tag: "test".into(), + received_at: std::time::Instant::now(), + }], + cancelled_events: vec![], + cancel_reason: None, + }; + + // ── run_prompt_task: fire and receive the PromptResult ──────────────── + let (result_tx, mut result_rx) = tokio::sync::mpsc::unbounded_channel::(); + run_prompt_task( + agent, + Some(batch), + None, + ctx, + result_tx, + None, // control_rx: None = non-cancellable path + "turn-dm-test".to_string(), + ) + .await; + + // Drain the result channel (run_prompt_task always sends one PromptResult). + let _result = result_rx.try_recv().expect("PromptResult must be sent"); + + // ── Assertion: DM fetch (kind-40002 query) must have reached the server ─ + // This is the mutation tripwire: changing pool.rs:1942 to pass `false` + // skips the DM fetch entirely so dm_requests stays 0. Profile-lookup + // requests (kind 0) are issued regardless and are NOT counted here. + let total_dm_requests = dm_requests.load(Ordering::SeqCst); + assert!( + total_dm_requests >= 1, + "run_prompt_task with a DM channel must issue at least one DM history \ + query (containing kind 40002) via fetch_conversation_context; \ + got {total_dm_requests} — if this is 0, the is_dm_turn forwarding \ + at pool.rs:1942 has regressed" + ); + + server.abort(); + } } From 52d2980360950f632b92c39bf4c170522de6060c Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Mon, 3 Aug 2026 15:44:42 -0400 Subject: [PATCH 6/6] test(acp): use unresolved metadata in run_prompt_task DM forwarding test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rework test_run_prompt_task_dm_classification_forwarding_is_end_to_end_authoritative to enter the fail-closed .unwrap_or(true) arm rather than a pre-seeded DM channel. The channel is no longer inserted into ChannelInfoResolver at startup. The kind-39000 resolver slow-path now hits the test server, which returns an empty array; fetch_channel_info returns None after both fetch_with_retry attempts. resolve() returns None → .unwrap_or(true) → is_dm_turn = true (fail-closed) → forwarded to fetch_conversation_context at pool.rs:1942 → kind-40002 query. Dual mutation check verified before push: - pool.rs:1622 .unwrap_or(false): is_dm_turn = false, DM fetch skipped, dm_requests = 0, assertion FAILS. - pool.rs:1942 hardcoded false: same outcome, assertion FAILS. Both mutations were reverted; production code is unchanged. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- crates/buzz-acp/src/pool.rs | 135 +++++++++++++++++++++--------------- 1 file changed, 81 insertions(+), 54 deletions(-) diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 41f79529cf..71ead9aaa1 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -7606,70 +7606,99 @@ mod tests { /// Proves the fail-closed `is_dm_turn` classification at `pool.rs:1616-1632` /// and its forwarding at `pool.rs:1942` as ONE tested behavior by entering - /// through `run_prompt_task` with a seeded-DM channel and a scripted bash - /// agent. The DM history HTTP endpoint must be reached from within the - /// production call path, proving the forwarded value governs - /// `fetch_conversation_context`. + /// through `run_prompt_task` with **unresolved** channel metadata and a + /// scripted bash agent. /// - /// ## Mutation check + /// ## How fail-close is exercised /// - /// Temporarily changing `pool.rs:1942` to pass a hardcoded `false` instead - /// of `is_dm_turn` causes `fetch_conversation_context` to skip the DM fetch - /// entirely (no thread root, not DM → `None`, zero HTTP requests). The - /// `requests >= 1` assertion then fails, catching the dropped-forwarding - /// regression Thufir identified as the exact defect this test must prove. + /// The channel is NOT pre-seeded in `ChannelInfoResolver`. On the slow + /// path, `resolve()` calls `fetch_channel_info`, which issues a kind-39000 + /// metadata query. The test server returns `[]` for kind-39000 (a valid + /// empty array — no metadata event), causing `fetch_channel_info` to return + /// `None` after its two attempts (initial + `fetch_with_retry` once). With + /// `resolve()` returning `None`, the classification at `pool.rs:1622` + /// reaches `.unwrap_or(true)` — fail-closed — setting `is_dm_turn = true`. + /// That value is forwarded to `fetch_conversation_context` at `pool.rs:1942`, + /// which then issues a kind-40002 DM history query. The kind-40002 counter + /// proves the entire chain is one tested behavior. + /// + /// ## Mutation checks + /// + /// 1. `.unwrap_or(true)` → `.unwrap_or(false)` at `pool.rs:1622`: + /// `is_dm_turn` becomes `false`; `fetch_conversation_context` skips the + /// DM fetch (no thread root, not DM → `None`, zero kind-40002 requests). + /// The `dm_requests >= 1` assertion fails. + /// + /// 2. `is_dm_turn` → hardcoded `false` at `pool.rs:1942`: + /// Same outcome — DM fetch skipped, assertion fails. + /// + /// Both mutations were verified before push; see the hand-back message for + /// explicit confirmation that both tripped. #[tokio::test] async fn test_run_prompt_task_dm_classification_forwarding_is_end_to_end_authoritative() { use std::sync::atomic::{AtomicUsize, Ordering}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; - // ── HTTP server: counts all requests, returns a valid DM event body ── - // Both the DM history fetch and the profile lookup hit `/query`; any - // request proves the DM path was reached. The DM parse only requires a - // non-empty array with `content` — `pubkey` defaults to "unknown". + // ── HTTP server: dispatches on request kind ─────────────────────────── + // kind-39000 (NIP-29 group metadata): returns `[]` — no metadata event. + // → fetch_channel_info returns None → resolve() returns None + // → .unwrap_or(true) fires → is_dm_turn = true (fail-closed) + // kind-40002 (KIND_STREAM_MESSAGE_V2, DM history): returns a minimal + // DM event array; the request is counted in dm_requests. + // kind-0 (profile lookup): returns `[]` (not counted). + // + // fetch_with_retry issues the kind-39000 call twice (initial attempt + + // one retry after CONTEXT_FETCH_RETRY_DELAY=500 ms); both return `[]`. let dm_body = serde_json::json!([{ "content": "hello from dm", "created_at": 1_700_000_000u64, "pubkey": "aabb" }]) .to_string(); + let empty_body = "[]".to_string(); let listener = tokio::net::TcpListener::bind("127.0.0.1:0") .await .expect("bind"); let base_url = format!("http://{}", listener.local_addr().unwrap()); - // `dm_requests` counts only requests whose body contains the DM-kind - // sentinel (40002 = KIND_STREAM_MESSAGE_V2). This distinguishes the - // DM context fetch from the profile-lookup request, which uses kind 0 - // (Metadata) and is always issued regardless of `is_dm_turn`. + + // `dm_requests` counts only requests whose body contains kind 40002 + // (KIND_STREAM_MESSAGE_V2). kind-39000 metadata queries and kind-0 + // profile lookups are never counted here. let dm_requests = std::sync::Arc::new(AtomicUsize::new(0)); let server_dm_requests = dm_requests.clone(); - let server_body = dm_body.clone(); + let server_dm_body = dm_body.clone(); + let server_empty_body = empty_body.clone(); let server = tokio::spawn(async move { while let Ok((mut socket, _)) = listener.accept().await { let mut buf = vec![0u8; 8192]; let n = socket.read(&mut buf).await.unwrap_or(0); let raw = String::from_utf8_lossy(&buf[..n]); - // Count only the DM context query (contains kind 40002). - // Profile-lookup queries contain only kind 0 (Metadata) and - // never 40002, so this counter is specific to the DM path. - if raw.contains("40002") { + + let body = if raw.contains("40002") { + // DM history query — count it and return the DM event. server_dm_requests.fetch_add(1, Ordering::SeqCst); - } + server_dm_body.clone() + } else { + // kind-39000 metadata query or kind-0 profile lookup — + // return an empty array so fetch_channel_info returns None. + server_empty_body.clone() + }; + let response = format!( "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", - server_body.len(), - server_body + body.len(), + body ); let _ = socket.write_all(response.as_bytes()).await; } }); // ── Scripted bash agent: responds to session/new (id=0) then ───────── - // session/prompt (id=1). The DM HTTP fetch happens between the two - // ACP calls, so the script needs to hold open long enough for the - // context fetch to complete before it reads and responds to the prompt. + // session/prompt (id=1). Two kind-39000 fetches (fetch_with_retry) + // plus the 500 ms retry sleep occur before the DM history fetch, so the + // script must tolerate a ~1–2 s pause; read -t 10 absorbs that. let script = r#" read -t 10 _req1 printf '{"jsonrpc":"2.0","id":0,"result":{"sessionId":"ses-dm-test"}}\n' @@ -7687,7 +7716,7 @@ mod tests { .await .expect("spawn bash agent"); - // ── OwnedAgent: protocol_version=2, agent_name not "goose" ──────────── + // ── OwnedAgent: protocol_version=2, agent_name not "goose" ─────────── let agent_keys = nostr::Keys::generate(); let agent = OwnedAgent { index: 0, @@ -7701,20 +7730,11 @@ mod tests { protocol_version: 2, }; - // ── PromptContext: channel seeded as DM type so no network fetch ────── - // needed for channel metadata; rest_client points to the HTTP server - // so the DM history fetch (and profile lookup) are counted. + // ── PromptContext: channel NOT pre-seeded — resolver takes the slow ─── + // path and issues kind-39000 queries to the test server. Both + // rest_client and channel_info_rest point to the same server so all + // HTTP traffic is observable. let channel_id = Uuid::from_u128(0x2001); - let mut channel_startup: std::collections::HashMap = - std::collections::HashMap::new(); - channel_startup.insert( - channel_id, - ChannelInfo { - name: "test-dm".to_string(), - channel_type: "dm".to_string(), - description: None, - }, - ); let rest_client = crate::relay::RestClient { http: reqwest::Client::new(), base_url: base_url.clone(), @@ -7731,14 +7751,16 @@ mod tests { rest_client, context_message_limit: 10, agent_keys: agent_keys.clone(), - channel_info: ChannelInfoResolver::new(channel_startup, channel_info_rest), + channel_info: ChannelInfoResolver::new( + std::collections::HashMap::new(), // no startup entries + channel_info_rest, + ), ..make_prompt_context_no_owner() }); // ── FlushBatch: no thread tags → DM non-reply path in ──────────────── - // fetch_conversation_context. The event pubkey is a fresh key so the - // profile lookup will attempt one additional `/query` request, which - // also contributes to the request count but does not affect correctness. + // fetch_conversation_context. A fresh event pubkey triggers a kind-0 + // profile lookup (always issued, never counted in dm_requests). let event = { let event_keys = nostr::Keys::generate(); nostr::EventBuilder::new(nostr::Kind::Custom(1), "hello dm") @@ -7774,16 +7796,21 @@ mod tests { let _result = result_rx.try_recv().expect("PromptResult must be sent"); // ── Assertion: DM fetch (kind-40002 query) must have reached the server ─ - // This is the mutation tripwire: changing pool.rs:1942 to pass `false` - // skips the DM fetch entirely so dm_requests stays 0. Profile-lookup - // requests (kind 0) are issued regardless and are NOT counted here. + // Driven by the fail-closed path: resolve() returns None → + // .unwrap_or(true) → is_dm_turn = true → forwarded to + // fetch_conversation_context at pool.rs:1942 → kind-40002 query issued. + // + // Mutation 1 — pool.rs:1622 .unwrap_or(false): is_dm_turn = false → + // DM fetch skipped → dm_requests stays 0 → assertion FAILS. + // Mutation 2 — pool.rs:1942 hardcoded false: same outcome → FAILS. let total_dm_requests = dm_requests.load(Ordering::SeqCst); assert!( total_dm_requests >= 1, - "run_prompt_task with a DM channel must issue at least one DM history \ - query (containing kind 40002) via fetch_conversation_context; \ - got {total_dm_requests} — if this is 0, the is_dm_turn forwarding \ - at pool.rs:1942 has regressed" + "run_prompt_task with unresolved channel metadata must issue at least \ + one DM history query (kind 40002) via the fail-closed is_dm_turn path; \ + got {total_dm_requests} — if this is 0, either the fail-closed \ + .unwrap_or(true) at pool.rs:1622 or the is_dm_turn forwarding at \ + pool.rs:1942 has regressed" ); server.abort();