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;
@@ -473,6 +473,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

@@ -74,4 +74,22 @@ pub trait ActionSink: Send + Sync {
         author_pubkey: &str,
         reply_to: Option<&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

@@ -671,26 +671,44 @@ pub async fn dispatch_action(
                             "AddReaction: no trigger.message_id available".into(),
                         ))
                     } else {
-                        #[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);
+
+                        let event_id = engine
+                            .action_sink()?
+                            .add_reaction(
+                                community_id,
+                                &trigger_ctx.message_id,
+                                emoji,
+                                &owner_pubkey_hex,
+                            )
+                            .await
+                            .map_err(WorkflowError::from)?;
 
-                        #[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 }),
-                            ))
-                        }
+                        Ok(StepResult::Completed(serde_json::json!({
+                            "added": true,
+                            "event_id": event_id,
+                        })))
                     }
                 }
-
                 CallWebhook {
                     url,
                     method,
@@ -976,54 +994,6 @@ fn shared_http_client() -> &'static reqwest::Client {
     &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`.
 ///

desktop/src/shared/api/relayClientSession.ts

@@ -40,6 +40,11 @@ import { getChannelReconnectRepairEvents } from "@/shared/api/channelReconnectRe
 import { replayLiveSubscriptions } from "@/shared/api/relayReconnectReplay";
 import { publishSessionEvent } from "@/shared/api/relayEventPublisher";
 import { activateRateLimitIfSignalled } from "@/shared/api/relayRateLimitGate";
+import {
+  assertWithinFrameLimit,
+  isOversizedFrameError,
+} from "@/shared/api/relayFrameLimit";
+import { noteFrameTooLargeLimit } from "@/shared/api/relayClientTransport";
 import {
   fetchChunkedHistory,
   requestFirstEventGated,
@@ -97,6 +102,8 @@ export class RelayClient {
   private stabilityTimer: number | null = null;
   private visibleChannelId: string | null = null;
   private authOkTracker = new AuthOkTracker();
+  /** Server-advertised maximum frame size in bytes; null until first NOTICE. */
+  private maxFrameBytes: number | null = null;
   private terminal = false;
 
   private connectionStateEmitter = new RelayConnectionStateEmitter("idle");
@@ -132,6 +139,7 @@ export class RelayClient {
     this.terminal = false;
     this.visibleChannelId = null;
     this.authOkTracker.reset();
+    this.maxFrameBytes = null;
     this.connectionStateEmitter.set("idle");
 
     if (this.wsId !== null) {
@@ -428,6 +436,7 @@ export class RelayClient {
     // terminal latch and AUTH rejection streak, and bypasses backoff once.
     this.terminal = false;
     this.authOkTracker.reset();
+    this.maxFrameBytes = null;
     this.keepAliveRequested = true;
     await this.connectBypassingBackoff();
   }
@@ -652,6 +661,9 @@ export class RelayClient {
     if (this.wsId === null) {
       throw new Error("Relay socket is not connected.");
     }
+    if (this.maxFrameBytes !== null) {
+      assertWithinFrameLimit(payload, this.maxFrameBytes);
+    }
 
     await invoke("plugin:websocket|send", {
       id: this.wsId,
@@ -666,6 +678,9 @@ export class RelayClient {
     if (generation !== this.connectionGeneration || this.wsId === null) {
       throw new Error("Relay publish was superseded by a session change.");
     }
+    if (this.maxFrameBytes !== null) {
+      assertWithinFrameLimit(payload, this.maxFrameBytes);
+    }
     const wsId = this.wsId;
     await invoke("plugin:websocket|send", {
       id: wsId,
@@ -693,6 +708,9 @@ export class RelayClient {
     try {
       await this.sendRaw(payload);
     } catch (error) {
+      if (isOversizedFrameError(error)) {
+        throw error;
+      }
       const normalizedError = this.recoverFromSocketFailure(
         error,
         fallbackMessage,
@@ -815,8 +833,14 @@ export class RelayClient {
     }
 
     if (type === "NOTICE" && typeof rest[0] === "string") {
+      const notice = rest[0];
       // Connection-scoped back-pressure — arm the gate until it expires.
-      activateRateLimitIfSignalled(rest[0]);
+      activateRateLimitIfSignalled(notice);
+      const newMax = noteFrameTooLargeLimit(notice, this.maxFrameBytes);
+      if (newMax !== null) {
+        this.maxFrameBytes = newMax;
+        console.warn(`[relay] Frame size limit updated to ${newMax} bytes`);
+      }
     }
   }