connector-bluesky: reply in the thread the mention came from (finding 124)

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>) -> 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() {

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<Option<CapturedReply>>,
+    }
+
+    impl RecordingBlueskyClient {
+        /// The recorded `reply` call, or `None` if `reply` never ran.
+        pub fn captured(&self) -> Option<CapturedReply> {
+            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<serde_json::Value, BlueskyError> {
+            Ok(serde_json::json!({}))
+        }
+
+        async fn reply(
+            &self,
+            text: &str,
+            parent_uri: &str,
+            parent_cid: &str,
+            root_uri: &str,
+            root_cid: &str,
+        ) -> Result<serde_json::Value, BlueskyError> {
+            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<serde_json::Value, BlueskyError> {
+            Ok(serde_json::json!({}))
+        }
+
+        async fn repost(
+            &self,
+            _subject_uri: &str,
+            _subject_cid: &str,
+        ) -> Result<serde_json::Value, BlueskyError> {
+            Ok(serde_json::json!({}))
+        }
+    }
+
     #[async_trait]
     impl BlueskyApi for MockBlueskyClient {
         async fn current_account(&self) -> Result<(String, String), BlueskyError> {

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(

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(

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"]
         })),
     }
 }

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(),