feat(workflow): support message thread context

crates/buzz-relay/src/workflow_sink.rs

@@ -176,10 +176,12 @@ impl ActionSink for RelayActionSink {
         channel_id: &str,
         text: &str,
         author_pubkey: &str,
+        thread_root: Option<&str>,
     ) -> Pin<Box<dyn Future<Output = Result<String, ActionSinkError>> + Send + '_>> {
         let channel_id = channel_id.to_owned();
         let text = text.to_owned();
         let author_pubkey = author_pubkey.to_owned();
+        let thread_root = thread_root.map(str::to_owned);
 
         Box::pin(async move {
             // 0. Upgrade weak reference — fails only during shutdown.
@@ -235,6 +237,46 @@ impl ActionSink for RelayActionSink {
                 ));
             }
 
+            let thread_root = if let Some(root_hex) = thread_root {
+                let root_id = nostr::EventId::from_hex(&root_hex).map_err(|e| {
+                    ActionSinkError::InvalidInput(format!("invalid thread root event ID: {e}"))
+                })?;
+                let root_bytes = root_id.as_bytes().to_vec();
+                let root_event = state
+                    .db
+                    .get_event_by_id(tenant.community(), &root_bytes)
+                    .await
+                    .map_err(|e| ActionSinkError::Database(e.to_string()))?
+                    .ok_or_else(|| {
+                        ActionSinkError::InvalidInput(format!(
+                            "thread root event {root_hex} was not found"
+                        ))
+                    })?;
+                if root_event.channel_id != Some(channel_uuid) {
+                    return Err(ActionSinkError::InvalidInput(
+                        "thread root belongs to a different channel".into(),
+                    ));
+                }
+                let root_meta = state
+                    .db
+                    .get_thread_metadata_by_event(tenant.community(), &root_bytes)
+                    .await
+                    .map_err(|e| ActionSinkError::Database(e.to_string()))?;
+                if root_meta.is_some_and(|meta| meta.parent_event_id.is_some()) {
+                    return Err(ActionSinkError::InvalidInput(
+                        "thread_root must identify a top-level message".into(),
+                    ));
+                }
+                let root_created = chrono::DateTime::from_timestamp(
+                    root_event.event.created_at.as_secs() as i64,
+                    0,
+                )
+                .unwrap_or_else(Utc::now);
+                Some((root_id.to_hex(), root_bytes, root_created))
+            } else {
+                None
+            };
+
             let author_pubkey = nostr::PublicKey::from_hex(&author_pubkey).map_err(|e| {
                 ActionSinkError::InvalidInput(format!("invalid author pubkey: {e}"))
             })?;
@@ -265,6 +307,12 @@ impl ActionSink for RelayActionSink {
                 Tag::parse(["buzz:workflow", "true"])
                     .map_err(|e| ActionSinkError::EventBuild(format!("workflow tag: {e}")))?,
             ];
+            if let Some((root_hex, _, _)) = &thread_root {
+                tags.push(
+                    Tag::parse(["e", root_hex, "", "reply"])
+                        .map_err(|e| ActionSinkError::EventBuild(format!("thread tag: {e}")))?,
+                );
+            }
 
             // Resolve `@Name` mentions to channel-member pubkeys and append a
             // `p` tag for each (skipping the author, already tagged above). A
@@ -321,16 +369,17 @@ impl ActionSink for RelayActionSink {
             );
 
             // 4. Persist event with thread metadata (matches REST handler path).
-            //    Workflow messages are always top-level: depth=0, no parent/root.
+            let parent_event_id = thread_root.as_ref().map(|(_, bytes, _)| bytes.as_slice());
+            let parent_created_at = thread_root.as_ref().map(|(_, _, created)| *created);
             let thread_meta = Some(buzz_db::event::ThreadMetadataParams {
                 event_id: &event_id_bytes,
                 event_created_at,
                 channel_id: channel_uuid,
-                parent_event_id: None,
-                parent_event_created_at: None,
-                root_event_id: None,
-                root_event_created_at: None,
-                depth: 0,
+                parent_event_id,
+                parent_event_created_at: parent_created_at,
+                root_event_id: parent_event_id,
+                root_event_created_at: parent_created_at,
+                depth: i32::from(parent_event_id.is_some()),
                 broadcast: false,
             });
 
@@ -676,6 +725,7 @@ mod integration_tests {
                 &channel.id.to_string(),
                 "heads up @Robby — please take a look",
                 &author_hex,
+                None,
             )
             .await
             .expect("send_message");

crates/buzz-workflow/src/action_sink.rs

@@ -57,6 +57,7 @@ pub trait ActionSink: Send + Sync {
     /// - `text`: message body (must not be empty/whitespace-only)
     /// - `author_pubkey`: hex-encoded pubkey of the workflow owner (used for
     ///   the `p` attribution tag; the relay keypair signs the event)
+    /// - `thread_root`: optional top-level event ID to post a direct reply under
     ///
     /// Returns the event ID hex string on success.
     fn send_message(
@@ -65,5 +66,6 @@ pub trait ActionSink: Send + Sync {
         channel_id: &str,
         text: &str,
         author_pubkey: &str,
+        thread_root: Option<&str>,
     ) -> Pin<Box<dyn Future<Output = Result<String, ActionSinkError>> + Send + '_>>;
 }

crates/buzz-workflow/src/executor.rs

@@ -37,6 +37,10 @@ pub struct TriggerContext {
     pub emoji: String,
     /// Event ID of the triggering message (hex string).
     pub message_id: String,
+    /// Root event ID for a threaded message, or the message's own ID when top-level.
+    pub root_message_id: String,
+    /// Immediate parent event ID for a reply, empty when the message is top-level.
+    pub parent_message_id: String,
     /// Arbitrary webhook body fields (webhook trigger).
     pub webhook_fields: HashMap<String, String>,
 }
@@ -54,6 +58,8 @@ impl TriggerContext {
             "timestamp" => Some(&self.timestamp),
             "emoji" => Some(&self.emoji),
             "message_id" => Some(&self.message_id),
+            "root_message_id" => Some(&self.root_message_id),
+            "parent_message_id" => Some(&self.parent_message_id),
             other => self.webhook_fields.get(other).map(|s| s.as_str()),
         }
     }
@@ -65,6 +71,7 @@ impl TriggerContext {
 /// - `| truncate(N)` — truncate to N characters
 /// - `| npub` — encode a hex pubkey as its full bech32 `npub` (non-pubkey
 ///   values pass through unchanged); `truncate_pubkey` is a legacy alias
+/// - `| json` — encode a string as a JSON string literal
 ///
 /// Unknown `{{keys}}` are left as literal text (no error, no substitution).
 pub fn resolve_template(
@@ -195,6 +202,11 @@ fn apply_filter(value: String, filter: &str) -> Result<String, WorkflowError> {
         return Ok(value);
     }
 
+    if filter == "json" {
+        return serde_json::to_string(&value)
+            .map_err(|error| WorkflowError::TemplateError(error.to_string()));
+    }
+
     Err(WorkflowError::TemplateError(format!(
         "unknown filter: {filter}"
     )))
@@ -213,6 +225,8 @@ fn apply_filter(value: String, filter: &str) -> Result<String, WorkflowError> {
 /// | `trigger.timestamp`               | `trigger_timestamp`       |
 /// | `trigger.emoji`                   | `trigger_emoji`           |
 /// | `trigger.message_id`              | `trigger_message_id`      |
+/// | `trigger.root_message_id`         | `trigger_root_message_id` |
+/// | `trigger.parent_message_id`       | `trigger_parent_message_id` |
 /// | `steps.STEP_ID.output.FIELD`      | `steps_STEP_ID_output_FIELD` |
 ///
 /// Also registers string helper functions that the `cron` crate's `evalexpr` v11
@@ -293,6 +307,14 @@ pub fn build_eval_context(
         ("trigger_timestamp", trigger_ctx.timestamp.as_str()),
         ("trigger_emoji", trigger_ctx.emoji.as_str()),
         ("trigger_message_id", trigger_ctx.message_id.as_str()),
+        (
+            "trigger_root_message_id",
+            trigger_ctx.root_message_id.as_str(),
+        ),
+        (
+            "trigger_parent_message_id",
+            trigger_ctx.parent_message_id.as_str(),
+        ),
     ];
 
     for (name, val) in &trigger_fields {
@@ -403,9 +425,14 @@ pub fn resolve_step_templates(
     };
 
     match &step.action {
-        SendMessage { text, channel } => Ok(SendMessage {
+        SendMessage {
+            text,
+            channel,
+            thread_root,
+        } => Ok(SendMessage {
             text: t(text)?,
             channel: t_opt(channel)?,
+            thread_root: t_opt(thread_root)?,
         }),
         SendDm { to, text } => Ok(SendDm {
             to: t(to)?,
@@ -527,7 +554,11 @@ pub async fn dispatch_action(
     use ActionDef::*;
 
     match action {
-        SendMessage { text, channel } => {
+        SendMessage {
+            text,
+            channel,
+            thread_root,
+        } => {
             // Look up workflow metadata for destination validation and
             // attribution, scoped to the run's community — the same run/workflow
             // UUID may exist in another community, so a bare-id lookup could
@@ -567,7 +598,13 @@ pub async fn dispatch_action(
 
             let event_id = engine
                 .action_sink()?
-                .send_message(community_id, &channel_id, text, &owner_pubkey_hex)
+                .send_message(
+                    community_id,
+                    &channel_id,
+                    text,
+                    &owner_pubkey_hex,
+                    thread_root.as_deref(),
+                )
                 .await
                 .map_err(WorkflowError::from)?;
 
@@ -1229,6 +1266,8 @@ mod tests {
             timestamp: "1700000000".to_owned(),
             emoji: "fire".to_owned(),
             message_id: "event-id-hex".to_owned(),
+            root_message_id: "root-event-id-hex".to_owned(),
+            parent_message_id: "parent-event-id-hex".to_owned(),
             webhook_fields: HashMap::new(),
         }
     }
@@ -1247,6 +1286,58 @@ mod tests {
         assert_eq!(out, "By abc123def456");
     }
 
+    #[test]
+    fn resolve_trigger_thread_ids() {
+        let ctx = make_trigger();
+        let out = resolve_template(
+            "{{trigger.root_message_id}}/{{trigger.parent_message_id}}",
+            &ctx,
+            &HashMap::new(),
+        )
+        .unwrap();
+        assert_eq!(out, "root-event-id-hex/parent-event-id-hex");
+    }
+
+    #[test]
+    fn resolve_send_message_thread_root_template() {
+        let step = Step {
+            id: "reply".to_owned(),
+            name: None,
+            if_expr: None,
+            timeout_secs: None,
+            action: ActionDef::SendMessage {
+                text: "{{trigger.text}}".to_owned(),
+                channel: None,
+                thread_root: Some("{{trigger.root_message_id}}".to_owned()),
+            },
+        };
+
+        let resolved = resolve_step_templates(&step, &make_trigger(), &HashMap::new()).unwrap();
+
+        match resolved {
+            ActionDef::SendMessage {
+                text, thread_root, ..
+            } => {
+                assert_eq!(text, "P1 incident in production");
+                assert_eq!(thread_root.as_deref(), Some("root-event-id-hex"));
+            }
+            other => panic!("unexpected action: {other:?}"),
+        }
+    }
+
+    #[test]
+    fn json_filter_escapes_human_text() {
+        let mut ctx = make_trigger();
+        ctx.text = "He said \"hello\"\nthen left.".to_owned();
+        let out = resolve_template(
+            "{\"message\":{{trigger.text | json}}}",
+            &ctx,
+            &HashMap::new(),
+        )
+        .unwrap();
+        assert_eq!(out, "{\"message\":\"He said \\\"hello\\\"\\nthen left.\"}");
+    }
+
     #[test]
     fn resolve_step_output() {
         let ctx = make_trigger();