diff --git a/connectors/connector-bluesky/src/actions/reply.rs b/connectors/connector-bluesky/src/actions/reply.rs index a5c9b12d..f136be9d 100644 --- a/connectors/connector-bluesky/src/actions/reply.rs +++ b/connectors/connector-bluesky/src/actions/reply.rs @@ -17,8 +17,8 @@ pub fn declaration() -> ActionDecl { "text": { "type": "string", "description": "Reply text." }, "parent_uri": { "type": "string", "description": "AT URI of the parent post." }, "parent_cid": { "type": "string", "description": "CID of the parent post." }, - "root_uri": { "type": "string", "description": "AT URI of the root post in the thread." }, - "root_cid": { "type": "string", "description": "CID of the root post." } + "root_uri": { "type": "string", "description": "AT URI of the root post in the thread — pass the `mention` trigger's `root_uri`, not the mentioned post's `uri`." }, + "root_cid": { "type": "string", "description": "CID of the root post — pass the `mention` trigger's `root_cid`." } }, "required": ["text", "parent_uri", "parent_cid", "root_uri", "root_cid"] })), @@ -87,7 +87,103 @@ pub async fn execute( mod tests { use super::*; - use crate::client::test_helpers::MockBlueskyClient; + use crate::client::test_helpers::{MockBlueskyClient, RecordingBlueskyClient}; + use crate::gateway::route_jetstream_event; + + const OWN: &str = "did:plc:me"; + + /// A real Jetstream `app.bsky.feed.post` create commit that mentions + /// us. `reply` is the record's own reply ref (`None` for a top-level + /// post), which is what decides the thread root. + fn mention_commit(reply: Option) -> serde_json::Value { + let mut record = serde_json::json!({ + "$type": "app.bsky.feed.post", + "text": "hey @me", + "facets": [{ + "features": [{ "$type": "app.bsky.richtext.facet#mention", "did": OWN }], + "index": { "byteStart": 4, "byteEnd": 7 } + }] + }); + if let Some(r) = reply { + record["reply"] = r; + } + serde_json::json!({ + "did": "did:plc:someone", + "time_us": 1_700_000_000_000_000u64, + "kind": "commit", + "commit": { + "operation": "create", + "collection": "app.bsky.feed.post", + "rkey": "3kxyz", + "cid": "bafymention", + "record": record + } + }) + } + + /// The wiring the `bluesky-mention-auto-ack` builtin recipe performs: + /// parent from the mentioned post, root from the trigger's root. + fn reply_input_from_mention(payload: &serde_json::Value) -> serde_json::Value { + serde_json::json!({ + "text": "ack", + "parent_uri": payload["uri"], + "parent_cid": payload["cid"], + "root_uri": payload["root_uri"], + "root_cid": payload["root_cid"], + }) + } + + #[tokio::test] + async fn test_reply_to_top_level_mention_roots_at_that_post() { + let payload = route_jetstream_event(&mention_commit(None), OWN) + .unwrap_or_else(|| panic!("mention routes")); + let client = RecordingBlueskyClient::default(); + + execute(&client, &reply_input_from_mention(&payload)) + .await + .unwrap_or_else(|e| panic!("reply failed: {e}")); + + let sent = client + .captured() + .unwrap_or_else(|| panic!("reply never reached the client")); + assert_eq!( + sent.parent_uri, + "at://did:plc:someone/app.bsky.feed.post/3kxyz" + ); + assert_eq!(sent.parent_cid, "bafymention"); + // A top-level mention is the root of its own thread. + assert_eq!(sent.root_uri, sent.parent_uri); + assert_eq!(sent.root_cid, sent.parent_cid); + } + + #[tokio::test] + async fn test_reply_to_nested_mention_roots_at_thread_root() { + let commit = mention_commit(Some(serde_json::json!({ + "root": { "uri": "at://did:plc:opener/app.bsky.feed.post/root", "cid": "bafyroot" }, + "parent": { "uri": "at://did:plc:other/app.bsky.feed.post/mid", "cid": "bafymid" } + }))); + let payload = + route_jetstream_event(&commit, OWN).unwrap_or_else(|| panic!("mention routes")); + let client = RecordingBlueskyClient::default(); + + execute(&client, &reply_input_from_mention(&payload)) + .await + .unwrap_or_else(|e| panic!("reply failed: {e}")); + + let sent = client + .captured() + .unwrap_or_else(|| panic!("reply never reached the client")); + // Parent is still the post that mentioned us... + assert_eq!( + sent.parent_uri, + "at://did:plc:someone/app.bsky.feed.post/3kxyz" + ); + // ...but the thread roots where the conversation started, not at + // the mention — otherwise clients file the reply as its own thread. + assert_eq!(sent.root_uri, "at://did:plc:opener/app.bsky.feed.post/root"); + assert_eq!(sent.root_cid, "bafyroot"); + assert_ne!(sent.root_uri, sent.parent_uri); + } #[test] fn test_declaration_name() { diff --git a/connectors/connector-bluesky/src/client/mod.rs b/connectors/connector-bluesky/src/client/mod.rs index 7ff6a51c..2a4a519c 100644 --- a/connectors/connector-bluesky/src/client/mod.rs +++ b/connectors/connector-bluesky/src/client/mod.rs @@ -402,6 +402,86 @@ pub mod test_helpers { pub response: serde_json::Value, } + /// The arguments one `reply` call carried to the API. + #[derive(Clone, Debug, PartialEq, Eq)] + pub struct CapturedReply { + pub text: String, + pub parent_uri: String, + pub parent_cid: String, + pub root_uri: String, + pub root_cid: String, + } + + /// Mock that records what `reply` was asked to send. + /// + /// [`MockBlueskyClient`] discards its arguments, so it cannot answer + /// "what did we put in the reply record?" — the question thread + /// rooting turns on. This one keeps the last call. + #[derive(Default)] + pub struct RecordingBlueskyClient { + pub last_reply: std::sync::Mutex>, + } + + impl RecordingBlueskyClient { + /// The recorded `reply` call, or `None` if `reply` never ran. + pub fn captured(&self) -> Option { + self.last_reply.lock().ok().and_then(|g| g.clone()) + } + } + + #[async_trait] + impl BlueskyApi for RecordingBlueskyClient { + async fn current_account(&self) -> Result<(String, String), BlueskyError> { + Ok(( + "did:plc:mocktestaccount".to_owned(), + "mock.bsky.social".to_owned(), + )) + } + + async fn create_post(&self, _text: &str) -> Result { + Ok(serde_json::json!({})) + } + + async fn reply( + &self, + text: &str, + parent_uri: &str, + parent_cid: &str, + root_uri: &str, + root_cid: &str, + ) -> Result { + if let Ok(mut slot) = self.last_reply.lock() { + *slot = Some(CapturedReply { + text: text.to_owned(), + parent_uri: parent_uri.to_owned(), + parent_cid: parent_cid.to_owned(), + root_uri: root_uri.to_owned(), + root_cid: root_cid.to_owned(), + }); + } + Ok(serde_json::json!({ + "uri": "at://did:plc:mocktestaccount/app.bsky.feed.post/sent", + "cid": "bafysent" + })) + } + + async fn like( + &self, + _subject_uri: &str, + _subject_cid: &str, + ) -> Result { + Ok(serde_json::json!({})) + } + + async fn repost( + &self, + _subject_uri: &str, + _subject_cid: &str, + ) -> Result { + Ok(serde_json::json!({})) + } + } + #[async_trait] impl BlueskyApi for MockBlueskyClient { async fn current_account(&self) -> Result<(String, String), BlueskyError> { diff --git a/connectors/connector-bluesky/src/firehose/mod.rs b/connectors/connector-bluesky/src/firehose/mod.rs index 11f9061a..bb8ab23b 100644 --- a/connectors/connector-bluesky/src/firehose/mod.rs +++ b/connectors/connector-bluesky/src/firehose/mod.rs @@ -99,11 +99,77 @@ pub fn post_mentions_did(record: &serde_json::Value, target_did: &str) -> bool { false } +/// Resolve the thread root of a post record. +/// +/// An AT Protocol reply record carries BOTH refs: +/// +/// ```json +/// "reply": { "root": { "uri": ..., "cid": ... }, +/// "parent": { "uri": ..., "cid": ... } } +/// ``` +/// +/// Clients group a thread by its ROOT, so a reply that names its parent +/// as the root splits off into a thread of its own. When the post is +/// itself a reply, the conversation's real root is `record.reply.root`. +/// A top-level post has no `reply` field and is the root of its own +/// thread, so it falls back to `(uri, cid)` — the post itself. +/// +/// A partial/malformed `reply.root` (only one of uri/cid) also falls +/// back: a half-formed root ref is worse than rooting at the post. +pub fn thread_root(record: &serde_json::Value, uri: &str, cid: &str) -> (String, String) { + let root = record.get("reply").and_then(|r| r.get("root")); + let root_uri = root + .and_then(|r| r.get("uri")) + .and_then(|v| v.as_str()) + .filter(|s| !s.is_empty()); + let root_cid = root + .and_then(|r| r.get("cid")) + .and_then(|v| v.as_str()) + .filter(|s| !s.is_empty()); + + match (root_uri, root_cid) { + (Some(u), Some(c)) => (u.to_owned(), c.to_owned()), + _ => (uri.to_owned(), cid.to_owned()), + } +} + #[cfg(test)] #[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] mod tests { use super::*; + #[test] + fn test_thread_root_top_level_post_is_its_own_root() { + let record = serde_json::json!({ "text": "top level" }); + let (uri, cid) = thread_root(&record, "at://did:plc:a/app.bsky.feed.post/1", "bafyone"); + assert_eq!(uri, "at://did:plc:a/app.bsky.feed.post/1"); + assert_eq!(cid, "bafyone"); + } + + #[test] + fn test_thread_root_reply_uses_record_root_not_parent() { + let record = serde_json::json!({ + "text": "nested reply", + "reply": { + "root": { "uri": "at://did:plc:r/app.bsky.feed.post/root", "cid": "bafyroot" }, + "parent": { "uri": "at://did:plc:p/app.bsky.feed.post/mid", "cid": "bafymid" } + } + }); + let (uri, cid) = thread_root(&record, "at://did:plc:a/app.bsky.feed.post/1", "bafyone"); + assert_eq!(uri, "at://did:plc:r/app.bsky.feed.post/root"); + assert_eq!(cid, "bafyroot"); + } + + #[test] + fn test_thread_root_partial_ref_falls_back_to_post() { + let record = serde_json::json!({ + "reply": { "root": { "uri": "at://did:plc:r/app.bsky.feed.post/root" } } + }); + let (uri, cid) = thread_root(&record, "at://did:plc:a/app.bsky.feed.post/1", "bafyone"); + assert_eq!(uri, "at://did:plc:a/app.bsky.feed.post/1"); + assert_eq!(cid, "bafyone"); + } + #[test] fn test_build_jetstream_url_basic() { let url = build_jetstream_url( diff --git a/connectors/connector-bluesky/src/gateway/mod.rs b/connectors/connector-bluesky/src/gateway/mod.rs index 240d7b4a..3d22bbe3 100644 --- a/connectors/connector-bluesky/src/gateway/mod.rs +++ b/connectors/connector-bluesky/src/gateway/mod.rs @@ -137,12 +137,26 @@ pub fn route_jetstream_event( .and_then(|r| r.as_str()) .unwrap_or_default(); + let uri = format!("at://{did}/app.bsky.feed.post/{rkey}"); + let cid = commit + .get("cid") + .and_then(|c| c.as_str()) + .unwrap_or_default(); + + // The mentioned post may itself be a reply. Its own record names the + // thread's real root, so carry it through: a reply rule then roots + // correctly (`root_uri`/`root_cid` are required by the `reply` + // action) without a second network call. + let (root_uri, root_cid) = firehose::thread_root(record, &uri, cid); + Some(serde_json::json!({ "trigger": trigger, "did": did, "text": record.get("text").and_then(|t| t.as_str()).unwrap_or_default(), - "uri": format!("at://{did}/app.bsky.feed.post/{rkey}"), - "cid": commit.get("cid").and_then(|c| c.as_str()).unwrap_or_default(), + "uri": uri, + "cid": cid, + "root_uri": root_uri, + "root_cid": root_cid, "created_at": event.get("time_us").cloned().unwrap_or(serde_json::Value::Null), })) } @@ -198,6 +212,44 @@ mod tests { assert_eq!(p["uri"], "at://did:plc:someone/app.bsky.feed.post/3kxyz"); } + #[test] + fn top_level_mention_roots_at_itself() { + let record = json!({ + "$type": "app.bsky.feed.post", + "text": "hey @me", + "facets": [{ + "features": [{ "$type": "app.bsky.richtext.facet#mention", "did": OWN }], + "index": { "byteStart": 4, "byteEnd": 7 } + }] + }); + let event = post_commit("did:plc:someone", record); + let p = route_jetstream_event(&event, OWN).expect("mention routes"); + assert_eq!(p["root_uri"], p["uri"]); + assert_eq!(p["root_cid"], p["cid"]); + } + + #[test] + fn nested_mention_carries_the_thread_root() { + let record = json!({ + "$type": "app.bsky.feed.post", + "text": "hey @me", + "facets": [{ + "features": [{ "$type": "app.bsky.richtext.facet#mention", "did": OWN }], + "index": { "byteStart": 4, "byteEnd": 7 } + }], + "reply": { + "root": { "uri": "at://did:plc:opener/app.bsky.feed.post/root", "cid": "bafyroot" }, + "parent": { "uri": "at://did:plc:other/app.bsky.feed.post/mid", "cid": "bafymid" } + } + }); + let event = post_commit("did:plc:someone", record); + let p = route_jetstream_event(&event, OWN).expect("mention routes"); + assert_eq!(p["uri"], "at://did:plc:someone/app.bsky.feed.post/3kxyz"); + assert_eq!(p["root_uri"], "at://did:plc:opener/app.bsky.feed.post/root"); + assert_eq!(p["root_cid"], "bafyroot"); + assert_ne!(p["root_uri"], p["uri"]); + } + #[test] fn ignores_unrelated_post() { let event = post_commit( diff --git a/connectors/connector-bluesky/src/triggers/mod.rs b/connectors/connector-bluesky/src/triggers/mod.rs index f1a391c8..769e0628 100644 --- a/connectors/connector-bluesky/src/triggers/mod.rs +++ b/connectors/connector-bluesky/src/triggers/mod.rs @@ -21,6 +21,14 @@ fn mention() -> TriggerDecl { "did": { "type": "string", "description": "DID of the post author." }, "uri": { "type": "string", "description": "AT URI of the post." }, "cid": { "type": "string", "description": "CID of the post." }, + "root_uri": { + "type": "string", + "description": "AT URI of the thread root. Equals `uri` when the mention is a top-level post; when the mention is itself a reply this is the conversation's real root, so a reply rule threads correctly." + }, + "root_cid": { + "type": "string", + "description": "CID of the thread root. Equals `cid` for a top-level mention." + }, "text": { "type": "string", "description": "Post text." }, "facets": { "type": "array", @@ -35,7 +43,7 @@ fn mention() -> TriggerDecl { }, "createdAt": { "type": "string", "description": "ISO 8601 timestamp." } }, - "required": ["did", "uri", "text"] + "required": ["did", "uri", "cid", "root_uri", "root_cid", "text"] })), } } diff --git a/crates/springtale-runtime/src/operations/recipes/builtin/messaging.rs b/crates/springtale-runtime/src/operations/recipes/builtin/messaging.rs index 4d0151e1..3e3d1b6d 100644 --- a/crates/springtale-runtime/src/operations/recipes/builtin/messaging.rs +++ b/crates/springtale-runtime/src/operations/recipes/builtin/messaging.rs @@ -968,8 +968,8 @@ action = "reply" [actions.params] parent_uri = "${trigger.uri}" parent_cid = "${trigger.cid}" -root_uri = "${trigger.uri}" -root_cid = "${trigger.cid}" +root_uri = "${trigger.root_uri}" +root_cid = "${trigger.root_cid}" text = "${last_ai_output}" "# .into(), diff --git a/docs/reference/connectors/bluesky.md b/docs/reference/connectors/bluesky.md index 6a34ccd2..c5cd0f2a 100644 --- a/docs/reference/connectors/bluesky.md +++ b/docs/reference/connectors/bluesky.md @@ -47,11 +47,17 @@ App passwords are recommended over account passwords — they can be revoked ind | Name | Source | Payload fields | |------|--------|---------------| -| `mention` | Jetstream (filtered `app.bsky.feed.post`) | `did`, `uri`, `cid`, `text`, `facets`, `createdAt` | +| `mention` | Jetstream (filtered `app.bsky.feed.post`) | `did`, `uri`, `cid`, `root_uri`, `root_cid`, `text`, `facets`, `createdAt` | | `follow` | Jetstream (`app.bsky.graph.follow`) | `did`, `uri`, `subject`, `createdAt` | | `like` | Jetstream (`app.bsky.feed.like`) | `did`, `subject` (`uri`, `cid`), `createdAt` | | `repost` | Jetstream (`app.bsky.feed.repost`) | `did`, `subject` (`uri`, `cid`), `createdAt` | +`root_uri`/`root_cid` are the **thread root** of the mentioned post: the +post's own `uri`/`cid` when it is top-level, and the conversation's real +root when the mention is itself a reply. Pass them straight to the +`reply` action — using `${trigger.uri}` as the root instead splits the +reply into a thread of its own, because clients group threads by root. + All triggers are delivered via the Jetstream WebSocket firehose — a real-time stream of ATProto events. ## 4. Actions