diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index 5ba9650e9..8c37e977d 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -872,6 +872,128 @@ pub(crate) async fn resolve_nip10_thread_meta( })) } +/// Resolved thread ancestry for a relay-built reply (workflow path). +/// +/// Carries the parent and root identifiers plus the reply's depth, so the +/// caller can both emit matching NIP-10 `root`/`reply` tags and persist thread +/// metadata for the signed reply event. +pub(crate) struct ReplyAncestry { + pub parent_event_id: Vec, + pub parent_event_created_at: chrono::DateTime, + pub root_event_id: Vec, + pub root_event_created_at: chrono::DateTime, + pub depth: i32, +} + +impl ReplyAncestry { + /// Root event ID as lowercase hex, for the NIP-10 `root` tag. + pub fn root_hex(&self) -> String { + hex::encode(&self.root_event_id) + } + + /// Parent event ID as lowercase hex, for the NIP-10 `reply` tag. + pub fn parent_hex(&self) -> String { + hex::encode(&self.parent_event_id) + } + + /// Build the DB thread-metadata params for the signed reply event. + pub fn into_thread_meta( + self, + reply_event_id: Vec, + reply_created_at: chrono::DateTime, + channel_id: Uuid, + ) -> ThreadMetadataOwned { + ThreadMetadataOwned { + event_id: reply_event_id, + event_created_at: reply_created_at, + channel_id, + parent_event_id: self.parent_event_id, + parent_event_created_at: self.parent_event_created_at, + root_event_id: self.root_event_id, + root_event_created_at: self.root_event_created_at, + depth: self.depth, + broadcast: false, + } + } +} + +/// Resolve thread ancestry for a reply built by the relay (workflow path). +/// +/// Unlike [`resolve_nip10_thread_meta`], which validates client-supplied NIP-10 +/// `e` tags, this derives ancestry from a known `parent_hex` (the triggering +/// event) and *computes* the correct root and depth. Enforces the same-channel +/// invariant and the depth limit that the ingest path applies. +pub(crate) async fn resolve_relay_reply_thread_meta( + community_id: CommunityId, + parent_hex: &str, + channel_id: Uuid, + state: &AppState, +) -> Result { + let parent_bytes = + hex::decode(parent_hex).map_err(|_| "invalid parent event ID hex".to_string())?; + + let (parent_event_result, parent_meta_result) = tokio::join!( + state.db.get_event_by_id(community_id, &parent_bytes), + state + .db + .get_thread_metadata_by_event(community_id, &parent_bytes), + ); + + let parent_event = parent_event_result + .map_err(|e| format!("db error looking up parent: {e}"))? + .ok_or_else(|| "reply parent not found".to_string())?; + + match parent_event.channel_id { + Some(parent_ch) if parent_ch != channel_id => { + return Err("parent event belongs to a different channel".to_string()); + } + None => return Err("parent event has no channel association".to_string()), + _ => {} + } + + let parent_created = + chrono::DateTime::from_timestamp(parent_event.event.created_at.as_secs() as i64, 0) + .unwrap_or_else(Utc::now); + + let parent_meta = + parent_meta_result.map_err(|e| format!("db error looking up thread metadata: {e}"))?; + + // Root = parent's root if the parent is itself a reply, else the parent. + // Depth = parent depth + 1 (a direct reply to a top-level message is depth 1). + let (root_bytes, root_created, depth) = match parent_meta { + Some(meta) => { + let effective_root = meta.root_event_id.unwrap_or_else(|| parent_bytes.clone()); + let root_ts = if effective_root == parent_bytes { + parent_created + } else if let Ok(Some(root_ev)) = state + .db + .get_event_by_id(community_id, &effective_root) + .await + { + chrono::DateTime::from_timestamp(root_ev.event.created_at.as_secs() as i64, 0) + .unwrap_or(parent_created) + } else { + parent_created + }; + (effective_root, root_ts, meta.depth + 1) + } + // No metadata row ⇒ parent is a top-level message; it is its own root. + None => (parent_bytes.clone(), parent_created, 1), + }; + + if depth > 100 { + return Err("thread depth limit exceeded".to_string()); + } + + Ok(ReplyAncestry { + parent_event_id: parent_bytes, + parent_event_created_at: parent_created, + root_event_id: root_bytes, + root_event_created_at: root_created, + depth, + }) +} + /// Count all `e` tags regardless of content validity. fn count_e_tags(event: &Event) -> usize { event diff --git a/crates/buzz-relay/src/workflow_sink.rs b/crates/buzz-relay/src/workflow_sink.rs index e056ff736..d07ca7d1d 100644 --- a/crates/buzz-relay/src/workflow_sink.rs +++ b/crates/buzz-relay/src/workflow_sink.rs @@ -176,10 +176,12 @@ impl ActionSink for RelayActionSink { channel_id: &str, text: &str, author_pubkey: &str, + reply_to: Option<&str>, ) -> Pin> + Send + '_>> { let channel_id = channel_id.to_owned(); let text = text.to_owned(); let author_pubkey = author_pubkey.to_owned(); + let reply_to = reply_to.map(str::to_owned); Box::pin(async move { // 0. Upgrade weak reference — fails only during shutdown. @@ -271,6 +273,39 @@ impl ActionSink for RelayActionSink { .map_err(|e| ActionSinkError::EventBuild(format!("workflow owner tag: {e}")))?, ]; + // Resolve thread ancestry when this is a threaded reply, so the + // built event carries NIP-10 `root`/`reply` e-tags and persists real + // thread metadata (matching the ingest path) instead of top-level. + let reply_ancestry = match reply_to.as_deref() { + Some(parent_hex) => Some( + crate::handlers::ingest::resolve_relay_reply_thread_meta( + tenant.community(), + parent_hex, + channel_uuid, + &state, + ) + .await + .map_err(ActionSinkError::InvalidInput)?, + ), + None => None, + }; + + // NIP-10 e-tags for the thread. Marked `root`/`reply` so clients and + // the ingest resolver read the ancestry the same way. When the reply + // is directly under the root, both markers point at the same event. + if let Some(ancestry) = &reply_ancestry { + let root_hex = ancestry.root_hex(); + let parent_hex = ancestry.parent_hex(); + tags.push( + Tag::parse(["e", &root_hex, "", "root"]) + .map_err(|e| ActionSinkError::EventBuild(format!("root e tag: {e}")))?, + ); + tags.push( + Tag::parse(["e", &parent_hex, "", "reply"]) + .map_err(|e| ActionSinkError::EventBuild(format!("reply e tag: {e}")))?, + ); + } + // Resolve `@Name` mentions to channel-member pubkeys and append a // `p` tag for each (skipping the author, already tagged above). A // resolution failure must not drop the message, so log and proceed @@ -326,17 +361,24 @@ 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 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, - broadcast: false, + // Threaded replies persist the resolved parent/root/depth; a + // non-reply workflow message stays top-level (depth=0, no parent). + let thread_meta_owned = reply_ancestry.map(|ancestry| { + ancestry.into_thread_meta(event_id_bytes.clone(), event_created_at, channel_uuid) + }); + let thread_meta = Some(match &thread_meta_owned { + Some(owned) => owned.as_params(), + None => 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, + broadcast: false, + }, }); let (stored_event, was_inserted) = state @@ -681,6 +723,7 @@ mod integration_tests { &channel.id.to_string(), "heads up @Robby — please take a look", &author_hex, + None, ) .await .expect("send_message"); @@ -726,4 +769,153 @@ mod integration_tests { so consumers never infer ownership from p-tag order" ); } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn workflow_reply_in_thread_threads_onto_parent() { + let state = test_state().await; + + let author = nostr::Keys::generate(); + let author_hex = author.public_key().to_hex(); + + let host = format!("wf-thread-{}.example", uuid::Uuid::new_v4().simple()); + let community = match state + .db + .create_community_with_owner(&host, &author_hex) + .await + .expect("create community") + { + CreateCommunityWithOwnerResult::Created(rec) => rec.id, + other => panic!("expected fresh community, got {other:?}"), + }; + + let channel = state + .db + .create_channel( + community, + "wf-thread", + ChannelType::Stream, + ChannelVisibility::Open, + None, + &author.public_key().to_bytes(), + None, + ) + .await + .expect("create channel"); + + let sink = RelayActionSink::new(&state); + + // 1. A top-level workflow message becomes the thread root. + let root_hex = sink + .send_message( + community, + &channel.id.to_string(), + "root message", + &author_hex, + None, + ) + .await + .expect("send root"); + + // 2. A reply_in_thread message threads onto it. + let reply_hex = sink + .send_message( + community, + &channel.id.to_string(), + "threaded reply", + &author_hex, + Some(&root_hex), + ) + .await + .expect("send reply"); + + // The reply event carries NIP-10 root+reply e-tags pointing at the root. + let reply_id_bytes = nostr::EventId::from_hex(&reply_hex) + .expect("reply id") + .as_bytes() + .to_vec(); + let stored = state + .db + .get_event_by_id(community, &reply_id_bytes) + .await + .expect("query reply") + .expect("reply persisted"); + let marker = |m: &str| -> Option { + stored.event.tags.iter().find_map(|t| { + let p = t.as_slice(); + if p.len() >= 4 && p[0] == "e" && p[3] == m { + Some(p[1].clone()) + } else { + None + } + }) + }; + assert_eq!(marker("root").as_deref(), Some(root_hex.as_str())); + assert_eq!(marker("reply").as_deref(), Some(root_hex.as_str())); + + // Thread metadata reflects a depth-1 reply parented on the root. + let meta = state + .db + .get_thread_metadata_by_event(community, &reply_id_bytes) + .await + .expect("query meta") + .expect("reply has thread metadata"); + assert_eq!( + meta.depth, 1, + "direct reply to a top-level message is depth 1" + ); + let root_bytes = nostr::EventId::from_hex(&root_hex) + .expect("root id") + .as_bytes() + .to_vec(); + assert_eq!(meta.parent_event_id.as_deref(), Some(root_bytes.as_slice())); + assert_eq!(meta.root_event_id.as_deref(), Some(root_bytes.as_slice())); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn workflow_reply_to_missing_parent_errors() { + let state = test_state().await; + let author = nostr::Keys::generate(); + let author_hex = author.public_key().to_hex(); + let host = format!("wf-missing-{}.example", uuid::Uuid::new_v4().simple()); + let community = match state + .db + .create_community_with_owner(&host, &author_hex) + .await + .expect("create community") + { + CreateCommunityWithOwnerResult::Created(rec) => rec.id, + other => panic!("expected fresh community, got {other:?}"), + }; + let channel = state + .db + .create_channel( + community, + "wf-missing", + ChannelType::Stream, + ChannelVisibility::Open, + None, + &author.public_key().to_bytes(), + None, + ) + .await + .expect("create channel"); + + let unknown = nostr::Keys::generate().public_key().to_hex(); + let err = RelayActionSink::new(&state) + .send_message( + community, + &channel.id.to_string(), + "orphan reply", + &author_hex, + Some(&unknown), + ) + .await + .expect_err("reply to a non-existent parent must fail"); + assert!( + matches!(err, ActionSinkError::InvalidInput(_)), + "expected InvalidInput, got {err:?}" + ); + } } diff --git a/crates/buzz-workflow/src/action_sink.rs b/crates/buzz-workflow/src/action_sink.rs index 0c6002e74..079c27a91 100644 --- a/crates/buzz-workflow/src/action_sink.rs +++ b/crates/buzz-workflow/src/action_sink.rs @@ -57,6 +57,9 @@ 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) + /// - `reply_to`: when `Some(event_id_hex)`, the message is posted as a + /// threaded reply to that event (NIP-10 root/reply tags + real thread + /// metadata); when `None`, it is a top-level channel message. /// /// Returns the event ID hex string on success. fn send_message( @@ -65,5 +68,6 @@ pub trait ActionSink: Send + Sync { channel_id: &str, text: &str, author_pubkey: &str, + reply_to: Option<&str>, ) -> Pin> + Send + '_>>; } diff --git a/crates/buzz-workflow/src/executor.rs b/crates/buzz-workflow/src/executor.rs index dffa49271..5c712dcff 100644 --- a/crates/buzz-workflow/src/executor.rs +++ b/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, + /// True when the triggering event is itself a threaded reply (carries a + /// NIP-10 `reply`/`root` marker e-tag). Lets a `message_posted` filter + /// select only top-level messages via `trigger_is_reply == false`. + pub is_reply: bool, /// Arbitrary webhook body fields (webhook trigger). pub webhook_fields: HashMap, } @@ -213,6 +217,7 @@ fn apply_filter(value: String, filter: &str) -> Result { /// | `trigger.timestamp` | `trigger_timestamp` | /// | `trigger.emoji` | `trigger_emoji` | /// | `trigger.message_id` | `trigger_message_id` | +/// | `trigger.is_reply` | `trigger_is_reply` (bool) | /// | `steps.STEP_ID.output.FIELD` | `steps_STEP_ID_output_FIELD` | /// /// Also registers string helper functions that the `cron` crate's `evalexpr` v11 @@ -300,6 +305,14 @@ pub fn build_eval_context( .map_err(|e| WorkflowError::ConditionError(e.to_string()))?; } + // `trigger_is_reply` is boolean (not a string field), so a filter can read + // `trigger_is_reply == false` to fire only on top-level messages. + ctx.set_value( + "trigger_is_reply".into(), + Value::Boolean(trigger_ctx.is_reply), + ) + .map_err(|e| WorkflowError::ConditionError(e.to_string()))?; + for (step_id, output) in step_outputs { if let JsonValue::Object(map) = output { for (field, val) in map { @@ -403,9 +416,14 @@ pub fn resolve_step_templates( }; match &step.action { - SendMessage { text, channel } => Ok(SendMessage { + SendMessage { + text, + channel, + reply_in_thread, + } => Ok(SendMessage { text: t(text)?, channel: t_opt(channel)?, + reply_in_thread: *reply_in_thread, }), SendDm { to, text } => Ok(SendDm { to: t(to)?, @@ -546,7 +564,11 @@ pub async fn dispatch_action( let result = serving_write .protect(async { match action { - SendMessage { text, channel } => { + SendMessage { + text, + channel, + reply_in_thread, + } => { // 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 @@ -577,16 +599,38 @@ pub async fn dispatch_action( )?; let owner_pubkey_hex = hex::encode(&workflow.owner_pubkey); + // Thread the reply onto the triggering message when requested. + // The trigger must carry the event to reply to; schema + // validation already forbids `reply_in_thread` on triggers + // that have no message, so an empty id here is a real fault. + let reply_to = if *reply_in_thread { + if trigger_ctx.message_id.is_empty() { + return Err(WorkflowError::InvalidDefinition( + "SendMessage: reply_in_thread is set but the trigger has no message_id to reply to".into(), + )); + } + Some(trigger_ctx.message_id.as_str()) + } else { + None + }; + info!( run_id = %run_id, step = step_id, channel = %channel_id, + reply_in_thread = *reply_in_thread, "SendMessage → {channel_id}: {text}" ); 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, + reply_to, + ) .await .map_err(WorkflowError::from)?; @@ -1266,6 +1310,7 @@ mod tests { timestamp: "1700000000".to_owned(), emoji: "fire".to_owned(), message_id: "event-id-hex".to_owned(), + is_reply: false, webhook_fields: HashMap::new(), } } @@ -1385,6 +1430,56 @@ mod tests { assert!(!result); } + #[tokio::test] + async fn condition_trigger_is_reply_selects_top_level_only() { + // The top-level-only filter from the feature's use case. + let mut ctx = make_trigger(); + + ctx.is_reply = false; + assert!( + evaluate_condition("trigger_is_reply == false", &ctx, &HashMap::new()) + .await + .unwrap(), + "top-level message should pass the filter" + ); + + ctx.is_reply = true; + assert!( + !evaluate_condition("trigger_is_reply == false", &ctx, &HashMap::new()) + .await + .unwrap(), + "threaded reply should be filtered out" + ); + } + + #[test] + fn resolve_step_templates_carries_reply_in_thread() { + let ctx = make_trigger(); + let step = Step { + id: "reply".to_owned(), + name: None, + if_expr: None, + timeout_secs: None, + action: ActionDef::SendMessage { + text: "hi {{trigger.author}}".to_owned(), + channel: None, + reply_in_thread: true, + }, + }; + let resolved = resolve_step_templates(&step, &ctx, &HashMap::new()).unwrap(); + match resolved { + ActionDef::SendMessage { + text, + reply_in_thread, + .. + } => { + assert_eq!(text, "hi abc123def456"); + assert!(reply_in_thread, "reply_in_thread must survive resolution"); + } + other => panic!("unexpected action: {other:?}"), + } + } + #[tokio::test] async fn condition_or_expression() { let ctx = make_trigger(); // text contains "P1" diff --git a/crates/buzz-workflow/src/lib.rs b/crates/buzz-workflow/src/lib.rs index fe8b477ba..bc068191b 100644 --- a/crates/buzz-workflow/src/lib.rs +++ b/crates/buzz-workflow/src/lib.rs @@ -1016,10 +1016,21 @@ pub fn build_trigger_context(event: &buzz_core::StoredEvent) -> executor::Trigge timestamp: event.event.created_at.as_secs().to_string(), emoji, message_id, + is_reply: event_is_reply(&event.event), webhook_fields: HashMap::new(), } } +/// True when an event is a threaded reply — it carries a NIP-10 `e` tag marked +/// `reply` or `root`. A top-level message has neither, so a `message_posted` +/// filter of `trigger_is_reply == false` fires only on new top-level messages. +fn event_is_reply(event: &nostr::Event) -> bool { + event.tags.iter().any(|tag| { + let parts = tag.as_slice(); + parts.len() >= 4 && parts[0] == "e" && (parts[3] == "reply" || parts[3] == "root") + }) +} + /// Pure authority decision for [`WorkflowEngine::check_owner_authority`]. /// /// `role` is the owner's *current* active role in the workflow's channel @@ -1564,6 +1575,53 @@ steps: // Non-reaction events have empty emoji. assert_eq!(ctx.emoji, ""); assert!(ctx.webhook_fields.is_empty()); + // A top-level message (no e-tags) is not a reply. + assert!(!ctx.is_reply); + } + + #[test] + fn build_trigger_context_is_reply_true_for_threaded_message() { + use nostr::{EventBuilder, Keys, Kind, Tag}; + use uuid::Uuid; + let root = Keys::generate(); + let root_event = EventBuilder::new(Kind::Custom(9), "root") + .tags([]) + .sign_with_keys(&root) + .expect("sign root"); + let root_hex = root_event.id.to_hex(); + + let keys = Keys::generate(); + let event = EventBuilder::new(Kind::Custom(9), "a threaded reply") + .tags([ + Tag::parse(["e", &root_hex, "", "root"]).expect("root tag"), + Tag::parse(["e", &root_hex, "", "reply"]).expect("reply tag"), + ]) + .sign_with_keys(&keys) + .expect("sign"); + let stored = buzz_core::StoredEvent::new(event, Some(Uuid::new_v4())); + let ctx = build_trigger_context(&stored); + assert!(ctx.is_reply, "message with reply/root e-tags is a reply"); + } + + #[test] + fn build_trigger_context_is_reply_false_for_unmarked_e_tag() { + // A bare `e` tag with no NIP-10 marker (e.g. a plain mention/quote) is + // not treated as a thread reply — only `reply`/`root` markers count. + use nostr::{EventBuilder, Keys, Kind, Tag}; + use uuid::Uuid; + let other = Keys::generate(); + let other_event = EventBuilder::new(Kind::Custom(9), "other") + .tags([]) + .sign_with_keys(&other) + .expect("sign"); + let keys = Keys::generate(); + let event = EventBuilder::new(Kind::Custom(9), "quotes another") + .tags([Tag::parse(["e", &other_event.id.to_hex()]).expect("bare e tag")]) + .sign_with_keys(&keys) + .expect("sign"); + let stored = buzz_core::StoredEvent::new(event, Some(Uuid::new_v4())); + let ctx = build_trigger_context(&stored); + assert!(!ctx.is_reply, "unmarked e-tag must not count as a reply"); } #[test] diff --git a/crates/buzz-workflow/src/schema.rs b/crates/buzz-workflow/src/schema.rs index 9bc79aa48..8b87498ec 100644 --- a/crates/buzz-workflow/src/schema.rs +++ b/crates/buzz-workflow/src/schema.rs @@ -97,6 +97,11 @@ pub enum ActionDef { /// Optional channel UUID override. Must be a valid UUID string. #[serde(default)] channel: Option, + /// Reply to the triggering message in its thread instead of posting a + /// new top-level message. Only valid for message-based triggers, which + /// carry a triggering event to reply to. + #[serde(default)] + reply_in_thread: bool, }, /// Send a direct message to a user. SendDm { @@ -205,6 +210,34 @@ impl WorkflowDef { } } + // `reply_in_thread` requires a triggering message to reply to. Schedule + // and webhook triggers have none, so reject the combination at + // definition time rather than failing silently at run time. + let trigger_has_message = matches!( + self.trigger, + TriggerDef::MessagePosted { .. } + | TriggerDef::ReactionAdded { .. } + | TriggerDef::DiffPosted { .. } + ); + if !trigger_has_message { + for step in &self.steps { + if matches!( + step.action, + ActionDef::SendMessage { + reply_in_thread: true, + .. + } + ) { + return Err(WorkflowError::InvalidDefinition(format!( + "step '{}': reply_in_thread requires a message-based trigger \ + (message_posted, reaction_added, or diff_posted); \ + schedule and webhook triggers have no message to reply to", + step.id + ))); + } + } + } + if let TriggerDef::Schedule { cron, interval } = &self.trigger { if cron.is_none() && interval.is_none() { return Err(WorkflowError::InvalidDefinition( @@ -454,6 +487,78 @@ mod tests { assert!(matches!(err, WorkflowError::InvalidDefinition(_))); } + #[test] + fn reply_in_thread_defaults_false_and_round_trips() { + // Absent field defaults to false. + let yaml = "name: Auto Reply\ntrigger:\n on: message_posted\nsteps:\n - id: s1\n action: send_message\n text: hi\n"; + let (def, _) = parse_yaml(yaml).expect("parse failed"); + match &def.steps[0].action { + ActionDef::SendMessage { + reply_in_thread, .. + } => assert!(!reply_in_thread, "should default to false"), + other => panic!("unexpected action: {other:?}"), + } + + // Explicit true parses, and survives a JSON round-trip. + let yaml = "name: Auto Reply\ntrigger:\n on: message_posted\nsteps:\n - id: s1\n action: send_message\n text: hi\n reply_in_thread: true\n"; + let (def, _) = parse_yaml(yaml).expect("parse failed"); + match &def.steps[0].action { + ActionDef::SendMessage { + reply_in_thread, .. + } => assert!(reply_in_thread), + other => panic!("unexpected action: {other:?}"), + } + let json = serde_json::to_string(&def).expect("serialize"); + let reparsed: WorkflowDef = serde_json::from_str(&json).expect("json round-trip"); + assert!(matches!( + &reparsed.steps[0].action, + ActionDef::SendMessage { + reply_in_thread: true, + .. + } + )); + } + + #[test] + fn validate_accepts_reply_in_thread_on_message_triggers() { + for on in ["message_posted", "reaction_added", "diff_posted"] { + let yaml = format!( + "name: Auto Reply\ntrigger:\n on: {on}\nsteps:\n - id: s1\n action: send_message\n text: hi\n reply_in_thread: true\n" + ); + parse_yaml(&yaml) + .unwrap_or_else(|e| panic!("reply_in_thread should be valid on {on}: {e}")); + } + } + + #[test] + fn validate_rejects_reply_in_thread_on_schedule_trigger() { + let yaml = "name: Bad\ntrigger:\n on: schedule\n cron: '0 9 * * 1-5'\nsteps:\n - id: s1\n action: send_message\n text: hi\n reply_in_thread: true\n"; + let err = parse_yaml(yaml).unwrap_err(); + match &err { + WorkflowError::InvalidDefinition(msg) => { + assert!( + msg.contains("reply_in_thread"), + "expected reply_in_thread in: {msg}" + ); + } + other => panic!("expected InvalidDefinition, got: {other}"), + } + } + + #[test] + fn validate_rejects_reply_in_thread_on_webhook_trigger() { + let yaml = "name: Bad\ntrigger:\n on: webhook\nsteps:\n - id: s1\n action: send_message\n text: hi\n channel: 00000000-0000-0000-0000-000000000000\n reply_in_thread: true\n"; + let err = parse_yaml(yaml).unwrap_err(); + assert!(matches!(err, WorkflowError::InvalidDefinition(_))); + } + + #[test] + fn validate_allows_reply_in_thread_false_on_schedule() { + // Explicit `false` on a schedule trigger is fine — no message needed. + let yaml = "name: OK\ntrigger:\n on: schedule\n cron: '0 9 * * 1-5'\nsteps:\n - id: s1\n action: send_message\n text: hi\n reply_in_thread: false\n"; + parse_yaml(yaml).expect("reply_in_thread: false on schedule should be valid"); + } + #[test] fn enabled_defaults_to_true() { let yaml = "name: Test\ntrigger:\n on: webhook\nsteps:\n - id: s1\n action: delay\n duration: 1m\n"; diff --git a/desktop/src/features/workflows/ui/WorkflowStepCard.tsx b/desktop/src/features/workflows/ui/WorkflowStepCard.tsx index ffe6b3546..5c09bd5d7 100644 --- a/desktop/src/features/workflows/ui/WorkflowStepCard.tsx +++ b/desktop/src/features/workflows/ui/WorkflowStepCard.tsx @@ -1,6 +1,7 @@ import { Trash2 } from "lucide-react"; import { Button } from "@/shared/ui/button"; +import { Checkbox } from "@/shared/ui/checkbox"; import { Input } from "@/shared/ui/input"; import { Textarea } from "@/shared/ui/textarea"; import { FieldLabel, FormSelect } from "./workflowFormPrimitives"; @@ -112,6 +113,21 @@ function StepConfigFields({

) : null} + {triggerType !== "webhook" && triggerType !== "schedule" ? ( +
+ + onUpdate({ ...step, replyInThread: checked === true }) + } + /> + +
+ ) : null} ); case "send_dm": diff --git a/desktop/src/features/workflows/ui/workflowFormTypes.test.mjs b/desktop/src/features/workflows/ui/workflowFormTypes.test.mjs new file mode 100644 index 000000000..816a92a43 --- /dev/null +++ b/desktop/src/features/workflows/ui/workflowFormTypes.test.mjs @@ -0,0 +1,63 @@ +import assert from "node:assert/strict"; +import { test } from "node:test"; + +import { + formStateToYaml, + yamlToFormState, + DEFAULT_FORM_STATE, +} from "./workflowFormTypes.ts"; + +function sendMessageState(overrides) { + return { + ...DEFAULT_FORM_STATE, + name: "Auto Reply", + trigger: { on: "message_posted", filter: "trigger_is_reply == false" }, + steps: [ + { + id: "step_1", + action: "send_message", + text: "pre-written reply", + ...overrides, + }, + ], + }; +} + +test("reply_in_thread is emitted only when the checkbox is on", () => { + const withReply = formStateToYaml(sendMessageState({ replyInThread: true })); + assert.match(withReply, /reply_in_thread: true/); + + const withoutReply = formStateToYaml( + sendMessageState({ replyInThread: false }), + ); + assert.doesNotMatch(withoutReply, /reply_in_thread/); + + const unset = formStateToYaml(sendMessageState({})); + assert.doesNotMatch(unset, /reply_in_thread/); +}); + +test("reply_in_thread round-trips YAML -> form -> YAML", () => { + const yaml = formStateToYaml(sendMessageState({ replyInThread: true })); + const parsed = yamlToFormState(yaml); + assert.equal(parsed.ok, true); + assert.equal(parsed.state.steps[0].replyInThread, true); + + const reserialized = formStateToYaml(parsed.state); + assert.match(reserialized, /reply_in_thread: true/); +}); + +test("absent reply_in_thread parses as false", () => { + const yaml = [ + "name: No Reply", + "trigger:", + " on: message_posted", + "steps:", + " - id: step_1", + " action: send_message", + " text: hi", + "", + ].join("\n"); + const parsed = yamlToFormState(yaml); + assert.equal(parsed.ok, true); + assert.equal(parsed.state.steps[0].replyInThread, false); +}); diff --git a/desktop/src/features/workflows/ui/workflowFormTypes.ts b/desktop/src/features/workflows/ui/workflowFormTypes.ts index 26669b79f..ddd7f536f 100644 --- a/desktop/src/features/workflows/ui/workflowFormTypes.ts +++ b/desktop/src/features/workflows/ui/workflowFormTypes.ts @@ -43,6 +43,7 @@ export type StepFormState = { duration?: string; text?: string; channel?: string; + replyInThread?: boolean; to?: string; url?: string; method?: string; @@ -141,6 +142,7 @@ function actionFieldsForStep(step: StepFormState): Record { case "send_message": if (step.text) fields.text = step.text; if (step.channel) fields.channel = step.channel; + if (step.replyInThread) fields.reply_in_thread = true; break; case "send_dm": if (step.to) fields.to = step.to; @@ -274,6 +276,7 @@ export function yamlToFormState( duration: step.duration as string | undefined, text: step.text as string | undefined, channel: step.channel as string | undefined, + replyInThread: step.reply_in_thread === true, to: step.to as string | undefined, url: step.url as string | undefined, method: step.method as string | undefined,