feat(slack): durable-host acknowledgement reaction on SlackHost

docs/slack.md

@@ -537,6 +537,11 @@ if event is not None:
     context = await host.thread_context(event, authorized_page_request)
 ```
 
+After deciding to handle a normalized event, call `await host.acknowledge(event,
+request)` before slower work such as reading thread context. This best-effort
+`reactions.add` uses the event's `message_ts`; it returns `False` when disabled,
+timed out, or not accepted.
+
 `authorized_page_request` is an async callback receiving `conversations.replies`
 parameters. It returns the raw Slack response (`ok`, `messages`, `has_more`, and
 `response_metadata.next_cursor`). Connector adapters may instead return their

python/agentchat/channels/slack/host.pyi

@@ -38,9 +38,15 @@ class SlackHost:
     def __init__(
         self, *, bot_user_id: str, workspace_id: str | None = ...,
         exclude_prefixes: Sequence[str] = ...,
+        ack_emoji: str | None = ...,
     ) -> None: ...
     def routing_text(self, text: str) -> SlackRouting: ...
     def normalize_event(self, payload: Mapping[str, object]) -> SlackHostEvent | None: ...
+    async def acknowledge(
+        self, event: Mapping[str, object],
+        request: Callable[[str, dict[str, object]], Awaitable[Mapping[str, object]]],
+        *, timeout: float = ...,  # noqa: ASYNC109 - bounded native callback protocol
+    ) -> bool: ...
     async def thread_context(
         self, origin: Mapping[str, object],
         request_page: Callable[[dict[str, object]], Awaitable[Mapping[str, object]]] | None = ...,

scripts/python-typing-smoke.py

@@ -50,13 +50,15 @@ async def check_api(
 async def check_slack_host(
     payload: Mapping[str, object],
     request_page: Callable[[dict[str, object]], Awaitable[Mapping[str, object]]],
+    request_ack: Callable[[str, dict[str, object]], Awaitable[Mapping[str, object]]],
     reply: RichReply,
 ) -> None:
     host = SlackHost(bot_user_id="UBOT00001", exclude_prefixes=("!aside",))
     assert_type(host.routing_text("hello"), SlackRouting)
     normalized = host.normalize_event(payload)
     assert_type(normalized, SlackHostEvent | None)
     if normalized is not None:
+        assert_type(await host.acknowledge(normalized, request_ack), bool)
         assert_type(await host.thread_context(normalized, request_page), ThreadContext)
         assert_type(await host.thread_context(normalized), ThreadContext)
     assert_type(message_request(reply, channel_id="CCHANNEL1"), SlackMessageRequest)

src/agentchat-slack/src/host.rs

@@ -10,6 +10,16 @@ pub struct SlackHost {
     pub bot_user_id: String,
     pub workspace_id: Option<String>,
     pub exclude_prefixes: Vec<String>,
+    pub ack_emoji: Option<String>,
+}
+
+pub fn ack_request(channel_id: &str, emoji: &str, message_ts: &str) -> Value {
+    json!({"channel":channel_id,"name":emoji,"timestamp":message_ts})
+}
+
+pub fn ack_accepted(response: &Value) -> bool {
+    response.get("ok").and_then(Value::as_bool) == Some(true)
+        || response.get("error").and_then(Value::as_str) == Some("already_reacted")
 }
 
 #[derive(Clone, Debug, Serialize)]
@@ -63,6 +73,13 @@ fn timestamp(value: &str) -> Option<SlackTimestamp> {
         .flatten()
 }
 impl SlackHost {
+    pub fn acknowledgement(&self, origin: &ThreadOrigin) -> Option<Value> {
+        self.ack_emoji
+            .as_deref()
+            .filter(|emoji| !emoji.is_empty())
+            .map(|emoji| ack_request(&origin.channel_id, emoji, &origin.message_ts))
+    }
+
     pub fn routing_text(&self, text: &str) -> SlackRouting {
         let routed = events::routing_text(text, Some(&self.bot_user_id));
         let lower = routed.text.to_lowercase();

src/agentchat-slack/src/socket.rs

@@ -17,7 +17,7 @@ use tokio_tungstenite::{
     MaybeTlsStream, WebSocketStream,
 };
 
-use super::{Admission, Inner, Slack, CLOSED, DELIVERY_SCOPE, NEW, RUNNING};
+use super::{host::ack_request, Admission, Inner, Slack, CLOSED, DELIVERY_SCOPE, NEW, RUNNING};
 
 type Socket = WebSocketStream<MaybeTlsStream<TcpStream>>;
 
@@ -328,9 +328,19 @@ impl Slack {
                             .client
                             .call(
                                 "reactions.add",
-                                &json!({
-                            "channel":message.metadata.get("channel_id"), "name":emoji,
-                            "timestamp":message.metadata.get("message_timestamp")}),
+                                &ack_request(
+                                    message
+                                        .metadata
+                                        .get("channel_id")
+                                        .and_then(Value::as_str)
+                                        .unwrap_or_default(),
+                                    emoji,
+                                    message
+                                        .metadata
+                                        .get("message_timestamp")
+                                        .and_then(Value::as_str)
+                                        .unwrap_or_default(),
+                                ),
                             )
                             .await;
                     }

src/agentchat-slack/tests/host.rs

@@ -1,5 +1,6 @@
 use agentchat_slack::host::{
-    ContextLimits, CoverageStatus, SlackHost, ThreadContextReader, ThreadOrigin,
+    ack_accepted, ack_request, ContextLimits, CoverageStatus, SlackHost, ThreadContextReader,
+    ThreadOrigin,
 };
 use serde_json::json;
 
@@ -8,17 +9,21 @@ fn host() -> SlackHost {
         bot_user_id: "UBOT00001".into(),
         workspace_id: Some("TTEST0001".into()),
         exclude_prefixes: vec!["!aside".into()],
+        ack_emoji: Some("eyes".into()),
+    }
+}
+fn origin() -> ThreadOrigin {
+    ThreadOrigin {
+        team_id: "TTEST0001".into(),
+        channel_id: "CTEST0001".into(),
+        thread_ts: "1780000000.000001".into(),
+        message_ts: "1780000000.000010".into(),
     }
 }
 fn reader(max_pages: usize, max_chars: usize) -> ThreadContextReader {
     ThreadContextReader::new(
         host(),
-        ThreadOrigin {
-            team_id: "TTEST0001".into(),
-            channel_id: "CTEST0001".into(),
-            thread_ts: "1780000000.000001".into(),
-            message_ts: "1780000000.000010".into(),
-        },
+        origin(),
         ContextLimits {
             max_pages,
             max_chars,
@@ -31,6 +36,36 @@ fn record(ts: &str, text: &str) -> serde_json::Value {
     json!({"ts":ts,"user":"UAUTHOR01","text":text})
 }
 
+#[test]
+fn acknowledgement_payload_uses_the_message_timestamp() {
+    let expected = json!({
+        "channel":"CTEST0001",
+        "name":"eyes",
+        "timestamp":"1780000000.000010",
+    });
+    assert_eq!(
+        ack_request("CTEST0001", "eyes", "1780000000.000010"),
+        expected
+    );
+    assert_eq!(host().acknowledgement(&origin()), Some(expected));
+}
+
+#[test]
+fn acknowledgement_can_be_disabled() {
+    let mut host = host();
+    host.ack_emoji = None;
+    assert_eq!(host.acknowledgement(&origin()), None);
+    host.ack_emoji = Some(String::new());
+    assert_eq!(host.acknowledgement(&origin()), None);
+}
+
+#[test]
+fn acknowledgement_acceptance_includes_already_reacted() {
+    assert!(ack_accepted(&json!({"ok":true})));
+    assert!(ack_accepted(&json!({"ok":false,"error":"already_reacted"})));
+    assert!(!ack_accepted(&json!({"ok":false,"error":"missing_scope"})));
+}
+
 #[test]
 fn parent_preview_survives_and_current_composer_text_is_not_repeated() {
     let mut read = reader(5, 40000);

src/bindings/python/src/providers/slack.rs

@@ -7,6 +7,7 @@ use crate::{convert, host};
 use agentchat_core::{Message, Role};
 use agentchat_slack::{
     events,
+    host::ack_request,
     policy::{
         self, DeliveryKind, MessageReceipt, MirroredReply, ReplyDestination, ReplyMessageRoute,
         ThreadHistory,
@@ -509,7 +510,20 @@ impl Slack {
         host::coroutine(py, async move {
             if let Some(emoji) = emoji.filter(|e| !e.is_empty()) {
                 let message = from_message(&message)?;
-                let result=acall(&web,"reactions_add",vec![],Some(j(json!({"channel":message.metadata.get("channel_id"),"name":emoji,"timestamp":message.metadata.get("message_timestamp")}))?)).await;
+                let payload = ack_request(
+                    message
+                        .metadata
+                        .get("channel_id")
+                        .and_then(|value| value.as_str())
+                        .unwrap_or_default(),
+                    &emoji,
+                    message
+                        .metadata
+                        .get("message_timestamp")
+                        .and_then(|value| value.as_str())
+                        .unwrap_or_default(),
+                );
+                let result = acall(&web, "reactions_add", vec![], Some(j(payload)?)).await;
                 if let Err(error) = result {
                     if !Python::attach(|py| error.is_instance_of::<PyException>(py)) {
                         return Err(error);

src/bindings/python/src/providers/slack_host.rs

@@ -2,10 +2,11 @@
 use super::{slack_files::policy_error, transport::*};
 use crate::{convert, host};
 use agentchat_slack::host::{
-    ContextLimits, CurrentReference, SlackHost as CoreHost, ThreadContextReader, ThreadOrigin,
+    ack_accepted, ContextLimits, CurrentReference, SlackHost as CoreHost, ThreadContextReader,
+    ThreadOrigin,
 };
 use pyo3::{
-    exceptions::{PyTimeoutError, PyValueError},
+    exceptions::{PyException, PyTimeoutError, PyValueError},
     prelude::*,
 };
 use serde_json::json;
@@ -18,17 +19,19 @@ struct SlackHost {
 #[pymethods]
 impl SlackHost {
     #[new]
-    #[pyo3(signature=(*,bot_user_id,workspace_id=None,exclude_prefixes=Vec::new()))]
+    #[pyo3(signature=(*,bot_user_id,workspace_id=None,exclude_prefixes=Vec::new(),ack_emoji=Some("eyes")))]
     fn new(
         bot_user_id: String,
         workspace_id: Option<String>,
         exclude_prefixes: Vec<String>,
+        ack_emoji: Option<&str>,
     ) -> Self {
         Self {
             inner: CoreHost {
                 bot_user_id,
                 workspace_id,
                 exclude_prefixes,
+                ack_emoji: ack_emoji.map(str::to_owned),
             },
         }
     }
@@ -42,6 +45,50 @@ impl SlackHost {
             .map_err(policy_error)?;
         convert::from_json(py, &json!(event))
     }
+    #[pyo3(signature=(event,request,*,timeout=3.0))]
+    fn acknowledge(
+        &self,
+        py: Python<'_>,
+        event: &Bound<'_, PyAny>,
+        request: Py<PyAny>,
+        timeout: f64,
+    ) -> PyResult<Py<PyAny>> {
+        if !timeout.is_finite() || timeout <= 0.0 {
+            return Err(PyValueError::new_err("Expected a positive finite timeout"));
+        }
+        let source = convert::to_json(event)
+            .map_err(|_| PyValueError::new_err("Invalid Slack event origin"))?;
+        let origin: ThreadOrigin = serde_json::from_value(source)
+            .map_err(|_| PyValueError::new_err("Invalid Slack event origin"))?;
+        let payload = self.inner.acknowledgement(&origin);
+        host::coroutine(py, async move {
+            let Some(payload) = payload else {
+                return j(json!(false));
+            };
+            let args = vec![j(json!("reactions.add"))?, j(payload)?];
+            let pending = match call(&request, "__call__", args, None) {
+                Ok(pending) => pending,
+                Err(error) if Python::attach(|py| error.is_instance_of::<PyException>(py)) => {
+                    return j(json!(false));
+                }
+                Err(error) => return Err(error),
+            };
+            let bounded = call(
+                &module("asyncio")?,
+                "wait_for",
+                vec![pending],
+                Some(j(json!({"timeout":timeout}))?),
+            )?;
+            let response = match host::await_py(bounded).await {
+                Ok(response) => response,
+                Err(error) if Python::attach(|py| error.is_instance_of::<PyException>(py)) => {
+                    return j(json!(false));
+                }
+                Err(error) => return Err(error),
+            };
+            j(json!(ack_accepted(&value(&response)?)))
+        })
+    }
     // Explicit Python keyword arguments are the public FFI contract.
     #[allow(clippy::too_many_arguments)]
     #[pyo3(signature=(origin,request_page=None,*,max_pages=5,max_chars=40_000,max_message_chars=8_000,timeout=8.0))]