fix(acp,agent): support reasoning_content fallback and turn text publishing for local reasoning models

.github/workflows/docker.yml

@@ -76,7 +76,8 @@ env:
   # Single source of truth for the image name. Set GHCR_IMAGE as a repo
   # variable to override (e.g., for forks that want to push to their own
   # namespace without forking this file).
-  IMAGE_NAME: ${{ vars.GHCR_IMAGE != '' && vars.GHCR_IMAGE || 'ghcr.io/block/buzz' }}
+  IMAGE_NAME: ${{ vars.GHCR_IMAGE != '' && vars.GHCR_IMAGE || format('ghcr.io/{0}', github.repository) }}
+  GATEWAY_IMAGE_NAME: ${{ vars.GHCR_GATEWAY_IMAGE != '' && vars.GHCR_GATEWAY_IMAGE || format('ghcr.io/{0}-push-gateway', github.repository) }}
 
 jobs:
   build:
@@ -387,7 +388,7 @@ jobs:
         id: meta
         uses: docker/metadata-action@80c7e94dd9b9319bd5eb7a0e0fe9291e23a2a2e9  # v6.1.0
         with:
-          images: ghcr.io/block/buzz-push-gateway
+          images: ${{ env.GATEWAY_IMAGE_NAME }}
           labels: |
             org.opencontainers.image.title=Buzz Push Gateway
             org.opencontainers.image.description=Capability-gated APNs last hop for Buzz
@@ -400,9 +401,9 @@ jobs:
           file: ./Dockerfile.push-gateway
           platforms: ${{ matrix.platform }}
           labels: ${{ steps.meta.outputs.labels }}
-          outputs: type=image,name=ghcr.io/block/buzz-push-gateway,push-by-digest=true,name-canonical=true,push=${{ github.event_name != 'pull_request' }}
-          cache-from: type=registry,ref=ghcr.io/block/buzz-push-gateway-buildcache:${{ matrix.arch }}
-          cache-to: ${{ (github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository) && format('type=registry,ref=ghcr.io/block/buzz-push-gateway-buildcache:{0},mode=max,compression=zstd', matrix.arch) || '' }}
+          outputs: type=image,name=${{ env.GATEWAY_IMAGE_NAME }},push-by-digest=true,name-canonical=true,push=${{ github.event_name != 'pull_request' }}
+          cache-from: type=registry,ref=${{ env.GATEWAY_IMAGE_NAME }}-buildcache:${{ matrix.arch }}
+          cache-to: ${{ (github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository) && format('type=registry,ref={0}-buildcache:{1},mode=max,compression=zstd', env.GATEWAY_IMAGE_NAME, matrix.arch) || '' }}
       - name: Export digest
         if: github.event_name != 'pull_request'
         env:
@@ -447,7 +448,7 @@ jobs:
         id: meta
         uses: docker/metadata-action@80c7e94dd9b9319bd5eb7a0e0fe9291e23a2a2e9  # v6.1.0
         with:
-          images: ghcr.io/block/buzz-push-gateway
+          images: ${{ env.GATEWAY_IMAGE_NAME }}
           tags: |
             type=ref,event=branch,enable=${{ github.event_name != 'workflow_dispatch' || inputs.version == '' }}
             type=sha,prefix=sha-,format=short,enable=${{ github.event_name != 'workflow_dispatch' || inputs.version == '' }}
@@ -461,15 +462,15 @@ jobs:
         run: |
           set -euo pipefail
           tags=(); while IFS= read -r tag; do [ -n "$tag" ] && tags+=("-t" "$tag"); done <<< "$META_TAGS"
-          digests=(); for digest in *; do digests+=("ghcr.io/block/buzz-push-gateway@sha256:${digest}"); done
+          digests=(); for digest in *; do digests+=("${{ env.GATEWAY_IMAGE_NAME }}@sha256:${digest}"); done
           docker buildx imagetools create "${tags[@]}" "${digests[@]}"
           first_tag=$(echo "$META_TAGS" | head -n1)
           digest=$(docker buildx imagetools inspect "$first_tag" --format '{{json .Manifest}}' | jq -r '.digest')
           echo "digest=${digest}" >> "$GITHUB_OUTPUT"
       - name: Attest gateway image provenance
         uses: actions/attest-build-provenance@0f67c3f4856b2e3261c31976d6725780e5e4c373  # v4.1.1
         with:
-          subject-name: ghcr.io/block/buzz-push-gateway
+          subject-name: ${{ env.GATEWAY_IMAGE_NAME }}
           subject-digest: ${{ steps.manifest.outputs.digest }}
           push-to-registry: true
       - name: Gateway publication summary
@@ -479,7 +480,7 @@ jobs:
         run: |
           set -euo pipefail
           {
-            echo "### Published \`ghcr.io/block/buzz-push-gateway\`"
+            echo "### Published \`${{ env.GATEWAY_IMAGE_NAME }}\`"
             echo
             printf "**Digest:** \`%s\`\n" "$GATEWAY_DIGEST"
             echo
@@ -490,6 +491,6 @@ jobs:
             echo
             echo 'Verify provenance before deployment:'
             echo "\`\`\`"
-            printf 'gh attestation verify oci://ghcr.io/block/buzz-push-gateway@%s --owner block\n' "$GATEWAY_DIGEST"
+            printf 'gh attestation verify oci://${{ env.GATEWAY_IMAGE_NAME }}@%s --owner ${{ github.repository_owner }}\n' "$GATEWAY_DIGEST"
             echo "\`\`\`"
           } >> "$GITHUB_STEP_SUMMARY"

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<String>,
+    /// 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<ObserverHandle>, 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
             }

crates/buzz-acp/src/pool.rs

@@ -2128,6 +2128,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,
@@ -3800,6 +3814,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
@@ -6749,4 +6821,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."
+        );
+    }
 }

crates/buzz-acp/src/relay.rs

@@ -2088,11 +2088,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;
                             }
                         };
@@ -3427,11 +3427,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<Uuid> {
+/// Extract a channel UUID from the d or h tag of a Nostr event.
+fn extract_channel_uuid_from_event(event: &nostr::Event) -> Option<Uuid> {
     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::<Uuid>().ok()
         } else {
             None
@@ -4160,6 +4160,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,""]"#;

crates/buzz-agent/src/llm.rs

@@ -1250,6 +1250,9 @@ fn parse_responses(v: Value) -> Result<LlmResponse, AgentError> {
         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`.
@@ -1553,7 +1556,7 @@ fn parse_openai(v: Value) -> Result<LlmResponse, AgentError> {
     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
@@ -1600,6 +1603,9 @@ fn parse_openai(v: Value) -> Result<LlmResponse, AgentError> {
         }
     }
     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);