fix: workflow reactions, Windows Claude shim, oversized frames

crates/buzz-relay/src/workflow_sink.rs

@@ -8,7 +8,7 @@ use std::future::Future;
 use std::pin::Pin;
 use std::sync::{Arc, Weak};
 
-use buzz_core::kind::KIND_STREAM_MESSAGE;
+use buzz_core::kind::{KIND_REACTION, KIND_STREAM_MESSAGE};
 use buzz_core::tenant::CommunityId;
 use buzz_workflow::action_sink::{ActionSink, ActionSinkError};
 use chrono::Utc;
@@ -362,6 +362,145 @@ impl ActionSink for RelayActionSink {
             Ok(event_id_hex)
         })
     }
+
+    fn add_reaction(
+        &self,
+        community_id: CommunityId,
+        message_id: &str,
+        emoji: &str,
+        author_pubkey: &str,
+    ) -> Pin<Box<dyn Future<Output = Result<String, ActionSinkError>> + Send + '_>> {
+        let message_id = message_id.to_owned();
+        let emoji = emoji.to_owned();
+        let author_pubkey = author_pubkey.to_owned();
+
+        Box::pin(async move {
+            // 0. Upgrade weak reference — fails only during shutdown.
+            let state = self
+                .state
+                .upgrade()
+                .ok_or_else(|| ActionSinkError::Database("relay is shutting down".into()))?;
+
+            // 1. Resolve community → TenantContext.
+            let host = state
+                .db
+                .lookup_community_host(community_id)
+                .await
+                .map_err(|e| ActionSinkError::Database(e.to_string()))?
+                .ok_or_else(|| {
+                    ActionSinkError::Database(format!(
+                        "workflow run community {community_id} is not mapped to a host"
+                    ))
+                })?;
+            let tenant = buzz_core::tenant::TenantContext::resolved(community_id, host);
+
+            // 2. Parse author pubkey.
+            let author_pubkey_key = nostr::PublicKey::from_hex(&author_pubkey).map_err(|e| {
+                ActionSinkError::InvalidInput(format!("invalid author pubkey: {e}"))
+            })?;
+            let author_pubkey_hex = author_pubkey_key.to_hex();
+            let author_bytes = author_pubkey_key.to_bytes().to_vec();
+
+            // 3. Parse message_id as EventId and look up target event.
+            let target_event_id = nostr::EventId::from_hex(&message_id)
+                .map_err(|e| ActionSinkError::InvalidInput(format!("invalid message_id: {e}")))?;
+            let target_bytes = target_event_id.as_bytes().to_vec();
+
+            let target = state
+                .db
+                .get_event_by_id(tenant.community(), &target_bytes)
+                .await
+                .map_err(|e| ActionSinkError::Database(e.to_string()))?
+                .ok_or_else(|| {
+                    ActionSinkError::InvalidInput(format!(
+                        "reaction target event not found: {message_id}"
+                    ))
+                })?;
+
+            // 4. Normalise emoji: empty → "+"; trim; reject if too long.
+            let emoji_normalised = if emoji.is_empty() {
+                "+".to_owned()
+            } else {
+                emoji.trim().to_owned()
+            };
+            if emoji_normalised.chars().count() > 64 {
+                return Err(ActionSinkError::InvalidInput(
+                    "emoji exceeds 64 characters".into(),
+                ));
+            }
+
+            // 5. Build kind:7 event.
+            //    Tags: e (target), p (author), buzz:workflow true.
+            let tags = vec![
+                Tag::parse(["e", &message_id])
+                    .map_err(|e| ActionSinkError::EventBuild(format!("e tag: {e}")))?,
+                Tag::parse(["p", &author_pubkey_hex])
+                    .map_err(|e| ActionSinkError::EventBuild(format!("p tag: {e}")))?,
+                Tag::parse(["buzz:workflow", "true"])
+                    .map_err(|e| ActionSinkError::EventBuild(format!("workflow tag: {e}")))?,
+            ];
+
+            let kind = Kind::from(KIND_REACTION as u16);
+            let event = EventBuilder::new(kind, &emoji_normalised)
+                .tags(tags)
+                .sign_with_keys(&state.relay_keypair)
+                .map_err(|e| ActionSinkError::EventBuild(format!("signing: {e}")))?;
+
+            let event_id_hex = event.id.to_hex();
+
+            info!(
+                event_id = %event_id_hex,
+                target = %message_id,
+                author = %author_pubkey_hex,
+                emoji = %emoji_normalised,
+                "Workflow AddReaction: posting kind:7 event"
+            );
+
+            // 6. Persist via insert_reaction_event_with_thread_metadata.
+            let channel_id = target.channel_id;
+            match state
+                .db
+                .insert_reaction_event_with_thread_metadata(
+                    tenant.community(),
+                    &event,
+                    channel_id,
+                    None,
+                    &target_bytes,
+                    &author_bytes,
+                    &emoji_normalised,
+                )
+                .await
+                .map_err(|e| ActionSinkError::Database(e.to_string()))?
+            {
+                buzz_db::ReactionEventInsertOutcome::Inserted {
+                    stored_event,
+                    was_inserted,
+                } => {
+                    if was_inserted {
+                        let _ = dispatch_persistent_event(
+                            &tenant,
+                            &state,
+                            &stored_event,
+                            KIND_REACTION,
+                            &author_pubkey_hex,
+                            None,
+                        )
+                        .await;
+                    }
+                    Ok(event_id_hex)
+                }
+                buzz_db::ReactionEventInsertOutcome::Duplicate => {
+                    // Idempotent — return the original message_id.
+                    Ok(message_id)
+                }
+                buzz_db::ReactionEventInsertOutcome::TargetMissing => {
+                    Err(ActionSinkError::InvalidInput(format!(
+                        "reaction target event not found: {message_id}"
+                    )))
+                }
+            }
+        })
+    }
 }
 
 #[cfg(test)]

crates/buzz-workflow/src/action_sink.rs

@@ -66,4 +66,22 @@ pub trait ActionSink: Send + Sync {
         text: &str,
         author_pubkey: &str,
     ) -> Pin<Box<dyn Future<Output = Result<String, ActionSinkError>> + Send + '_>>;
+
+    /// Add a reaction to a message on behalf of a workflow owner.
+    ///
+    /// - `community_id`: the community that owns the workflow run
+    /// - `message_id`: hex event ID of the target message
+    /// - `emoji`: reaction emoji (empty → "+"; must be ≤ 64 chars)
+    /// - `author_pubkey`: hex-encoded pubkey of the workflow owner (used for
+    ///   the `p` attribution tag; the relay keypair signs the kind:7 event)
+    ///
+    /// Returns the event ID hex string on success. Duplicate reactions are
+    /// treated as idempotent and return the original message_id.
+    fn add_reaction(
+        &self,
+        community_id: CommunityId,
+        message_id: &str,
+        emoji: &str,
+        author_pubkey: &str,
+    ) -> Pin<Box<dyn Future<Output = Result<String, ActionSinkError>> + Send + '_>>;
 }

crates/buzz-workflow/src/executor.rs

@@ -597,23 +597,42 @@ pub async fn dispatch_action(
                 ));
             }
 
-            #[cfg(feature = "reqwest")]
-            {
-                let result = add_reaction_impl(&trigger_ctx.message_id, emoji).await?;
-                Ok(StepResult::Completed(result))
-            }
+            let wf_run = engine
+                .db
+                .get_workflow_run(community_id, run_id)
+                .await
+                .map_err(|e| {
+                    WorkflowError::WebhookError(format!(
+                        "AddReaction: failed to load workflow run {run_id}: {e}"
+                    ))
+                })?;
+            let workflow = engine
+                .db
+                .get_workflow(community_id, wf_run.workflow_id)
+                .await
+                .map_err(|e| {
+                    WorkflowError::WebhookError(format!(
+                        "AddReaction: failed to load workflow {}: {e}",
+                        wf_run.workflow_id
+                    ))
+                })?;
+            let owner_pubkey_hex = hex::encode(&workflow.owner_pubkey);
 
-            #[cfg(not(feature = "reqwest"))]
-            {
-                warn!(
-                    run_id = %run_id,
-                    step = step_id,
-                    "AddReaction: reqwest feature not enabled, skipping HTTP call"
-                );
-                Ok(StepResult::Completed(
-                    serde_json::json!({ "added": false, "skipped": true }),
-                ))
-            }
+            let event_id = engine
+                .action_sink()?
+                .add_reaction(
+                    community_id,
+                    &trigger_ctx.message_id,
+                    emoji,
+                    &owner_pubkey_hex,
+                )
+                .await
+                .map_err(WorkflowError::from)?;
+
+            Ok(StepResult::Completed(serde_json::json!({
+                "added": true,
+                "event_id": event_id,
+            })))
         }
 
         CallWebhook {
@@ -865,70 +884,6 @@ async fn call_webhook_impl(
     }))
 }
 
-/// Returns a shared `reqwest::Client` reused across all workflow HTTP calls.
-/// Sharing a single client reuses the underlying connection pool.
-#[cfg(feature = "reqwest")]
-fn shared_http_client() -> &'static reqwest::Client {
-    use std::sync::LazyLock;
-    use std::time::Duration;
-    static CLIENT: LazyLock<reqwest::Client> = LazyLock::new(|| {
-        reqwest::Client::builder()
-            .timeout(Duration::from_secs(10))
-            .build()
-            .expect("HTTP client build must succeed")
-    });
-    &CLIENT
-}
-
-/// POST `{"emoji": emoji}` to `POST /api/messages/{message_id}/reactions`.
-#[cfg(feature = "reqwest")]
-async fn add_reaction_impl(message_id: &str, emoji: &str) -> Result<JsonValue, WorkflowError> {
-    let base_url =
-        std::env::var("BUZZ_RELAY_BASE_URL").unwrap_or_else(|_| "http://localhost:3000".to_owned());
-
-    let url = format!("{base_url}/api/messages/{message_id}/reactions");
-
-    let client = shared_http_client();
-
-    let mut req = client
-        .post(&url)
-        .header("Content-Type", "application/json")
-        .json(&serde_json::json!({ "emoji": emoji }));
-
-    if let Ok(token) = std::env::var("BUZZ_API_TOKEN") {
-        req = req.header("Authorization", format!("Bearer {token}"));
-    } else if let Ok(pubkey) = std::env::var("BUZZ_RELAY_PUBKEY") {
-        req = req.header("X-Pubkey", pubkey);
-    }
-
-    let resp = req
-        .send()
-        .await
-        .map_err(|e| WorkflowError::WebhookError(format!("AddReaction HTTP error: {e}")))?;
-
-    let status = resp.status();
-
-    if !status.is_success() {
-        let body = resp
-            .text()
-            .await
-            .unwrap_or_else(|_| "<unreadable>".to_owned());
-        return Err(WorkflowError::WebhookError(format!(
-            "AddReaction: relay returned {status} for message {message_id}: {body}"
-        )));
-    }
-
-    let body_text = resp.text().await.unwrap_or_else(|_| String::new());
-    let body_json: JsonValue = serde_json::from_str(&body_text)
-        .unwrap_or_else(|_| serde_json::json!({ "raw": body_text }));
-
-    Ok(serde_json::json!({
-        "added": true,
-        "status": status.as_u16(),
-        "response": body_json,
-    }))
-}
-
 /// Rich return type from `execute_run` / `execute_from_step`.
 ///
 /// Carries enough information for the caller to:

desktop/src-tauri/src/managed_agents/runtime.rs

@@ -14,6 +14,9 @@ use crate::{
     util::now_iso,
 };
 
+mod configure_cli;
+pub(crate) use configure_cli::configure_runtime_cli;
+
 mod path;
 pub(in crate::managed_agents) use path::build_augmented_path;
 
@@ -1591,20 +1594,6 @@ pub(crate) fn build_respond_to_env(
     Ok((set, remove))
 }
 
-pub(crate) fn configure_runtime_cli(
-    command: &mut std::process::Command,
-    runtime: Option<&KnownAcpRuntime>,
-) {
-    let Some(runtime) = runtime else {
-        return;
-    };
-    if runtime.id != "claude" {
-        return;
-    }
-    if let Some(cli_path) = runtime.underlying_cli.and_then(resolve_command) {
-        command.env("CLAUDE_CODE_EXECUTABLE", cli_path);
-    }
-}
 
 /// Spawn an agent process without holding any locks on records or runtimes.
 /// Returns the child process and log path on success. The caller is responsible