From 611d544cd1f8ba226468abcaad81df9160a30627 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Sat, 17 Oct 2026 23:54:18 +0000 Subject: [PATCH] feat(slack): durable-host acknowledgement reaction on SlackHost Co-Authored-By: Ishaan Jaffer <155045088+ishaan-berri@users.noreply.github.com> --- docs/slack.md | 5 + python/agentchat/channels/slack/host.pyi | 6 + scripts/python-typing-smoke.py | 2 + src/agentchat-slack/src/host.rs | 17 +++ src/agentchat-slack/src/socket.rs | 18 ++- src/agentchat-slack/tests/host.rs | 49 ++++++-- src/bindings/python/src/providers/slack.rs | 16 ++- .../python/src/providers/slack_host.rs | 53 ++++++++- tests/channels/slack/test_host.py | 109 ++++++++++++++++++ 9 files changed, 260 insertions(+), 15 deletions(-) diff --git a/docs/slack.md b/docs/slack.md index 8be4f9b..ea785e8 100644 --- a/docs/slack.md +++ b/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 diff --git a/python/agentchat/channels/slack/host.pyi b/python/agentchat/channels/slack/host.pyi index 96e3bfa..dda3c15 100644 --- a/python/agentchat/channels/slack/host.pyi +++ b/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 = ..., diff --git a/scripts/python-typing-smoke.py b/scripts/python-typing-smoke.py index 103427d..7e2e971 100644 --- a/scripts/python-typing-smoke.py +++ b/scripts/python-typing-smoke.py @@ -50,6 +50,7 @@ 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",)) @@ -57,6 +58,7 @@ async def check_slack_host( 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) diff --git a/src/agentchat-slack/src/host.rs b/src/agentchat-slack/src/host.rs index 83a212d..bcd5b9d 100644 --- a/src/agentchat-slack/src/host.rs +++ b/src/agentchat-slack/src/host.rs @@ -10,6 +10,16 @@ pub struct SlackHost { pub bot_user_id: String, pub workspace_id: Option, pub exclude_prefixes: Vec, + pub ack_emoji: Option, +} + +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 { .flatten() } impl SlackHost { + pub fn acknowledgement(&self, origin: &ThreadOrigin) -> Option { + 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(); diff --git a/src/agentchat-slack/src/socket.rs b/src/agentchat-slack/src/socket.rs index f7dd0c0..3fa0807 100644 --- a/src/agentchat-slack/src/socket.rs +++ b/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>; @@ -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; } diff --git a/src/agentchat-slack/tests/host.rs b/src/agentchat-slack/tests/host.rs index 86e711c..8cc5733 100644 --- a/src/agentchat-slack/tests/host.rs +++ b/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); diff --git a/src/bindings/python/src/providers/slack.rs b/src/bindings/python/src/providers/slack.rs index 3c8e405..3112197 100644 --- a/src/bindings/python/src/providers/slack.rs +++ b/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::(py)) { return Err(error); diff --git a/src/bindings/python/src/providers/slack_host.rs b/src/bindings/python/src/providers/slack_host.rs index 953a910..2cd5994 100644 --- a/src/bindings/python/src/providers/slack_host.rs +++ b/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, exclude_prefixes: Vec, + 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, + timeout: f64, + ) -> PyResult> { + 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::(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::(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))] diff --git a/tests/channels/slack/test_host.py b/tests/channels/slack/test_host.py index c718974..30fbef1 100644 --- a/tests/channels/slack/test_host.py +++ b/tests/channels/slack/test_host.py @@ -306,3 +306,112 @@ async def request(params): with pytest.raises(ValueError, match="different event"): await HOST.thread_context(source, request) assert not calls + + +@pytest.mark.asyncio +async def test_acknowledgement_uses_the_thread_reply_message_timestamp(): + source = HOST.normalize_event(event()) + assert source is not None + calls = [] + + async def request(api_method, payload): + calls.append((api_method, payload)) + return {"ok": True} + + assert await HOST.acknowledge(source, request) is True + assert calls == [( + "reactions.add", + {"channel": ORIGIN["channel_id"], "name": "eyes", "timestamp": ORIGIN["message_ts"]}, + )] + + +@pytest.mark.asyncio +async def test_acknowledgement_is_disabled_without_an_emoji(): + source = HOST.normalize_event(event()) + assert source is not None + calls = [] + + async def request(api_method, payload): + calls.append((api_method, payload)) + return {"ok": True} + + host = SlackHost(bot_user_id=BOT, workspace_id=ORIGIN["team_id"], ack_emoji=None) + assert await host.acknowledge(source, request) is False + assert calls == [] + + +@pytest.mark.asyncio +async def test_acknowledgement_accepts_already_reacted_and_rejects_provider_errors(): + source = HOST.normalize_event(event()) + assert source is not None + + async def already_reacted(api_method, payload): + return {"ok": False, "error": "already_reacted"} + + async def missing_scope(api_method, payload): + return {"ok": False, "error": "missing_scope"} + + assert await HOST.acknowledge(source, already_reacted) is True + assert await HOST.acknowledge(source, missing_scope) is False + + +@pytest.mark.asyncio +async def test_acknowledgement_swallows_callback_exceptions_and_timeouts(): + source = HOST.normalize_event(event()) + assert source is not None + + async def raising(api_method, payload): + raise PermissionError("authorization revoked") + + async def slow(api_method, payload): + await asyncio.Event().wait() + + assert await HOST.acknowledge(source, raising) is False + assert await HOST.acknowledge(source, slow, timeout=0.01) is False + + +@pytest.mark.asyncio +async def test_acknowledgement_cancellation_propagates(): + source = HOST.normalize_event(event()) + assert source is not None + started = asyncio.Event() + + async def blocked(api_method, payload): + started.set() + await asyncio.Event().wait() + + pending = asyncio.create_task(HOST.acknowledge(source, blocked)) + await started.wait() + pending.cancel() + with pytest.raises(asyncio.CancelledError): + await pending + + +@pytest.mark.asyncio +async def test_acknowledgement_rejects_invalid_event_before_callback(): + source = HOST.normalize_event(event()) + assert source is not None + invalid = dict(source) + invalid.pop("message_ts") + calls = [] + + async def request(api_method, payload): + calls.append((api_method, payload)) + return {"ok": True} + + with pytest.raises(ValueError, match="Invalid Slack event origin"): + await HOST.acknowledge(invalid, request) + assert calls == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("timeout_value", [0, -1, float("inf"), float("nan")]) +async def test_acknowledgement_rejects_nonpositive_or_nonfinite_timeout(timeout_value): + source = HOST.normalize_event(event()) + assert source is not None + + async def request(api_method, payload): + return {"ok": True} + + with pytest.raises(ValueError, match="positive finite timeout"): + await HOST.acknowledge(source, request, timeout=timeout_value)