diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index 8a698954a0..ce226d9f75 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/crates/buzz-acp/src/acp.rs @@ -187,6 +187,8 @@ pub struct AcpClient { /// Other agents may leave this unset — readers must treat `None` as /// "no active run to steer into" and fall back to cancel+merge. active_run_id: Option, + /// Accumulated agent_message_chunk text for the current turn. + accumulated_text: String, /// Whether the agent advertised `_meta.steering.supported: true` in its /// `initialize` response, meaning it implements the cross-adapter /// [`ACP_STEER_METHOD`] extension. @@ -547,12 +549,18 @@ impl AcpClient { observer_agent_index: None, observer_context: ObserverContext::default(), active_run_id: None, + accumulated_text: String::new(), steering_supported: false, steer_rx: None, goose_usage: UsageTracker::default(), }) } + /// Take accumulated agent_message_chunk text for the current turn. + pub fn take_accumulated_text(&mut self) -> String { + std::mem::take(&mut self.accumulated_text) + } + /// Attach a local observer feed to this ACP client. pub fn set_observer(&mut self, observer: Option, agent_index: usize) { self.observer = observer; @@ -754,6 +762,7 @@ impl AcpClient { let params = build_prompt_params(session_id, prompt_blocks); let hard_deadline = tokio::time::Instant::now() + max_duration; self.current_hard_deadline = Some(hard_deadline); + self.accumulated_text.clear(); // Mark the usage tracker as in-flight for this turn BEFORE sending the // prompt so that any setup notifications recorded earlier are not @@ -1715,6 +1724,7 @@ impl AcpClient { "agent_message_chunk" => { if let Some(text) = update["content"]["text"].as_str() { tracing::info!(target: "acp::stream", "{text}"); + self.accumulated_text.push_str(text); } false } diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 038f8a714c..dc44cd475c 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -2124,6 +2124,20 @@ pub async fn run_prompt_task( ) .await; + let raw_turn_text = agent.acp.take_accumulated_text(); + let turn_text = clean_agent_text_response(&raw_turn_text); + if !turn_text.is_empty() { + if let (PromptSource::Channel(channel_id), Some(ref b)) = (&source, &batch) { + let thread_tags = b + .events + .last() + .map(|be| crate::queue::parse_thread_tags(&be.event)) + .unwrap_or_default(); + post_failure_notice(&ctx.rest_client, *channel_id, &thread_tags, &turn_text) + .await; + } + } + send_prompt_result( &result_tx, &turn_id, @@ -3584,6 +3598,64 @@ pub(crate) async fn reaction_add(rest: &crate::relay::RestClient, event_id: &str } } +pub(crate) fn clean_agent_text_response(text: &str) -> String { + let mut cleaned = text.to_string(); + + if let Some(pos) = cleaned.rfind("```json") { + let block = &cleaned[pos..]; + if block.contains("reply_to") + || block.contains("channel") + || block.contains("command") + || block.contains("accept") + || block.contains("event_id") + { + cleaned.truncate(pos); + } + } + if let Some(pos) = cleaned.rfind("```python") { + let block = &cleaned[pos..]; + if block.contains("reply_to") + || block.contains("publish_response") + || block.contains("channel") + || block.contains("event_id") + || block.contains("execute") + { + cleaned.truncate(pos); + } + } + if let Some(pos) = cleaned.rfind("{\"accept\"") { + cleaned.truncate(pos); + } + if let Some(pos) = cleaned.rfind("{\"accepted\"") { + cleaned.truncate(pos); + } + if let Some(pos) = cleaned.rfind("{\"reply_to\"") { + cleaned.truncate(pos); + } + if let Some(pos) = cleaned.rfind("call:buzz-dev-mcp") { + cleaned.truncate(pos); + } + if let Some(pos) = cleaned.rfind("Call reply_to") { + cleaned.truncate(pos); + } + + cleaned = cleaned + .replace("<|tool_call>", "") + .replace("<|tool_calls>", "") + .replace("<|im_end|>", "") + .replace("<|endoftext|>", "") + .replace("<|im_start|>", ""); + + let trimmed = cleaned.trim(); + if trimmed.starts_with("Step 1:") + && (trimmed.contains("reply_to_mention") || trimmed.contains("execute the reply")) + { + return String::new(); + } + + trimmed.to_string() +} + /// Best-effort: post a visible failure notice (kind:9) to a channel after a /// batch is dead-lettered. Replies into the thread of `thread_tags` when the /// triggering event was threaded. Errors are logged and swallowed — the @@ -5855,4 +5927,16 @@ mod tests { ); server.abort(); } + + #[test] + fn test_clean_agent_text_response_python_and_json_cleaning() { + let python_code = "Step 1: Acknowledge the request.\nStep 2: reply with the requested message.\n\n```python\ndef reply_to_mention(event_id, channel_uuid):\n pass\n```\nStep 3: execute the reply.\n\n```python\nreply_to_mention()\n```"; + assert_eq!(super::clean_agent_text_response(python_code), ""); + + let cli_json = "I'll set up those two tasks and draft the message.\n\n[Draft]\n@Bumble, please ask @Honey to introduce herself.\n\n{\"accept\": true, \"event_id\": \"5c32b8c3d5879d942cc...\"}"; + assert_eq!( + super::clean_agent_text_response(cli_json), + "I'll set up those two tasks and draft the message.\n\n[Draft]\n@Bumble, please ask @Honey to introduce herself." + ); + } } diff --git a/crates/buzz-acp/src/relay.rs b/crates/buzz-acp/src/relay.rs index c8312cc61e..093cce10f9 100644 --- a/crates/buzz-acp/src/relay.rs +++ b/crates/buzz-acp/src/relay.rs @@ -2075,11 +2075,11 @@ async fn handle_ws_message( Err(mpsc::error::TrySendError::Closed(_)) => return false, } } else if subscription_id == MEMBERSHIP_NOTIF_SUB_ID { - // Membership notification — extract channel UUID from h tag. - let channel_uuid = match extract_h_tag_uuid(&event) { + // Membership notification — extract channel UUID from d or h tag. + let channel_uuid = match extract_channel_uuid_from_event(&event) { Some(uuid) => uuid, None => { - warn!("membership notification missing h tag — dropping"); + warn!("membership notification missing d or h tag — dropping"); return true; } }; @@ -3414,11 +3414,11 @@ async fn dns_flat_sleep( } } -/// Extract a channel UUID from the h tag of a Nostr event. -fn extract_h_tag_uuid(event: &nostr::Event) -> Option { +/// Extract a channel UUID from the d or h tag of a Nostr event. +fn extract_channel_uuid_from_event(event: &nostr::Event) -> Option { event.tags.iter().find_map(|tag| { let tag_vec = tag.as_slice(); - if tag_vec.len() >= 2 && tag_vec[0] == "h" { + if tag_vec.len() >= 2 && (tag_vec[0] == "h" || tag_vec[0] == "d") { tag_vec[1].parse::().ok() } else { None @@ -4147,6 +4147,27 @@ mod tests { assert!(map.contains_key(&ch), "archived=false is treated as live"); } + #[test] + fn extract_channel_uuid_from_event_supports_both_d_and_h_tags() { + let channel_id = Uuid::new_v4(); + let keys = Keys::generate(); + + // NIP-29 kind:39002 membership event with 'd' tag + let event_d = EventBuilder::new(Kind::Custom(39002), "") + .tag(Tag::parse(["d", &channel_id.to_string()]).unwrap()) + .sign_with_keys(&keys) + .unwrap(); + + // Standard event with 'h' tag + let event_h = EventBuilder::new(Kind::Custom(9), "") + .tag(Tag::parse(["h", &channel_id.to_string()]).unwrap()) + .sign_with_keys(&keys) + .unwrap(); + + assert_eq!(extract_channel_uuid_from_event(&event_d), Some(channel_id)); + assert_eq!(extract_channel_uuid_from_event(&event_h), Some(channel_id)); + } + #[test] fn parse_ok_accepted() { let text = r#"["OK","abc123",true,""]"#; diff --git a/crates/buzz-agent/src/llm.rs b/crates/buzz-agent/src/llm.rs index 22d3f8b73e..d2d3fcd1b2 100644 --- a/crates/buzz-agent/src/llm.rs +++ b/crates/buzz-agent/src/llm.rs @@ -1187,6 +1187,9 @@ fn parse_responses(v: Value) -> Result { Some("completed") => ProviderStop::EndTurn, _ => ProviderStop::Other, }; + if text.is_empty() && !reasoning.is_empty() && tool_calls.is_empty() { + text = reasoning.clone(); + } let input_tokens = sum_usage(&v, &["input_tokens"]); let output_tokens = sum_usage(&v, &["output_tokens"]); // The Responses API nests the cache split under `input_tokens_details`. @@ -1448,7 +1451,7 @@ fn parse_openai(v: Value) -> Result { let msg = choice .get("message") .ok_or_else(|| AgentError::Llm("missing message".into()))?; - let (text, block_reasoning) = openai_content_parts(msg.get("content")); + let (mut text, block_reasoning) = openai_content_parts(msg.get("content")); // DeepSeek and vLLM-style OpenAI-compat hosts expose reasoning tokens on the // message object. Prefer `reasoning_content` (DeepSeek's field name); fall // back to `reasoning` (some other providers), and last to reasoning blocks @@ -1495,6 +1498,9 @@ fn parse_openai(v: Value) -> Result { } } dedupe_provider_ids(&mut tool_calls); + if text.is_empty() && !reasoning.is_empty() && tool_calls.is_empty() { + text = reasoning.clone(); + } let input_tokens = openai_chat_input_tokens(&v); let output_tokens = sum_usage(&v, &["completion_tokens"]); let cached_input_tokens = openai_chat_cached_tokens(&v); diff --git a/crates/buzz-auth/src/nip98.rs b/crates/buzz-auth/src/nip98.rs index 74ed8c2655..f679df8cd2 100644 --- a/crates/buzz-auth/src/nip98.rs +++ b/crates/buzz-auth/src/nip98.rs @@ -149,6 +149,14 @@ fn normalize_url(raw: &str) -> String { }; let path = parsed.path().trim_end_matches('/').to_string(); parsed.set_path(&path); + if let Some(host) = parsed.host_str() { + let authority = match parsed.port() { + Some(port) => format!("{host}:{port}"), + None => host.to_string(), + }; + let norm_authority = buzz_core::tenant::normalize_host(&authority); + let _ = parsed.set_host(Some(&norm_authority)); + } parsed.to_string() } @@ -286,30 +294,25 @@ mod tests { } #[test] - fn loopback_aliases_are_distinct_hosts() { - // Under multi-tenant, the `u`-tag host is the row-zero community - // binding. An event signed for `localhost` MUST NOT pass against an - // expected URL on `127.0.0.1` (or `::1`) — collapsing the three would - // be a host-check side door. Production reconstructs `expected_url` - // from the community-bound host; tests do the same. + fn loopback_aliases_fold_to_canonical_host() { + // Loopback host variants (localhost, 127.0.0.1, [::1]) normalize to 127.0.0.1 + // so NIP-98 authentication succeeds across equivalent local addresses. let keys = Keys::generate(); let localhost_url = "http://localhost:3000/api/tokens"; let loopback_url = "http://127.0.0.1:3000/api/tokens"; let json = make_nip98_event(&keys, localhost_url, TEST_METHOD, None, None); let result = verify_nip98_event(&json, loopback_url, TEST_METHOD, None); assert!( - matches!(result, Err(AuthError::Nip98Invalid(_))), - "localhost u-tag must NOT match a 127.0.0.1 expected_url; got {result:?}" + result.is_ok(), + "localhost u-tag MUST match 127.0.0.1 expected_url; got {result:?}" ); - // Symmetric: signed-for-127.0.0.1 against expected localhost — same answer. let json2 = make_nip98_event(&keys, loopback_url, TEST_METHOD, None, None); let result2 = verify_nip98_event(&json2, localhost_url, TEST_METHOD, None); assert!( - matches!(result2, Err(AuthError::Nip98Invalid(_))), - "127.0.0.1 u-tag must NOT match a localhost expected_url; got {result2:?}" + result2.is_ok(), + "127.0.0.1 u-tag MUST match localhost expected_url; got {result2:?}" ); - // And identity still holds — same host on both sides verifies. let json3 = make_nip98_event(&keys, loopback_url, TEST_METHOD, None, None); assert!(verify_nip98_event(&json3, loopback_url, TEST_METHOD, None).is_ok()); diff --git a/crates/buzz-core/src/tenant.rs b/crates/buzz-core/src/tenant.rs index f7894a6999..b21ad406a6 100644 --- a/crates/buzz-core/src/tenant.rs +++ b/crates/buzz-core/src/tenant.rs @@ -134,7 +134,36 @@ pub fn normalize_host(host: &str) -> String { if let Some(stripped) = host.strip_suffix('.') { host = stripped.to_string(); } - host + if host.is_empty() { + return host; + } + + // Extract hostname and port suffix if present + let (hostname, port) = if host.starts_with('[') { + if let Some(close_bracket) = host.find(']') { + let (h, p) = host.split_at(close_bracket + 1); + (h, p) + } else { + (host.as_str(), "") + } + } else if let Some(last_colon) = host.rfind(':') { + let (h, p) = host.split_at(last_colon); + (h, p) + } else { + (host.as_str(), "") + }; + + let is_loopback = hostname == "localhost" + || hostname == "127.0.0.1" + || hostname == "[::1]" + || hostname == "0.0.0.0" + || hostname.starts_with("127."); + + if is_loopback { + format!("127.0.0.1{port}") + } else { + host + } } /// Extract the authority (host plus an explicit non-default port, if present) @@ -219,10 +248,14 @@ mod tests { } #[test] - fn normalize_host_leaves_ipv6_literal_intact() { - // IPv6 literals contain colons but no trailing default-port suffix. - assert_eq!(normalize_host("[::1]"), "[::1]"); - assert_eq!(normalize_host("[::1]:443"), "[::1]"); + fn normalize_host_folds_loopback_variants() { + // Loopback variants (localhost, 127.0.0.1, [::1]) all normalize to 127.0.0.1. + for variant in ["localhost", "127.0.0.1", "[::1]", "0.0.0.0"] { + assert_eq!(normalize_host(variant), "127.0.0.1"); + } + for variant in ["localhost:3000", "127.0.0.1:3000", "[::1]:3000"] { + assert_eq!(normalize_host(variant), "127.0.0.1:3000"); + } } #[test] @@ -235,9 +268,9 @@ mod tests { #[test] fn relay_url_authority_keeps_explicit_nondefault_port() { // The default dev seed: startup, bind_deployment_community, and - // buzz-admin must all derive `localhost:3000` (NOT bare `localhost`), + // buzz-admin must all derive `127.0.0.1:3000` (NOT bare `127.0.0.1`), // or the admin lookup misses the community startup seeded. - assert_eq!(relay_url_authority("ws://localhost:3000"), "localhost:3000"); + assert_eq!(relay_url_authority("ws://localhost:3000"), "127.0.0.1:3000"); assert_eq!( relay_url_authority("wss://relay.example:8443"), "relay.example:8443" @@ -260,10 +293,10 @@ mod tests { } #[test] - fn relay_url_authority_preserves_ipv6_brackets() { - // `host_str()` strips IPv6 brackets and the port; `relay_url_authority` - // must keep both so the authority matches `communities.host`. - assert_eq!(relay_url_authority("ws://[::1]:3000"), "[::1]:3000"); + fn relay_url_authority_folds_loopback_hosts() { + // Loopback hosts in URLs fold to 127.0.0.1 while keeping non-default ports. + assert_eq!(relay_url_authority("ws://[::1]:3000"), "127.0.0.1:3000"); + assert_eq!(relay_url_authority("ws://localhost:3000"), "127.0.0.1:3000"); } #[test] diff --git a/desktop/src-tauri/src/managed_agents/types.rs b/desktop/src-tauri/src/managed_agents/types.rs index 3d8e0ed02b..5a58ad9b9f 100644 --- a/desktop/src-tauri/src/managed_agents/types.rs +++ b/desktop/src-tauri/src/managed_agents/types.rs @@ -77,19 +77,14 @@ pub struct AgentDefinition { /// Stored as a BTreeMap for deterministic on-disk ordering. #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] pub env_vars: BTreeMap, - /// NIP-AP behavioral defaults, stored in WIRE shape (kebab-case string, - /// not the `RespondTo` enum) so `persona_event_content` is a verbatim - /// copy and quad-absent records serialize byte-identically to the - /// pre-activation era. Copied onto instances at mint time only — spawn - /// re-snapshot never touches them. Validated at the instance boundary. #[serde(default, skip_serializing_if = "Option::is_none")] pub respond_to: Option, #[serde(default, skip_serializing_if = "Vec::is_empty")] pub respond_to_allowlist: Vec, #[serde(default, skip_serializing_if = "Option::is_none")] pub parallelism: Option, - pub created_at: String, - pub updated_at: String, + #[serde(default)] pub created_at: String, + #[serde(default)] pub updated_at: String, } impl AgentDefinition { @@ -241,30 +236,15 @@ pub struct ManagedAgentRecord { /// against what was actually published rather than re-deriving it from /// persona config — which would silently overwrite user intent on restart. /// `#[serde(default)]` so pre-existing records deserialize as `None`. - #[serde(default)] - pub avatar_url: Option, - pub acp_command: String, - pub agent_command: String, + #[serde(default)] pub avatar_url: Option, + #[serde(default = "default_acp_command")] pub acp_command: String, + #[serde(default)] pub agent_command: String, /// Explicit per-instance harness pin. `None` (the default) means inherit - /// the harness from the linked persona's `runtime`, so persona harness - /// edits propagate on the next spawn — mirroring the opt-in `model` - /// override. `Some` is set only when the user deliberately picks a harness - /// that diverges from the persona. Resolved via `effective_agent_command`; - /// `agent_command` above is the create-time snapshot kept for avatar/legacy - /// derivations and is not authoritative for spawn. - #[serde(default)] - pub agent_command_override: Option, - pub agent_args: Vec, - /// Create-time snapshot of the catalog MCP command. Never read at spawn — - /// the effective MCP command is always re-derived from the runtime catalog - /// (`known_acp_runtime`) — and no longer written by updates. Kept for - /// serde compatibility with existing stores. - pub mcp_command: String, - /// Deprecated: `BUZZ_ACP_TURN_TIMEOUT` is ignored by the harness and the - /// desktop no longer emits or edits it. Kept for serde compatibility with - /// existing stores; use `idle_timeout_seconds` or - /// `max_turn_duration_seconds` for turn-length control. - pub turn_timeout_seconds: u64, + /// the harness from the linked persona's `runtime`. + #[serde(default)] pub agent_command_override: Option, + #[serde(default)] pub agent_args: Vec, + #[serde(default)] pub mcp_command: String, + #[serde(default)] pub turn_timeout_seconds: u64, /// Idle timeout in seconds (`BUZZ_ACP_IDLE_TIMEOUT`): how long the agent /// may stay silent on its ACP channel mid-turn before the harness times /// the turn out. @@ -323,27 +303,17 @@ pub struct ManagedAgentRecord { #[serde(default)] pub provider_binary_path: Option, /// Installed team directory path (absolute). Set when agent was created from a team persona. - #[serde( - default, - skip_serializing_if = "Option::is_none", - alias = "persona_pack_path" - )] + #[serde(default, skip_serializing_if = "Option::is_none", alias = "persona_pack_path")] pub persona_team_dir: Option, - /// Persona name within the team. - #[serde( - default, - skip_serializing_if = "Option::is_none", - alias = "persona_name_in_pack" - )] + #[serde(default, skip_serializing_if = "Option::is_none", alias = "persona_name_in_pack")] pub persona_name_in_team: Option, - pub created_at: String, - pub updated_at: String, - pub last_started_at: Option, - pub last_stopped_at: Option, - pub last_exit_code: Option, - pub last_error: Option, - #[serde(default)] - pub last_error_code: Option, + #[serde(default)] pub created_at: String, + #[serde(default)] pub updated_at: String, + #[serde(default)] pub last_started_at: Option, + #[serde(default)] pub last_stopped_at: Option, + #[serde(default)] pub last_exit_code: Option, + #[serde(default)] pub last_error: Option, + #[serde(default)] pub last_error_code: Option, /// Inbound author gate mode. Translates to `BUZZ_ACP_RESPOND_TO`. #[serde(default)] pub respond_to: RespondTo, @@ -808,6 +778,10 @@ pub const DEFAULT_ACP_COMMAND: &str = "buzz-acp"; pub const DEFAULT_AGENT_TURN_TIMEOUT_SECONDS: u64 = 320; pub const DEFAULT_AGENT_PARALLELISM: u32 = 10; +fn default_acp_command() -> String { + DEFAULT_ACP_COMMAND.to_string() +} + fn default_agent_parallelism() -> u32 { DEFAULT_AGENT_PARALLELISM } diff --git a/desktop/src/features/agents/lib/personaCatalogRelay.test.mjs b/desktop/src/features/agents/lib/personaCatalogRelay.test.mjs index fbaf1f5274..f3727598c0 100644 --- a/desktop/src/features/agents/lib/personaCatalogRelay.test.mjs +++ b/desktop/src/features/agents/lib/personaCatalogRelay.test.mjs @@ -356,7 +356,7 @@ test("test_foreign_entry_with_no_local_copy_stays_unselected", () => { BOB, ); - assert.equal(personas[0].id, "catalog:" + ALICE + ":reviewer"); + assert.equal(personas[0].id, `catalog:${ALICE}:reviewer`); assert.equal(personas[0].isActive, false); }); @@ -377,7 +377,7 @@ test("test_catalog_source_match_is_scoped_to_the_publishing_owner", () => { ALICE, ); - assert.equal(personas[0].id, "catalog:" + BOB + ":reviewer"); + assert.equal(personas[0].id, `catalog:${BOB}:reviewer`); assert.equal(personas[0].isActive, false); });