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..71ead9aaa1 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,68 @@ 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() + } + + /// 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()) + } + } } } @@ -455,18 +507,47 @@ pub enum PromptOutcome { CancelDrainTimeout(Duration), } -/// Immutable config subset shared (via `Arc`) by all spawned prompt tasks. +/// 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. /// -/// Built once from `Config` at startup. Avoids cloning the full config -/// into every task. +/// 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>>, + /// 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, } @@ -475,42 +556,119 @@ 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, }, )) }) .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, } } 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 — 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); } + // 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. + /// + /// 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) => { + 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; + } + } + } + } + // 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, @@ -564,6 +722,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 +1001,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,18 +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`). -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 +1029,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 +1328,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 +1415,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,39 +1610,42 @@ 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. + // 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 non-DM channel turns. // - // 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`. - 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)); - } - } + // 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) if !is_dm_turn => { + let result = fetch_canvas_pointer(*cid, &ctx.rest_client).await; + ctx.canvas_cache.resolve_for_turn(cid, result) } - } + _ => None, + }; // The core section to fold into the system prompt for this turn's session. // Channel-scoped; heartbeats carry no owner core. @@ -1544,18 +1654,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 +1667,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 +1677,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 +1711,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 +1772,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 +1787,21 @@ 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. + // 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 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(), + ); + format!("{ctx_block}\n\n{init_msg}") + }; let init_result = agent .acp .session_prompt_with_idle_timeout( @@ -1835,11 +1934,12 @@ 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 + fetch_conversation_context(b, &ctx, is_dm_turn).await } else { None }; @@ -1874,7 +1974,8 @@ 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(), + is_dm: is_dm_turn, }, ) } else { @@ -2366,17 +2467,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 +2507,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,9 +2527,10 @@ 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,31 +2560,39 @@ async fn fetch_canvas_section(channel_id: Uuid, rest: &RestClient) -> Option 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) => { @@ -2485,38 +2600,41 @@ 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 — 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 section", + "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 section", + "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 @@ -2525,19 +2643,20 @@ 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 — 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 section" + "latest canvas event has blank content — canvas absent" ); - return None; + return CanvasFetchResult::Absent; } let id = event.id.to_hex(); @@ -2546,15 +2665,16 @@ pub(crate) fn canvas_section_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 section", + "canvas event created_at overflows i64 — serving stale", ); - return None; + return CanvasFetchResult::Failed; } }; let timestamp = match chrono::DateTime::from_timestamp(ts_secs, 0) { @@ -2564,9 +2684,9 @@ 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 — serving stale", ); - return None; + return CanvasFetchResult::Failed; } }; @@ -2574,22 +2694,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}" - ) + CanvasFetchResult::Present(crate::queue::CanvasPointer { + event_id: id, + timestamp, + }) } /// Fetch conversation context (thread or DM) for a batch before prompting. @@ -2603,14 +2713,10 @@ pub(crate) fn render_canvas_section(event_id: &str, timestamp: &str, channel_uui /// 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) @@ -2629,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; } @@ -4068,80 +4174,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 +6489,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,62 +6512,75 @@ 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 = 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!(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); - assert!(result.is_none()); + fn test_canvas_pointer_from_query_response_empty_array_returns_absent() { + let result = canvas_pointer_from_query_response(&[], CHANNEL_UUID); + assert!( + matches!(result, CanvasFetchResult::Absent), + "empty array must be Absent (confirmed no canvas)" + ); } #[test] - fn test_canvas_section_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_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)" + matches!(result, CanvasFetchResult::Absent), + "blank content must be Absent (cleared canvas)" ); } #[test] - fn test_canvas_section_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_section_from_query_response(&[ev], CHANNEL_UUID); - assert!(result.is_none()); + let result = canvas_pointer_from_query_response(&[ev], CHANNEL_UUID); + 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_section_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, "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" + 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_section_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( @@ -6644,17 +6592,17 @@ 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" + 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_section_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( @@ -6665,19 +6613,19 @@ 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" + 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_section_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( @@ -6688,17 +6636,17 @@ 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" + 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_section_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( @@ -6713,16 +6661,16 @@ 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" + 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_section_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( @@ -6732,14 +6680,17 @@ mod tests { .expect("sign"), ) .expect("serialise"); - let result = canvas_section_from_query_response(&[ev], CHANNEL_UUID); - assert!(result.is_none(), "wrong kind must return None"); + let result = canvas_pointer_from_query_response(&[ev], CHANNEL_UUID); + 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_section_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( @@ -6749,63 +6700,274 @@ mod tests { .expect("sign"), ) .expect("serialise"); - let result = canvas_section_from_query_response(&[ev], CHANNEL_UUID); - assert!(result.is_none(), "mismatched h-tag must return None"); + let result = canvas_pointer_from_query_response(&[ev], CHANNEL_UUID); + assert!( + matches!(result, CanvasFetchResult::Failed), + "mismatched h-tag must be Failed — not Absent" + ); } #[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 = match result { + CanvasFetchResult::Present(p) => p, + other => panic!("expected Present, got {other:?}"), + }; 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" ); } - // ── new-session channel context (one resolve, two consumers) ───────────── + // ── CanvasRevisionCache tri-state (A1) ──────────────────────────────────── - /// A [`ChannelInfoResolver`] whose lazy REST fallback is served by a local - /// HTTP server, plus a counter of the requests that actually reached it. - /// Counting real requests is the point: the composition tests are pure and - /// cannot see duplicated I/O. - async fn counting_resolver( - response: serde_json::Value, - ) -> ( - ChannelInfoResolver, - std::sync::Arc, - tokio::task::JoinHandle<()>, - ) { - use std::sync::atomic::{AtomicUsize, Ordering}; - use tokio::io::{AsyncReadExt, AsyncWriteExt}; + #[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: a Failed fetch returns the cached pointer as stale. + let stale = cache.resolve_for_turn(&ch, CanvasFetchResult::Failed); + assert_eq!(stale, Some(pointer)); + } - let listener = tokio::net::TcpListener::bind("127.0.0.1:0") - .await - .expect("bind test HTTP server"); - let base_url = format!("http://{}", listener.local_addr().unwrap()); - let requests = std::sync::Arc::new(AtomicUsize::new(0)); - let server_requests = requests.clone(); - let body = response.to_string(); - let server = tokio::spawn(async move { - while let Ok((mut socket, _)) = listener.accept().await { - let mut buf = vec![0; 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 rest = crate::relay::RestClient { - http: reqwest::Client::new(), + #[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"); + // 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] + 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"); + } + + // ── 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) ───────────── + + /// A [`ChannelInfoResolver`] whose lazy REST fallback is served by a local + /// HTTP server, plus a counter of the requests that actually reached it. + /// Counting real requests is the point: the composition tests are pure and + /// cannot see duplicated I/O. + async fn counting_resolver( + response: serde_json::Value, + ) -> ( + ChannelInfoResolver, + std::sync::Arc, + tokio::task::JoinHandle<()>, + ) { + 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 test HTTP server"); + let base_url = format!("http://{}", listener.local_addr().unwrap()); + let requests = std::sync::Arc::new(AtomicUsize::new(0)); + let server_requests = requests.clone(); + let body = response.to_string(); + let server = tokio::spawn(async move { + while let Ok((mut socket, _)) = listener.accept().await { + let mut buf = vec![0; 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 rest = crate::relay::RestClient { + http: reqwest::Client::new(), base_url, keys: nostr::Keys::generate(), auth_tag_json: None, @@ -6823,23 +6985,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 +7011,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 +7056,763 @@ 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(); + // 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(); + } + + // ── 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}" + ); + } + + // ── 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}" + ); + } + + // ── 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(); + } + + // ── 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 **unresolved** channel metadata and a + /// scripted bash agent. + /// + /// ## How fail-close is exercised + /// + /// 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: 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 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_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]); + + 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{}", + 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). 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' + 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 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 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( + 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. 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") + .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 ─ + // 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 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(); + } } diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index 5c960de202..6147551028 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. @@ -1235,13 +1249,14 @@ 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, 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,88 @@ 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 + )); +} + /// Format a conversation context section (thread or DM). fn format_conversation_context( ctx: &ConversationContext, @@ -1370,13 +1454,19 @@ 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. + /// + /// 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>, + /// Authoritative DM classification for this turn, computed once at turn + /// start with fail-closed semantics (unresolved metadata → `true`). /// - /// 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>, + /// 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. @@ -1421,10 +1511,7 @@ pub fn format_prompt(batch: &FlushBatch, args: &FormatPromptArgs<'_>) -> Vec = Vec::with_capacity(7); @@ -1456,10 +1543,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 } @@ -405,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). 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