From c2e7435fc3819569dc3c9f13747f017ec60b0956 Mon Sep 17 00:00:00 2001 From: Ishaan Jaffer Date: Sat, 17 Oct 2026 11:14:34 -0700 Subject: [PATCH] feat: share Slack context and delivery policies with durable hosts --- Cargo.lock | 1 + docs/slack.md | 82 ++++ python/agentchat/channels/slack/host.py | 38 ++ python/agentchat/channels/slack/host.pyi | 50 ++ python/agentchat/channels/slack/rendering.py | 11 +- python/agentchat/channels/slack/rendering.pyi | 12 + scripts/python-typing-smoke.py | 19 + src/agentchat-slack/Cargo.toml | 1 + src/agentchat-slack/src/events.rs | 166 ++++++- src/agentchat-slack/src/host.rs | 451 ++++++++++++++++++ src/agentchat-slack/src/lib.rs | 2 + src/agentchat-slack/src/message_request.rs | 139 ++++++ src/agentchat-slack/src/policy.rs | 39 +- src/agentchat-slack/src/rendering.rs | 9 +- src/agentchat-slack/tests/host.rs | 108 +++++ src/bindings/python/src/lib.rs | 2 + .../python/src/providers/slack_host.rs | 122 +++++ .../python/src/providers/slack_policy.rs | 27 ++ tests/channels/slack/test_host.py | 308 ++++++++++++ tests/channels/slack/test_message_request.py | 58 +++ 20 files changed, 1590 insertions(+), 55 deletions(-) create mode 100644 python/agentchat/channels/slack/host.py create mode 100644 python/agentchat/channels/slack/host.pyi create mode 100644 src/agentchat-slack/src/host.rs create mode 100644 src/agentchat-slack/src/message_request.rs create mode 100644 src/agentchat-slack/tests/host.rs create mode 100644 src/bindings/python/src/providers/slack_host.rs create mode 100644 tests/channels/slack/test_host.py create mode 100644 tests/channels/slack/test_message_request.py diff --git a/Cargo.lock b/Cargo.lock index 5132361..60372ea 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -74,6 +74,7 @@ dependencies = [ "async-trait", "axum", "futures-util", + "regex", "reqwest", "serde", "serde_json", diff --git a/docs/slack.md b/docs/slack.md index 098af89..8be4f9b 100644 --- a/docs/slack.md +++ b/docs/slack.md @@ -338,6 +338,32 @@ before every allocation, byte transfer, and final share. The helper owns upload validation, transfer and receipt verification; it does not require Socket Mode. `agentchat.channels.slack.rendering.rich_payload(reply)` builds content fields while the host supplies its saved routing. +Durable hosts can also delegate routing and update payloads to +`agentchat.channels.slack.rendering.message_request`. It validates the channel and +timestamps, renders the reply, selects `chat.postMessage` or `chat.update`, and +clears old attachments on updates. It does not send, authenticate or retry: + +```python +from agentchat import RichReply +from agentchat.channels.slack.rendering import message_request + +request = message_request( + RichReply("Completed", blocks=()), + channel_id="C01234567", + thread_ts="1792100000.000", + message_ts="1792100000.001", # Omit to post a new reply. + metadata={"event_type": "saved_delivery", "event_payload": {"id": "one"}}, + fallback_blocks=False, +) +# After the host's durable claim and authorization recheck: +result = await request_once(request["method"], request["payload"]) +``` + +By default, empty `RichReply.blocks` generates readable text sections. +`fallback_blocks=False` preserves explicit empty blocks for plain replacements +or card-only replies. Persist the host's delivery identity before calling its +one-attempt transport, and reconcile uncertain outcomes instead of resending. + ## Mirroring web inputs into Slack Use `app.mirror()` to display an already accepted web input in a bound Slack @@ -490,3 +516,59 @@ python -m pip install -e ".[dev]" ruff check . pytest ``` + +### Applications with their own Slack inbox + +`SlackHost` provides the same event routing and message interpretation as the Slack +channel without creating another inbox, credential store, or subscription store. +The application authenticates the incoming event, authorizes each context read, +and owns admission, retries, and persistence. + +```python +from agentchat.channels.slack.host import SlackHost + +host = SlackHost( + bot_user_id="UBOT00001", + workspace_id="TWORK0001", + exclude_prefixes=("!aside",), # Optional application policy. +) +event = host.normalize_event(authenticated_payload) +if event is not None: + context = await host.thread_context(event, authorized_page_request) +``` + +`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 +next cursor in `next_arguments.cursor`. The callback runs on the caller's event +loop; it supplies credentials and enforces the application's access policy. +Authorization exceptions propagate. A provider error returned as `ok: false` +produces explicit context coverage; a failed later page preserves earlier messages. + +The normalized event carries its current preview data bound to the event origin +and author. Passing that event to `thread_context` preserves its previews if the +provider read fails and merges them with fetched previews without duplication. +Pass `request_page=None` (or omit it) when application policy forbids a history +read, such as a direct-message workflow. This performs no I/O and reports +`history_not_requested` with partial or unavailable coverage. A reference copied +from a different event is rejected before the callback runs. + +Context includes the parent and recent messages, with author IDs, file references, +and link previews. Current-message text, files, and composer `rich_text` blocks are +excluded because the current request already contains them; current link previews +remain. Source data is returned as bounded JSON text and must be presented as +untrusted reference material, separately from the user's request. Context reading +does not fetch linked URLs or download attachments. + +Defaults limit the read to five pages, 40,000 content characters, 8,000 characters +per message, and eight seconds. Results report `complete`, `partial`, or +`unavailable`; missing roots, pagination limits, truncation, and provider failures +are explicit. The root is retained when older replies must be removed to meet the +budget. The timestamp cutoff is checked locally even when a connector cannot send +Slack's `latest` parameter. Code and quoted mentions remain literal text and do +not address the bot. Malformed supported events raise `ValueError`; ignored +subtypes and configured exclusions return `None`. + +Existing `Slack.thread_history` remains supported and shares record normalization +with this API. Applications should persist admitted context if retries must see +the same snapshot. diff --git a/python/agentchat/channels/slack/host.py b/python/agentchat/channels/slack/host.py new file mode 100644 index 0000000..f125aff --- /dev/null +++ b/python/agentchat/channels/slack/host.py @@ -0,0 +1,38 @@ +"""Slack protocol helpers for applications that own authentication and durable ingress.""" +from typing import Literal, NotRequired, TypedDict + +from agentchat._native import SlackHost as SlackHost + + +class SlackRouting(TypedDict): + text: str + mentioned: bool + excluded: bool + +class ThreadOrigin(TypedDict): + team_id: str + channel_id: str + thread_ts: str + message_ts: str + +class CurrentReference(ThreadOrigin): + user_id: str + data: dict[str, object] + +class SlackHostEvent(ThreadOrigin, SlackRouting): + user_id: str + direct: bool + files: list[dict[str, object]] + reference: CurrentReference + +class ContextMessage(TypedDict): + ts: str + author: str + content: str + truncated: bool + +class ThreadContext(ThreadOrigin): + status: Literal["complete", "partial", "unavailable"] + messages: list[ContextMessage] + error: NotRequired[str] + warning: NotRequired[str] diff --git a/python/agentchat/channels/slack/host.pyi b/python/agentchat/channels/slack/host.pyi new file mode 100644 index 0000000..96e3bfa --- /dev/null +++ b/python/agentchat/channels/slack/host.pyi @@ -0,0 +1,50 @@ +from collections.abc import Awaitable, Callable, Mapping, Sequence +from typing import Literal, NotRequired, TypedDict + +class SlackRouting(TypedDict): + text: str + mentioned: bool + excluded: bool + +class ThreadOrigin(TypedDict): + team_id: str + channel_id: str + thread_ts: str + message_ts: str + +class CurrentReference(ThreadOrigin): + user_id: str + data: dict[str, object] + +class SlackHostEvent(ThreadOrigin, SlackRouting): + user_id: str + direct: bool + files: list[dict[str, object]] + reference: CurrentReference + +class ContextMessage(TypedDict): + ts: str + author: str + content: str + truncated: bool + +class ThreadContext(ThreadOrigin): + status: Literal["complete", "partial", "unavailable"] + messages: list[ContextMessage] + error: NotRequired[str] + warning: NotRequired[str] + +class SlackHost: + def __init__( + self, *, bot_user_id: str, workspace_id: str | None = ..., + exclude_prefixes: Sequence[str] = ..., + ) -> None: ... + def routing_text(self, text: str) -> SlackRouting: ... + def normalize_event(self, payload: Mapping[str, object]) -> SlackHostEvent | None: ... + async def thread_context( + self, origin: Mapping[str, object], + request_page: Callable[[dict[str, object]], Awaitable[Mapping[str, object]]] | None = ..., + *, + max_pages: int = ..., max_chars: int = ..., max_message_chars: int = ..., + timeout: float = ..., # noqa: ASYNC109 - bounded native callback protocol + ) -> ThreadContext: ... diff --git a/python/agentchat/channels/slack/rendering.py b/python/agentchat/channels/slack/rendering.py index 3bbf9a2..db5fabe 100644 --- a/python/agentchat/channels/slack/rendering.py +++ b/python/agentchat/channels/slack/rendering.py @@ -1,5 +1,14 @@ """Slack rendering implemented in Rust.""" +from typing import Literal, TypedDict + +from agentchat._native import slack_message_request as message_request from agentchat._native import slack_rich_payload as rich_payload from agentchat._native import slack_text_blocks as text_blocks -__all__ = ['rich_payload', 'text_blocks'] + +class SlackMessageRequest(TypedDict): + method: Literal["chat.postMessage", "chat.update"] + payload: dict[str, object] + + +__all__ = ['SlackMessageRequest', 'message_request', 'rich_payload', 'text_blocks'] diff --git a/python/agentchat/channels/slack/rendering.pyi b/python/agentchat/channels/slack/rendering.pyi index 19f5294..cdd4da2 100644 --- a/python/agentchat/channels/slack/rendering.pyi +++ b/python/agentchat/channels/slack/rendering.pyi @@ -1,5 +1,17 @@ # Public compatibility signatures; update with the corresponding Rust binding API. +from collections.abc import Mapping +from typing import Literal, TypedDict + from agentchat.models import RichReply +class SlackMessageRequest(TypedDict): + method: Literal["chat.postMessage", "chat.update"] + payload: dict[str, object] + def text_blocks(text: str) -> list[dict[str, object]]: ... def rich_payload(reply: RichReply) -> dict[str, object]: ... +def message_request( + reply: RichReply, *, channel_id: str, thread_ts: str | None = ..., + message_ts: str | None = ..., metadata: Mapping[str, object] | None = ..., + fallback_blocks: bool = ..., +) -> SlackMessageRequest: ... diff --git a/scripts/python-typing-smoke.py b/scripts/python-typing-smoke.py index 581e3cf..103427d 100644 --- a/scripts/python-typing-smoke.py +++ b/scripts/python-typing-smoke.py @@ -4,6 +4,7 @@ scripts/python-typing-smoke.py, using the project environment's Python executable. """ +from collections.abc import Awaitable, Callable, Mapping from typing import assert_type from agentchat import AgentChat, Conversation, Message, MessageContext, Sender @@ -12,11 +13,14 @@ from agentchat.channels.discord import Discord from agentchat.channels.imessage import IMessage from agentchat.channels.slack import Slack, SlackUser +from agentchat.channels.slack.host import SlackHost, SlackHostEvent, SlackRouting, ThreadContext +from agentchat.channels.slack.rendering import SlackMessageRequest, message_request from agentchat.channels.teams import Teams from agentchat.channels.telegram import Telegram from agentchat.channels.telegram.rendering import RenderedText, render from agentchat.channels.telegram.transport import TelegramBotClient from agentchat.channels.whatsapp import WhatsApp +from agentchat.rich import RichReply from agentchat.state import MemoryState @@ -41,3 +45,18 @@ async def check_api( assert_type(await whatsapp.reply(message, "hello"), Message) assert_type(render("**hello**"), RenderedText) assert_type(TelegramBotClient("99:token"), TelegramBotClient) + + +async def check_slack_host( + payload: Mapping[str, object], + request_page: Callable[[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.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/Cargo.toml b/src/agentchat-slack/Cargo.toml index cdf9909..b69e2b5 100644 --- a/src/agentchat-slack/Cargo.toml +++ b/src/agentchat-slack/Cargo.toml @@ -18,6 +18,7 @@ tokio-tungstenite = { version = "0.30", features = ["rustls-tls-webpki-roots"] } tokio-util = { version = "0.7", features = ["rt"] } tracing = "0.1" unicode-general-category = "1" +regex = "1" [dev-dependencies] axum = { version = "0.8", features = ["ws"] } diff --git a/src/agentchat-slack/src/events.rs b/src/agentchat-slack/src/events.rs index 5c5a017..79e2540 100644 --- a/src/agentchat-slack/src/events.rs +++ b/src/agentchat-slack/src/events.rs @@ -1,7 +1,10 @@ //! Pure Slack event normalization. No authentication, downloads, or network I/O. use agentchat_core::{Attachment, Message, Role, Sender}; +use regex::Regex; +use serde::Serialize; use serde_json::{json, Value}; +use std::sync::OnceLock; use unicode_general_category::{get_general_category, GeneralCategory}; pub(crate) fn nonempty(value: &Value) -> Option<&str> { @@ -12,6 +15,121 @@ pub(crate) fn trim_python(text: &str) -> &str { text.trim_matches(|c: char| c.is_whitespace() || matches!(c, '\u{001c}'..='\u{001f}')) } +#[derive(Clone, Debug, Serialize)] +pub struct RoutedText { + pub text: String, + pub mentioned: bool, +} + +/// Code and quoted examples are source text, not routing instructions. +pub fn routing_text(text: &str, bot_user_id: Option<&str>) -> RoutedText { + static QUOTED: OnceLock = OnceLock::new(); + let quoted = QUOTED.get_or_init(|| { + Regex::new(r"(?m)```[\s\S]*?(?:```|\z)|`[^`\n]*(?:`|$)|^[ \t]*>>>[\s\S]*|^[ \t]*>[^\n]*") + .expect("static Slack quote expression") + }); + let mention = bot_user_id + .filter(|id| !id.is_empty()) + .map(|id| format!("<@{id}>")); + let mut result = String::new(); + let mut mentioned = false; + let mut end = 0; + for span in quoted.find_iter(text) { + let plain = &text[end..span.start()]; + if let Some(ref token) = mention { + mentioned |= plain.contains(token); + result.push_str(&plain.replace(token, "")); + } else { + result.push_str(plain); + } + result.push_str(span.as_str()); + end = span.end(); + } + let plain = &text[end..]; + if let Some(ref token) = mention { + mentioned |= plain.contains(token); + result.push_str(&plain.replace(token, "")); + } else { + result.push_str(plain); + } + RoutedText { + text: trim_python(&result).into(), + mentioned, + } +} + +/// Preserve provider previews as reference data, never instructions or downloads. +pub fn reference(record: &Value, current: bool) -> Value { + let mut source = serde_json::Map::new(); + let fields: &[&str] = if current { + &["attachments", "blocks"] + } else { + &["text", "attachments", "blocks", "files"] + }; + for key in fields { + if let Some(value) = record.get(*key).filter(|v| !v.is_null()) { + source.insert((*key).into(), value.clone()); + } + } + if current { + if let Some(Value::Array(blocks)) = source.get_mut("blocks") { + blocks.retain(|block| block["type"] != "rich_text"); + if blocks.is_empty() { + source.remove("blocks"); + } + } + } + Value::Object(source) +} + +pub(crate) struct HistoryRecord { + pub timestamp: super::route::SlackTimestamp, + pub sender: Sender, + pub role: Role, + pub text: String, + pub files: Vec, + pub reference: Value, +} + +/// Shared by transport history and host context; speaker/mirror semantics have one owner. +pub(crate) fn history_record(record: &Value, bot_user_id: Option<&str>) -> Option { + let timestamp = record["ts"] + .as_str() + .and_then(super::route::SlackTimestamp::parse)?; + let sender = nonempty(&record["user"]).or_else(|| nonempty(&record["bot_id"]))?; + let text = record.get("text").map_or(Some(""), Value::as_str)?; + let files = attachments(&record["files"]); + let source = reference(record, false); + if trim_python(text).is_empty() + && files.is_empty() + && record["attachments"].is_null() + && record["blocks"].is_null() + { + return None; + } + let mirror = bot_user_id.and_then(|bot| super::rendering::read_mirror(record, bot)); + let role = if Some(sender) == bot_user_id && mirror.is_none() { + Role::Assistant + } else { + Role::User + }; + let input = mirror.unwrap_or_else(|| super::rendering::MirroredInput { + sender: Sender { + id: sender.into(), + display_name: None, + }, + text: text.into(), + }); + Some(HistoryRecord { + timestamp, + sender: input.sender, + role, + text: input.text, + files, + reference: source, + }) +} + pub fn valid_file_id(id: &str) -> bool { (2..=31).contains(&id.len()) && id.starts_with('F') @@ -120,23 +238,14 @@ pub fn normalize_scoped( Some(nonempty(get("thread_ts"))?) }; let root = supplied_thread.unwrap_or(timestamp); - let mention = bot_user_id - .filter(|id| !id.is_empty()) - .map(|id| format!("<@{id}>")); + let routed = routing_text(text, bot_user_id); let channel_message = event_type == "message" && matches!(get("channel_type").as_str(), Some("channel" | "group")); - let mentioned = event_type == "app_mention" - || (channel_message - && mention - .as_ref() - .is_some_and(|mention| text.contains(mention))); + let mentioned = event_type == "app_mention" || (channel_message && routed.mentioned); if !(direct || mentioned || (channel_message && supplied_thread.is_some())) { return None; } - let without_mention = mention - .as_ref() - .map_or_else(|| text.to_owned(), |mention| text.replace(mention, "")); - let normalized_text = trim_python(&without_mention).to_owned(); + let normalized_text = routed.text; if normalized_text.is_empty() && files.is_empty() { return None; } @@ -145,22 +254,37 @@ pub fn normalize_scoped( .as_str() .map(str::to_owned) .unwrap_or_else(|| format!("{channel}:{timestamp}")); + let mut metadata = json!({ + "channel_id": channel, "team_id": team, "event_id": event_id, + "channel_type": if direct { Value::String("im".into()) } else { event.get("channel_type").cloned().unwrap_or_else(|| Value::String("channel".into())) }, + "reply_thread_timestamp": thread, "message_timestamp": timestamp, + "requires_subscription": !direct && !mentioned, + }).as_object()?.clone(); + let reference = reference(&payload["event"], true); + if reference + .as_object() + .is_some_and(|source| !source.is_empty()) + { + metadata.insert("slack_reference".into(), reference); + } Some(Message { id: format!("slack:{team}:{channel}:{timestamp}").into(), - conversation_id: if direct { format!("slack:{team}:{channel}") } - else { format!("slack:{team}:{channel}:{root}") }.into(), + conversation_id: if direct { + format!("slack:{team}:{channel}") + } else { + format!("slack:{team}:{channel}:{root}") + } + .into(), channel: channel_name.into(), - sender: Sender { id: user.into(), display_name: None }, + sender: Sender { + id: user.into(), + display_name: None, + }, text: normalized_text, role: Role::User, thread_id: thread.map(str::to_owned), addressed: direct || mentioned, attachments: files, - metadata: json!({ - "channel_id": channel, "team_id": team, "event_id": event_id, - "channel_type": if direct { Value::String("im".into()) } else { event.get("channel_type").cloned().unwrap_or_else(|| Value::String("channel".into())) }, - "reply_thread_timestamp": thread, "message_timestamp": timestamp, - "requires_subscription": !direct && !mentioned, - }).as_object()?.clone(), + metadata, }) } diff --git a/src/agentchat-slack/src/host.rs b/src/agentchat-slack/src/host.rs new file mode 100644 index 0000000..83a212d --- /dev/null +++ b/src/agentchat-slack/src/host.rs @@ -0,0 +1,451 @@ +//! Slack protocol helpers for applications that own authentication and durable ingress. +use crate::{events, policy, route::SlackTimestamp}; +use agentchat_core::{Error, Result}; +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; +use std::collections::HashSet; + +#[derive(Clone, Debug)] +pub struct SlackHost { + pub bot_user_id: String, + pub workspace_id: Option, + pub exclude_prefixes: Vec, +} + +#[derive(Clone, Debug, Serialize)] +pub struct SlackRouting { + pub text: String, + pub mentioned: bool, + pub excluded: bool, +} + +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +pub struct ThreadOrigin { + pub team_id: String, + pub channel_id: String, + pub thread_ts: String, + pub message_ts: String, +} + +/// Projected current-message data bound to the authenticated event's provenance. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct CurrentReference { + #[serde(flatten)] + pub origin: ThreadOrigin, + pub user_id: String, + pub data: Value, +} + +#[derive(Clone, Debug, Serialize)] +pub struct SlackHostEvent { + #[serde(flatten)] + pub origin: ThreadOrigin, + pub user_id: String, + pub text: String, + pub mentioned: bool, + pub direct: bool, + pub excluded: bool, + pub files: Vec, + pub reference: CurrentReference, +} + +fn identifier(value: &str, prefixes: &str) -> bool { + (8..=31).contains(&value.len()) + && value.starts_with(|c| prefixes.contains(c)) + && value + .bytes() + .all(|c| c.is_ascii_uppercase() || c.is_ascii_digit()) +} +fn timestamp(value: &str) -> Option { + let (seconds, fraction) = value.split_once('.')?; + ((10..=16).contains(&seconds.len()) && (1..=9).contains(&fraction.len())) + .then(|| SlackTimestamp::parse(value)) + .flatten() +} +impl SlackHost { + 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(); + let excluded = self.exclude_prefixes.iter().any(|prefix| { + lower + .strip_prefix(&prefix.to_lowercase()) + .is_some_and(|rest| rest.is_empty() || rest.starts_with(char::is_whitespace)) + }); + SlackRouting { + text: routed.text, + mentioned: routed.mentioned, + excluded, + } + } + + pub fn normalize_event(&self, payload: &Value) -> Result> { + let Some(event) = payload.get("event").and_then(Value::as_object) else { + return Ok(None); + }; + if !matches!( + event.get("type").and_then(Value::as_str), + Some("message" | "app_mention") + ) { + return Ok(None); + } + let invalid = || Error::InvalidInput("Invalid Slack event fields".into()); + if ["subtype", "channel_type"].iter().any(|key| { + event + .get(*key) + .is_some_and(|value| !value.is_null() && !value.is_string()) + }) { + return Err(invalid()); + } + if event + .get("subtype") + .is_some_and(|value| !value.is_null() && value != "file_share") + { + return Ok(None); + } + if event.get("bot_id").is_some_and(|value| !value.is_null()) + || event + .get("bot_profile") + .is_some_and(|value| !value.is_null()) + || event.get("user").and_then(Value::as_str) == Some(&self.bot_user_id) + { + return Ok(None); + } + let text = event + .get("text") + .map_or(Some(""), Value::as_str) + .ok_or_else(invalid)?; + let user = event + .get("user") + .and_then(Value::as_str) + .ok_or_else(invalid)?; + let origin = ThreadOrigin { + team_id: payload["team_id"].as_str().ok_or_else(invalid)?.into(), + channel_id: event + .get("channel") + .and_then(Value::as_str) + .ok_or_else(invalid)? + .into(), + message_ts: event + .get("ts") + .and_then(Value::as_str) + .ok_or_else(invalid)? + .into(), + thread_ts: event + .get("thread_ts") + .filter(|v| !v.is_null()) + .or_else(|| event.get("ts")) + .and_then(Value::as_str) + .ok_or_else(invalid)? + .into(), + }; + self.validate_origin(&origin)?; + if !identifier(user, "UW") { + return Err(invalid()); + } + let routing = self.routing_text(text); + if routing.excluded { + return Ok(None); + } + let mut input = payload.clone(); + if input["event"]["channel_type"].is_null() && !origin.channel_id.starts_with('D') { + input["event"]["channel_type"] = json!("channel"); + } + let Some(message) = events::normalize_scoped( + &input, + Some(&self.bot_user_id), + self.workspace_id.as_deref(), + "slack", + ) else { + return Ok(None); + }; + Ok(Some(SlackHostEvent { + direct: origin.channel_id.starts_with('D') && payload["event"]["channel_type"] == "im", + mentioned: event["type"] == "app_mention" || routing.mentioned, + user_id: user.into(), + text: message.text, + excluded: routing.excluded, + files: message.attachments, + reference: CurrentReference { + origin: origin.clone(), + user_id: user.into(), + data: message + .metadata + .get("slack_reference") + .cloned() + .unwrap_or_else(|| json!({})), + }, + origin, + })) + } + + fn validate_origin(&self, origin: &ThreadOrigin) -> Result<()> { + let invalid = || Error::InvalidInput("Invalid Slack thread origin or workspace".into()); + if !identifier(&origin.team_id, "T") + || !identifier(&origin.channel_id, "CDG") + || self + .workspace_id + .as_ref() + .is_some_and(|team| team != &origin.team_id) + { + return Err(invalid()); + } + policy::ReplyDestination::validate(&origin.channel_id, Some(&origin.thread_ts))?; + let root = timestamp(&origin.thread_ts).ok_or_else(invalid)?; + let current = timestamp(&origin.message_ts).ok_or_else(invalid)?; + if current.is_before(&root) { + return Err(invalid()); + } + Ok(()) + } +} + +#[derive(Clone, Copy, Debug)] +pub struct ContextLimits { + pub max_pages: usize, + pub max_chars: usize, + pub max_message_chars: usize, +} +impl ContextLimits { + pub fn validate(self) -> Result { + if self.max_pages == 0 + || self.max_pages > 100 + || self.max_chars == 0 + || self.max_chars > 1_000_000 + || self.max_message_chars == 0 + || self.max_message_chars > 1_000_000 + { + return Err(Error::InvalidInput("Invalid Slack context limits".into())); + } + Ok(self) + } +} +#[derive(Clone, Copy, Debug, Serialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum CoverageStatus { + Complete, + Partial, + Unavailable, +} +#[derive(Clone, Debug, Serialize)] +pub struct ContextMessage { + pub ts: String, + pub author: String, + pub content: String, + pub truncated: bool, +} +#[derive(Clone, Debug, Serialize)] +pub struct ThreadContext { + #[serde(flatten)] + pub origin: ThreadOrigin, + pub status: CoverageStatus, + pub messages: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + pub error: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub warning: Option, +} + +/// Stateful page policy reused by host runtimes; contains no credentials or I/O. +pub struct ThreadContextReader { + host: SlackHost, + context: ThreadContext, + limits: ContextLimits, + root: SlackTimestamp, + current: SlackTimestamp, + cursor: String, + cursors: HashSet, + seen: HashSet, + pages: usize, + current_reference: Value, + current_author: Option, +} +impl ThreadContextReader { + pub fn new(host: SlackHost, origin: ThreadOrigin, limits: ContextLimits) -> Result { + host.validate_origin(&origin)?; + let root = timestamp(&origin.thread_ts).expect("validated root"); + let current = timestamp(&origin.message_ts).expect("validated message"); + Ok(Self { + host, + limits: limits.validate()?, + root, + current, + cursor: String::new(), + cursors: HashSet::new(), + seen: HashSet::new(), + pages: 0, + current_reference: json!({}), + current_author: None, + context: ThreadContext { + origin, + status: CoverageStatus::Complete, + messages: Vec::new(), + error: None, + warning: None, + }, + }) + } + pub fn seed(&mut self, reference: CurrentReference, user_id: &str) -> Result<()> { + if reference.origin != self.context.origin + || reference.user_id != user_id + || !identifier(user_id, "UW") + { + return Err(Error::InvalidInput( + "Slack reference belongs to a different event".into(), + )); + } + self.current_author = Some(user_id.into()); + let source = self.merge_current(events::reference(&reference.data, true)); + let ts = self.current.as_str().to_owned(); + self.store(&ts, user_id, source); + Ok(()) + } + fn merge_current(&mut self, source: Value) -> Value { + for (key, incoming) in source.as_object().expect("projected reference") { + if let (Some(Value::Array(existing)), Value::Array(items)) = + (self.current_reference.get_mut(key), incoming) + { + for item in items { + if !existing.contains(item) { + existing.push(item.clone()); + } + } + } else { + self.current_reference[key] = incoming.clone(); + } + } + self.current_reference.clone() + } + pub fn request(&self) -> Value { + json!({"channel":self.context.origin.channel_id,"ts":self.context.origin.thread_ts, + "latest":self.context.origin.message_ts,"inclusive":true,"limit":100, + "cursor":self.cursor,"include_all_metadata":true}) + } + pub fn fail(&mut self, code: &str) { + self.context.status = if self.context.messages.is_empty() { + CoverageStatus::Unavailable + } else { + CoverageStatus::Partial + }; + self.context.error = Some(code.chars().take(100).collect()); + } + fn partial(&mut self, reason: &str) { + self.context.status = CoverageStatus::Partial; + self.context.warning = Some(reason.into()); + } + fn include(&mut self, record: &Value) { + let Some(input) = events::history_record(record, Some(&self.host.bot_user_id)) else { + return; + }; + let ts = &input.timestamp; + if ts.is_before(&self.root) + || self.current.is_before(ts) + || record + .get("thread_ts") + .is_some_and(|thread| thread != self.root.as_str()) + || self.host.routing_text(&input.text).excluded + || !self.seen.insert(ts.as_str().into()) + { + return; + } + let source = if ts.as_str() == self.current.as_str() { + if self + .current_author + .as_ref() + .is_some_and(|author| author != &input.sender.id) + { + self.partial("current message author mismatch"); + return; + } + self.merge_current(events::reference(record, true)) + } else { + input.reference + }; + self.store(ts.as_str(), &input.sender.id, source); + } + fn store(&mut self, ts: &str, author: &str, source: Value) { + if source.as_object().is_none_or(|v| v.is_empty()) { + return; + } + let serialized = source.to_string(); + let limit = self.limits.max_message_chars.min(self.limits.max_chars); + let content: String = serialized.chars().take(limit).collect(); + let truncated = serialized.chars().count() > limit; + if truncated { + self.partial("message limit reached"); + } + self.context.messages.retain(|message| message.ts != ts); + self.context.messages.push(ContextMessage { + ts: ts.into(), + author: author.into(), + content, + truncated, + }); + self.context.messages.sort_by(|a, b| { + let a = SlackTimestamp::parse(&a.ts).expect("validated timestamp"); + let b = SlackTimestamp::parse(&b.ts).expect("validated timestamp"); + if a.is_before(&b) { + std::cmp::Ordering::Less + } else if b.is_before(&a) { + std::cmp::Ordering::Greater + } else { + std::cmp::Ordering::Equal + } + }); + while self + .context + .messages + .iter() + .map(|m| m.content.chars().count()) + .sum::() + > self.limits.max_chars + { + let index = usize::from(self.context.messages[0].ts == self.root.as_str()); + self.context.messages.remove(index); + self.partial("context limit reached"); + } + } + /// Returns true when fetching should stop; incomplete reads remain explicit. + pub fn accept(&mut self, page: &Value) -> bool { + if page["ok"] != true { + self.fail(page["error"].as_str().unwrap_or("invalid_response")); + return true; + } + let Some(records) = page["messages"].as_array() else { + self.fail("invalid_response"); + return true; + }; + for record in records { + self.include(record); + } + self.pages += 1; + if page["has_more"] == false { + return true; + } + if page["has_more"] != true { + self.partial("pagination coverage unavailable"); + return true; + } + let next = page["response_metadata"]["next_cursor"] + .as_str() + .or_else(|| page["next_arguments"]["cursor"].as_str()) + .unwrap_or(""); + if next.is_empty() || !self.cursors.insert(next.into()) { + self.partial("invalid pagination cursor"); + return true; + } + if self.pages >= self.limits.max_pages { + self.partial("page limit reached"); + return true; + } + self.cursor = next.into(); + false + } + pub fn finish(mut self) -> ThreadContext { + if self.context.status == CoverageStatus::Complete + && !self.seen.contains(self.root.as_str()) + { + self.partial("The thread root was not returned."); + } + self.context + } +} diff --git a/src/agentchat-slack/src/lib.rs b/src/agentchat-slack/src/lib.rs index b63f5d9..e4b01d2 100644 --- a/src/agentchat-slack/src/lib.rs +++ b/src/agentchat-slack/src/lib.rs @@ -7,6 +7,8 @@ pub mod events; pub mod file_transfers; mod files; +pub mod host; +pub mod message_request; pub mod policy; pub mod rendering; mod route; diff --git a/src/agentchat-slack/src/message_request.rs b/src/agentchat-slack/src/message_request.rs new file mode 100644 index 0000000..cb94e46 --- /dev/null +++ b/src/agentchat-slack/src/message_request.rs @@ -0,0 +1,139 @@ +//! Validated Slack message requests for hosts that own durable delivery. + +use crate::{policy::ReplyDestination, rendering, route::SlackTimestamp}; +use agentchat_core::{Error, Result, RichReply}; +use serde::Serialize; +use serde_json::{json, Value}; + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +pub enum MessageMethod { + #[serde(rename = "chat.postMessage")] + Post, + #[serde(rename = "chat.update")] + Update, +} + +/// A prepared request, with no transport, retry, authorization or queue policy. +#[derive(Clone, Debug, Serialize)] +pub struct MessageRequest { + pub method: MessageMethod, + pub payload: Value, +} + +impl MessageRequest { + pub fn new( + reply: &RichReply, + channel_id: &str, + thread_ts: Option<&str>, + message_ts: Option<&str>, + metadata: Option<&Value>, + fallback_blocks: bool, + ) -> Result { + let destination = ReplyDestination::validate(channel_id, thread_ts)?; + let timestamp = message_ts + .map(|value| { + SlackTimestamp::parse(value) + .ok_or_else(|| Error::InvalidInput("Expected a Slack message timestamp".into())) + }) + .transpose()?; + if metadata.is_some_and(|value| !value.is_object()) { + return Err(Error::InvalidInput( + "Slack message metadata must be an object".into(), + )); + } + let mut payload = rendering::rich_payload_with_fallback_blocks(reply, fallback_blocks)?; + payload["channel"] = json!(destination.channel_id); + let method = if let Some(timestamp) = timestamp { + payload["ts"] = json!(timestamp); + // Slack otherwise retains attachments from the message being replaced. + payload["attachments"] = payload + .get("attachments") + .cloned() + .unwrap_or_else(|| json!([])); + MessageMethod::Update + } else { + if let Some(thread) = destination.thread_ts { + payload["thread_ts"] = json!(thread); + } + MessageMethod::Post + }; + if let Some(metadata) = metadata { + payload["metadata"] = metadata.clone(); + } + Ok(Self { method, payload }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn reply() -> RichReply { + RichReply { + text: "Ready".into(), + blocks: Vec::new(), + attachments: Vec::new(), + } + } + + #[test] + fn posts_to_the_validated_thread_with_safe_rendering() { + let metadata = json!({"event_type":"host_delivery","event_payload":{"id":"saved-1"}}); + let request = MessageRequest::new( + &reply(), + "C123", + Some("123.001"), + None, + Some(&metadata), + true, + ) + .unwrap(); + assert_eq!(request.method, MessageMethod::Post); + assert_eq!(request.payload["channel"], "C123"); + assert_eq!(request.payload["thread_ts"], "123.001"); + assert!(request.payload.get("ts").is_none()); + assert_eq!(request.payload["metadata"], metadata); + assert_eq!(request.payload["blocks"][0]["expand"], true); + assert_eq!(request.payload["unfurl_links"], false); + assert_eq!(request.payload["parse"], "none"); + assert_eq!(request.payload["link_names"], false); + } + + #[test] + fn updates_replace_old_attachments_and_do_not_move_the_message() { + let request = MessageRequest::new( + &reply(), + "D123", + Some("123.001"), + Some("123.002"), + None, + true, + ) + .unwrap(); + assert_eq!(request.method, MessageMethod::Update); + assert_eq!(request.payload["ts"], "123.002"); + assert!(request.payload.get("thread_ts").is_none()); + assert_eq!(request.payload["attachments"], json!([])); + assert!(!request.payload["blocks"].as_array().unwrap().is_empty()); + } + + #[test] + fn rejects_invalid_destinations_and_metadata_before_transport() { + for (channel, thread, message) in [ + ("other", None, None), + ("C123", Some("bad"), None), + ("C123", None, Some("bad")), + ] { + assert!(MessageRequest::new(&reply(), channel, thread, message, None, true).is_err()); + } + assert!(MessageRequest::new(&reply(), "C123", None, None, Some(&json!([])), true).is_err()); + } + + #[test] + fn preserves_explicit_empty_blocks_for_card_only_and_plain_updates() { + let request = + MessageRequest::new(&reply(), "C123", None, Some("123.002"), None, false).unwrap(); + assert_eq!(request.payload["blocks"], json!([])); + assert_eq!(request.payload["attachments"], json!([])); + } +} diff --git a/src/agentchat-slack/src/policy.rs b/src/agentchat-slack/src/policy.rs index 1cd96c9..a1cd655 100644 --- a/src/agentchat-slack/src/policy.rs +++ b/src/agentchat-slack/src/policy.rs @@ -1,5 +1,5 @@ //! Slack policies shared by native transports and foreign SDK adapters. -use super::{events, rendering, route::SlackTimestamp, SlackUser}; +use super::{events, route::SlackTimestamp, SlackUser}; use agentchat_core::{Error, HistoryLimit, Message, Result, Role, Sender}; use serde_json::{json, Value}; use std::collections::{HashSet, VecDeque}; @@ -162,39 +162,13 @@ impl ThreadHistory { .as_array() .ok_or_else(|| Error::InvalidInput("Slack did not return thread history".into()))?; for record in records { - let Some(ts) = record["ts"].as_str().and_then(SlackTimestamp::parse) else { + let Some(input) = events::history_record(record, self.bot_user_id.as_deref()) else { continue; }; - let Some(sender_id) = - events::nonempty(&record["user"]).or_else(|| events::nonempty(&record["bot_id"])) - else { - continue; - }; - let Some(text) = record.get("text").map_or(Some(""), Value::as_str) else { - continue; - }; - let files = events::attachments(&record["files"]); - if (events::trim_python(text).is_empty() && files.is_empty()) - || !ts.is_before(&self.current) - { + let ts = &input.timestamp; + if !ts.is_before(&self.current) { continue; } - let mirror = self - .bot_user_id - .as_deref() - .and_then(|bot| rendering::read_mirror(record, bot)); - let role = if Some(sender_id) == self.bot_user_id.as_deref() && mirror.is_none() { - Role::Assistant - } else { - Role::User - }; - let input = mirror.unwrap_or_else(|| rendering::MirroredInput { - sender: Sender { - id: sender_id.into(), - display_name: None, - }, - text: text.into(), - }); let item = Message { id: format!( "slack:{}:{}:{ts}", @@ -208,15 +182,16 @@ impl ThreadHistory { channel: self.source.channel.clone(), sender: input.sender, text: input.text, - role, + role: input.role, thread_id: Some(self.thread.clone()), addressed: true, - attachments: files, + attachments: input.files, metadata: json!({ "message_timestamp": ts.as_str(), "team_id": self.source.metadata.get("team_id"), "channel_id": self.source.metadata.get("channel_id"), "reply_thread_timestamp": self.thread, + "slack_reference": input.reference, }) .as_object() .cloned() diff --git a/src/agentchat-slack/src/rendering.rs b/src/agentchat-slack/src/rendering.rs index 6e84c12..e7f426c 100644 --- a/src/agentchat-slack/src/rendering.rs +++ b/src/agentchat-slack/src/rendering.rs @@ -72,12 +72,19 @@ fn expanded_blocks(blocks: &[Map]) -> Vec { } pub fn rich_payload(reply: &RichReply) -> Result { + rich_payload_with_fallback_blocks(reply, true) +} + +pub(crate) fn rich_payload_with_fallback_blocks( + reply: &RichReply, + fallback_blocks: bool, +) -> Result { if super::events::trim_python(&reply.text).is_empty() { return Err(Error::InvalidInput( "Rich replies require fallback text".into(), )); } - let blocks = if reply.blocks.is_empty() { + let blocks = if reply.blocks.is_empty() && fallback_blocks { text_blocks(&reply.text)? } else { expanded_blocks(&reply.blocks) diff --git a/src/agentchat-slack/tests/host.rs b/src/agentchat-slack/tests/host.rs new file mode 100644 index 0000000..86e711c --- /dev/null +++ b/src/agentchat-slack/tests/host.rs @@ -0,0 +1,108 @@ +use agentchat_slack::host::{ + ContextLimits, CoverageStatus, SlackHost, ThreadContextReader, ThreadOrigin, +}; +use serde_json::json; + +fn host() -> SlackHost { + SlackHost { + bot_user_id: "UBOT00001".into(), + workspace_id: Some("TTEST0001".into()), + exclude_prefixes: vec!["!aside".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(), + }, + ContextLimits { + max_pages, + max_chars, + max_message_chars: 8000, + }, + ) + .unwrap() +} +fn record(ts: &str, text: &str) -> serde_json::Value { + json!({"ts":ts,"user":"UAUTHOR01","text":text}) +} + +#[test] +fn parent_preview_survives_and_current_composer_text_is_not_repeated() { + let mut read = reader(5, 40000); + let root = json!({"ts":"1780000000.000001","user":"UAUTHOR01","text":"https://x.com/deepseek_ai/status/2111375640717033777?s=46 new deepseek","attachments":[{"text":"DeepSeek-V4.1-Flash-1017 is here!"}]}); + let current = json!({"ts":"1780000000.000010","user":"UAUTHOR01","text":"do we support this?","blocks":[{"type":"rich_text","text":"do we support this?"}],"attachments":[{"text":"current preview"}]}); + assert!(read.accept(&json!({"ok":true,"messages":[root,current],"has_more":false}))); + let context = read.finish(); + assert_eq!(context.status, CoverageStatus::Complete); + let text = context + .messages + .iter() + .map(|m| m.content.as_str()) + .collect::(); + assert!(text.contains("DeepSeek-V4.1-Flash-1017")); + assert!(text.contains("current preview")); + assert!(!text.contains("do we support this?")); +} +#[test] +fn prior_pages_are_kept_when_provider_fails() { + let mut read = reader(5, 40000); + assert!(!read.accept(&json!({"ok":true,"messages":[record("1780000000.000001","root")],"has_more":true,"response_metadata":{"next_cursor":"next"}}))); + assert_eq!(read.request()["cursor"], "next"); + assert!(read.accept(&json!({"ok":false,"error":"missing_scope"}))); + let context = read.finish(); + assert_eq!(context.status, CoverageStatus::Partial); + assert_eq!(context.messages.len(), 1); + assert_eq!(context.error.as_deref(), Some("missing_scope")); +} +#[test] +fn bounded_context_keeps_root_and_recent_and_reports_missing_root() { + let mut read = reader(1, 29); + assert!(read.accept(&json!({"ok":true,"messages":[record("1780000000.000001","root"),record("1780000000.000002","old"),record("1780000000.000003","new")],"has_more":true,"response_metadata":{"next_cursor":"next"}}))); + let context = read.finish(); + assert_eq!(context.status, CoverageStatus::Partial); + assert_eq!(context.messages.len(), 2); + assert!(context.messages[0].content.contains("root")); + assert!(context.messages[1].content.contains("new")); + let mut read = reader(1, 100); + read.accept(&json!({"ok":true,"messages":[],"has_more":false})); + assert_eq!(read.finish().status, CoverageStatus::Partial); +} +#[test] +fn routing_keeps_literal_mentions_and_excludes_only_prefix_boundaries() { + let host = host(); + for text in [ + "`<@UBOT00001>`", + "```<@UBOT00001>", + "> <@UBOT00001>", + ">>> <@UBOT00001>\nrest", + ] { + let route = host.routing_text(text); + assert_eq!(route.text, text); + assert!(!route.mentioned); + } + assert!(host.routing_text("<@UBOT00001> !ASIDE private").excluded); + assert!(!host.routing_text("!asideways").excluded); +} + +#[test] +fn current_event_reference_is_bound_and_survives_read_failure() { + let input = json!({"team_id":"TTEST0001","event":{"type":"app_mention","channel":"CTEST0001","ts":"1780000000.000010","thread_ts":"1780000000.000001","user":"UAUTHOR01","text":"<@UBOT00001> do we support this?","attachments":[{"text":"DeepSeek-V4.1-Flash-1017"}]}}); + let event = host().normalize_event(&input).unwrap().unwrap(); + let mut read = reader(5, 40000); + let mut wrong = event.reference.clone(); + wrong.origin.message_ts = "1780000000.000009".into(); + assert!(read.seed(wrong, &event.user_id).is_err()); + read.seed(event.reference, &event.user_id).unwrap(); + read.fail("missing_scope"); + let context = read.finish(); + assert_eq!(context.status, CoverageStatus::Partial); + assert!(context.messages[0] + .content + .contains("DeepSeek-V4.1-Flash-1017")); + assert!(!context.messages[0].content.contains("do we support this?")); +} diff --git a/src/bindings/python/src/lib.rs b/src/bindings/python/src/lib.rs index c62a74b..a6b50ce 100644 --- a/src/bindings/python/src/lib.rs +++ b/src/bindings/python/src/lib.rs @@ -23,6 +23,7 @@ mod providers { pub mod imessage; pub mod slack; pub mod slack_files; + pub mod slack_host; pub mod slack_policy; pub mod slack_socket; pub mod teams; @@ -183,6 +184,7 @@ fn _native(module: &Bound<'_, PyModule>) -> PyResult<()> { runtime::register(module)?; providers::slack::register(module)?; providers::slack_files::register(module)?; + providers::slack_host::register(module)?; providers::slack_policy::register(module)?; providers::discord::register(module)?; providers::imessage::register(module)?; diff --git a/src/bindings/python/src/providers/slack_host.rs b/src/bindings/python/src/providers/slack_host.rs new file mode 100644 index 0000000..953a910 --- /dev/null +++ b/src/bindings/python/src/providers/slack_host.rs @@ -0,0 +1,122 @@ +//! Host callbacks run on the caller's asyncio loop; Slack interpretation lives in the SDK. +use super::{slack_files::policy_error, transport::*}; +use crate::{convert, host}; +use agentchat_slack::host::{ + ContextLimits, CurrentReference, SlackHost as CoreHost, ThreadContextReader, ThreadOrigin, +}; +use pyo3::{ + exceptions::{PyTimeoutError, PyValueError}, + prelude::*, +}; +use serde_json::json; +use std::time::Instant; + +#[pyclass(frozen, module = "agentchat.channels.slack.host")] +struct SlackHost { + inner: CoreHost, +} +#[pymethods] +impl SlackHost { + #[new] + #[pyo3(signature=(*,bot_user_id,workspace_id=None,exclude_prefixes=Vec::new()))] + fn new( + bot_user_id: String, + workspace_id: Option, + exclude_prefixes: Vec, + ) -> Self { + Self { + inner: CoreHost { + bot_user_id, + workspace_id, + exclude_prefixes, + }, + } + } + fn routing_text(&self, py: Python<'_>, text: &str) -> PyResult> { + convert::from_json(py, &json!(self.inner.routing_text(text))) + } + fn normalize_event(&self, py: Python<'_>, payload: &Bound<'_, PyAny>) -> PyResult> { + let event = self + .inner + .normalize_event(&convert::to_json(payload)?) + .map_err(policy_error)?; + convert::from_json(py, &json!(event)) + } + // 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))] + fn thread_context( + &self, + py: Python<'_>, + origin: &Bound<'_, PyAny>, + request_page: Option>, + max_pages: usize, + max_chars: usize, + max_message_chars: usize, + 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(origin)?; + let origin: ThreadOrigin = serde_json::from_value(source.clone()) + .map_err(|_| PyValueError::new_err("Invalid Slack thread origin"))?; + let mut reader = ThreadContextReader::new( + self.inner.clone(), + origin, + ContextLimits { + max_pages, + max_chars, + max_message_chars, + }, + ) + .map_err(policy_error)?; + if let Some(reference) = source.get("reference") { + let reference: CurrentReference = serde_json::from_value(reference.clone()) + .map_err(|_| PyValueError::new_err("Invalid Slack current reference"))?; + let user_id = source["user_id"] + .as_str() + .ok_or_else(|| PyValueError::new_err("Expected Slack reference author"))?; + reader.seed(reference, user_id).map_err(policy_error)?; + } + host::coroutine(py, async move { + let Some(request_page) = request_page else { + reader.fail("history_not_requested"); + return j(json!(reader.finish())); + }; + let started = Instant::now(); + loop { + let remaining = timeout - started.elapsed().as_secs_f64(); + if remaining <= 0.0 { + reader.fail("timeout"); + break; + } + let pending = call(&request_page, "__call__", vec![j(reader.request())?], None)?; + let bounded = call( + &module("asyncio")?, + "wait_for", + vec![pending], + Some(j(json!({"timeout":remaining}))?), + )?; + let result = host::await_py(bounded).await; + let response = match result { + Ok(response) => response, + Err(error) + if Python::attach(|py| error.is_instance_of::(py)) => + { + reader.fail("timeout"); + break; + } + Err(error) => return Err(error), + }; + if reader.accept(&value(&response)?) { + break; + } + } + j(json!(reader.finish())) + }) + } +} +pub fn register(module: &Bound<'_, PyModule>) -> PyResult<()> { + module.add_class::() +} diff --git a/src/bindings/python/src/providers/slack_policy.rs b/src/bindings/python/src/providers/slack_policy.rs index b9444a1..0155c6e 100644 --- a/src/bindings/python/src/providers/slack_policy.rs +++ b/src/bindings/python/src/providers/slack_policy.rs @@ -65,6 +65,29 @@ fn slack_rich_payload(py: Python<'_>, content: &Bound<'_, PyAny>) -> PyResult, + reply: &Bound<'_, PyAny>, + channel_id: &str, + thread_ts: Option<&str>, + message_ts: Option<&str>, + metadata: Option<&Bound<'_, PyAny>>, + fallback_blocks: bool, +) -> PyResult> { + let metadata = metadata.map(convert::to_json).transpose()?; + let request = agentchat_slack::message_request::MessageRequest::new( + &convert::from_rich_reply(reply)?, + channel_id, + thread_ts, + message_ts, + metadata.as_ref(), + fallback_blocks, + ) + .map_err(policy_error)?; + convert::from_json(py, &serde_json::json!(request)) +} +#[pyfunction] #[pyo3(signature=(message,*,origin="web"))] fn slack_mirror_payload( py: Python<'_>, @@ -104,6 +127,7 @@ pub fn register(module: &Bound<'_, PyModule>) -> PyResult<()> { module.add_function(wrap_pyfunction!(slack_to_message, module)?)?; module.add_function(wrap_pyfunction!(slack_text_blocks, module)?)?; module.add_function(wrap_pyfunction!(slack_rich_payload, module)?)?; + module.add_function(wrap_pyfunction!(slack_message_request, module)?)?; module.add_function(wrap_pyfunction!(slack_mirror_payload, module)?)?; module.add_function(wrap_pyfunction!(slack_read_mirror, module)?)?; module @@ -121,6 +145,9 @@ pub fn register(module: &Bound<'_, PyModule>) -> PyResult<()> { module .getattr("slack_rich_payload")? .setattr("__module__", "agentchat.channels.slack.rendering")?; + module + .getattr("slack_message_request")? + .setattr("__module__", "agentchat.channels.slack.rendering")?; module .getattr("slack_mirror_payload")? .setattr("__module__", "agentchat.channels.slack.mirroring")?; diff --git a/tests/channels/slack/test_host.py b/tests/channels/slack/test_host.py new file mode 100644 index 0000000..c718974 --- /dev/null +++ b/tests/channels/slack/test_host.py @@ -0,0 +1,308 @@ +"""Public host API: use real native policies with a credential-free page callback.""" +import asyncio +import json + +import pytest + +from agentchat.channels.slack.events import to_message +from agentchat.channels.slack.host import SlackHost + +BOT = "UBOT00001" +ORIGIN = { + "team_id": "TTEST0001", "channel_id": "CTEST0001", + "thread_ts": "1780000000.000001", "message_ts": "1780000000.000010", +} +URL = "https://x.com/deepseek_ai/status/2111375640717033777?s=46" +QUESTION = "do we support this?" +HOST = SlackHost(bot_user_id=BOT, workspace_id=ORIGIN["team_id"], + exclude_prefixes=("!aside", "(aside)")) + + +def message(ts="1780000000.000001", text="new deepseek", **extra): + return {"ts": ts, "user": "UAUTHOR01", "text": text, **extra} + + +def page(messages, cursor=None): + return {"ok": True, "messages": messages, "has_more": cursor is not None, + "response_metadata": {"next_cursor": cursor or ""}} + + +def event(text=f"<@{BOT}> {QUESTION}", **extra): + return {"team_id": ORIGIN["team_id"], "event": { + "type": "message", "channel_type": "channel", "channel": ORIGIN["channel_id"], + "ts": ORIGIN["message_ts"], "thread_ts": ORIGIN["thread_ts"], + "user": "UAUTHOR01", "text": text, **extra, + }} + + +@pytest.mark.asyncio +async def test_parent_link_unfurl_and_current_extras_survive_without_repeating_question(): + payload = event() + normalized = HOST.normalize_event(payload) + assert normalized and normalized["text"] == QUESTION and normalized["mentioned"] + calls = [] + + async def request(params): + calls.append(params) + return page([ + message(text=f"{URL} new deepseek", attachments=[{ + "title": "DeepSeek (@deepseek_ai) on X", "title_link": URL, + "text": "DeepSeek-V4.1-Flash-1017 is here!", "image_url": "https://example.com/chart.png", + }]), + message(ORIGIN["message_ts"], f"<@{BOT}> {QUESTION}", files=[{"id": "FFILE0001"}], + blocks=[{"type": "rich_text", "elements": [{"text": QUESTION}]}], + attachments=[{"text": "current link preview"}]), + ]) + + context = await HOST.thread_context(normalized, request) + references = "\n".join(item["content"] for item in context["messages"]) + assert context["status"] == "complete" + assert URL in references and "DeepSeek-V4.1-Flash-1017" in references + assert "current link preview" in references + assert QUESTION not in references and "FFILE0001" not in references + assert context["messages"][0]["author"] == "UAUTHOR01" + assert calls == [{"channel": ORIGIN["channel_id"], "ts": ORIGIN["thread_ts"], + "latest": ORIGIN["message_ts"], "inclusive": True, "limit": 100, + "cursor": "", "include_all_metadata": True}] + + +@pytest.mark.parametrize("quoted", [ + f"`<@{BOT}>`", f"```\n<@{BOT}>\n```", f"> <@{BOT}>", f">>> <@{BOT}>\nrest", + f"`<@{BOT}>", f"```<@{BOT}>", +]) +def test_quoted_mentions_are_literal_in_host_and_existing_event_api(quoted): + payload = event(quoted) + normalized = HOST.normalize_event(payload) + legacy = to_message(payload, payload["event"], bot_user_id=BOT, + workspace_id=ORIGIN["team_id"]) + assert normalized and normalized["text"] == quoted and not normalized["mentioned"] + assert legacy and legacy.text == quoted and not legacy.addressed + routed = HOST.routing_text(f"<@{BOT}> explain\n{quoted}") + assert routed == {"text": f"explain\n{quoted}", "mentioned": True, "excluded": False} + + +@pytest.mark.parametrize("updates", [ + {"text": 1}, {"channel": "not-a-channel"}, {"ts": "NaN"}, + {"channel_type": {}}, {"subtype": {}}, + {"thread_ts": "1780000001.000001"}, +]) +def test_invalid_supported_event_is_rejected(updates): + with pytest.raises(ValueError): + HOST.normalize_event(event(**updates)) + + +def test_ignored_events_and_explicit_dm_routing(): + assert HOST.normalize_event(event("!ASIDE private thought")) is None + assert HOST.normalize_event(event(subtype="message_changed")) is None + assert HOST.normalize_event(event(bot_id="BBOT00001")) is None + assert HOST.normalize_event(event(bot_id=None, bot_profile=None)) is not None + assert HOST.normalize_event(event(channel="DTEST0001", channel_type=None)) is None + direct = HOST.normalize_event(event(channel="DTEST0001", channel_type="im")) + assert direct and direct["direct"] + assert not HOST.routing_text("!asideways")['excluded'] + + +@pytest.mark.asyncio +async def test_paginated_history_excludes_future_other_threads_asides_and_duplicates(): + calls = [] + + async def request(params): + calls.append(params["cursor"]) + if not params["cursor"]: + return page([message(), message("1780000000.000002", "prior"), + message("1780000000.000003", "!aside hidden"), + message("1780000000.000004", "foreign", thread_ts="1780000000.000009"), + message("1780000000.000011", "future")], "next") + return page([message(), message("1780000000.000005", "recent")]) + + context = await HOST.thread_context(ORIGIN, request) + assert calls == ["", "next"] + assert context["status"] == "complete" + assert [json.loads(item["content"])["text"] for item in context["messages"]] == [ + "new deepseek", "prior", "recent", + ] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("first_page", [True, False]) +async def test_provider_failure_retains_good_pages_and_reports_coverage(first_page): + calls = 0 + + async def request(_): + nonlocal calls + calls += 1 + if not first_page and calls == 1: + return page([message()], "next") + return {"ok": False, "error": "missing_scope"} + + context = await HOST.thread_context(ORIGIN, request) + assert context["status"] == ("unavailable" if first_page else "partial") + assert context["error"] == "missing_scope" + assert len(context["messages"]) == (0 if first_page else 1) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("origin", [ORIGIN, {**ORIGIN, "message_ts": ORIGIN["thread_ts"]}]) +async def test_missing_root_is_never_complete(origin): + async def request(_): + return page([]) + + context = await HOST.thread_context(origin, request) + assert context["status"] == "partial" and "root" in context["warning"] + + +@pytest.mark.asyncio +async def test_timeout_cancels_callback_and_authorization_errors_propagate(): + cancelled = asyncio.Event() + + async def slow(_): + try: + await asyncio.Event().wait() + finally: + cancelled.set() + + context = await HOST.thread_context(ORIGIN, slow, timeout=0.01) + assert context["status"] == "unavailable" and context["error"] == "timeout" + assert cancelled.is_set() + + async def denied(_): + raise PermissionError("authorization revoked") + + with pytest.raises(PermissionError, match="authorization revoked"): + await HOST.thread_context(ORIGIN, denied) + + +@pytest.mark.asyncio +async def test_page_and_character_limits_keep_root_and_recent_context(): + calls = [] + + async def request(params): + calls.append(params) + return page([message(text="root"), message("1780000000.000002", "old"), + message("1780000000.000003", "new")], "more") + + context = await HOST.thread_context(ORIGIN, request, max_pages=1, max_chars=29) + assert len(calls) == 1 and context["status"] == "partial" + assert [json.loads(item["content"])["text"] for item in context["messages"]] == ["root", "new"] + assert sum(len(item["content"]) for item in context["messages"]) <= 29 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("response, status, reason", [ + ({"ok": True, "messages": "bad"}, "unavailable", "invalid_response"), + ({"ok": True, "messages": [message()]}, "partial", "pagination"), + ({**page([message()], ""), "has_more": True}, "partial", "cursor"), +]) +async def test_malformed_page_coverage_is_explicit(response, status, reason): + async def request(_): + return response + + context = await HOST.thread_context(ORIGIN, request) + assert context["status"] == status + assert reason in context.get("error", context.get("warning", "")) + + +@pytest.mark.asyncio +async def test_timeout_bounds_the_whole_read_and_keeps_first_page(): + calls = 0 + + async def request(_): + nonlocal calls + calls += 1 + if calls == 1: + return page([message()], "next") + await asyncio.Event().wait() + + context = await HOST.thread_context(ORIGIN, request, timeout=0.01) + assert calls == 2 and context["status"] == "partial" + assert context["error"] == "timeout" and len(context["messages"]) == 1 + + +@pytest.mark.asyncio +async def test_individual_reference_truncation_is_visible_and_unicode_bounded(): + async def request(_): + return page([message(text="🗿" * 100)]) + + context = await HOST.thread_context(ORIGIN, request, max_message_chars=20) + assert context["status"] == "partial" + assert context["messages"][0]["truncated"] + assert len(context["messages"][0]["content"]) == 20 + + +def test_existing_event_api_preserves_unfurls_and_noncomposer_blocks(): + payload = event( + attachments=[{"title_link": URL, "text": "DeepSeek-V4.1-Flash-1017"}], + blocks=[{"type": "rich_text", "text": QUESTION}, + {"type": "image", "image_url": "https://example.com/chart.png"}], + ) + converted = to_message(payload, payload["event"], bot_user_id=BOT, + workspace_id=ORIGIN["team_id"]) + assert converted is not None + reference = converted.metadata["slack_reference"] + assert reference["attachments"][0]["title_link"] == URL + assert reference["blocks"] == [{"type": "image", "image_url": "https://example.com/chart.png"}] + assert converted.text == QUESTION + + +@pytest.mark.asyncio +async def test_current_event_previews_survive_without_history_and_on_provider_failure(): + source = HOST.normalize_event(event( + channel="DTEST0001", channel_type="im", + attachments=[{"title_link": URL, "text": "DeepSeek-V4.1-Flash-1017"}], + )) + assert source is not None + local = await HOST.thread_context(source) + assert local["status"] == "partial" and local["error"] == "history_not_requested" + assert local["messages"][0]["author"] == source["user_id"] + assert URL in local["messages"][0]["content"] + assert QUESTION not in local["messages"][0]["content"] + + async def denied(_): + return {"ok": False, "error": "missing_scope"} + + failed = await HOST.thread_context(source, denied) + assert failed["status"] == "partial" and failed["error"] == "missing_scope" + assert failed["messages"] == local["messages"] + empty = await HOST.thread_context(ORIGIN) + assert empty["status"] == "unavailable" and empty["error"] == "history_not_requested" + + +@pytest.mark.asyncio +async def test_seeded_and_fetched_current_previews_are_merged_once(): + preview = {"title_link": URL, "text": "DeepSeek-V4.1-Flash-1017"} + source = HOST.normalize_event(event(attachments=[preview])) + assert source is not None + + async def request(_): + return page([message(), message(ORIGIN["message_ts"], QUESTION, + attachments=[preview, {"text": "new preview"}])]) + + context = await HOST.thread_context(source, request) + assert context["status"] == "complete" + current = context["messages"][-1] + assert current["ts"] == ORIGIN["message_ts"] + assert current["content"].count("DeepSeek-V4.1-Flash-1017") == 1 + assert "new preview" in current["content"] + assert QUESTION not in current["content"] + assert len(context["messages"]) == 2 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("field, changed", [ + ("team_id", "TOTHER001"), ("channel_id", "COTHER001"), + ("thread_ts", "1780000000.000002"), ("message_ts", "1780000000.000009"), + ("user_id", "UOTHER001"), +]) +async def test_reference_from_another_event_is_rejected_before_callback(field, changed): + source = HOST.normalize_event(event(attachments=[{"text": "private preview"}])) + assert source is not None + source["reference"][field] = changed + calls = [] + + async def request(params): + calls.append(params) + return page([]) + + with pytest.raises(ValueError, match="different event"): + await HOST.thread_context(source, request) + assert not calls diff --git a/tests/channels/slack/test_message_request.py b/tests/channels/slack/test_message_request.py new file mode 100644 index 0000000..7a9540e --- /dev/null +++ b/tests/channels/slack/test_message_request.py @@ -0,0 +1,58 @@ +import pytest + +from agentchat.channels.slack.rendering import message_request +from agentchat.models import RichReply + + +def test_request_routes_a_rendered_reply_without_sending(): + metadata = {"event_type": "saved_delivery", "event_payload": {"id": "one"}} + request = message_request( + RichReply("Ready"), channel_id="C123", thread_ts="123.001", metadata=metadata, + ) + assert request["method"] == "chat.postMessage" + payload = request["payload"] + assert payload["channel"] == "C123" and payload["thread_ts"] == "123.001" + assert payload["metadata"] == metadata + assert payload["blocks"][0]["expand"] is True + assert payload["parse"] == "none" and payload["link_names"] is False + + +def test_update_clears_old_attachments_and_retains_product_controls(): + control = {"type": "actions", "elements": []} + request = message_request( + RichReply("Complete", blocks=(control,)), + channel_id="C123", thread_ts="123.001", message_ts="123.002", + ) + assert request["method"] == "chat.update" + assert request["payload"]["ts"] == "123.002" + assert "thread_ts" not in request["payload"] + assert request["payload"]["attachments"] == [] + assert request["payload"]["blocks"] == [control] + + +def test_plain_update_clears_blocks_without_adding_fallback_sections(): + request = message_request( + RichReply("Complete"), channel_id="C123", message_ts="123.002", fallback_blocks=False, + ) + assert request["payload"]["blocks"] == [] + assert request["payload"]["attachments"] == [] + + +def test_card_only_reply_does_not_repeat_its_accessible_fallback(): + card = {"fallback": "View PR", "text": "Changes are ready"} + request = message_request( + RichReply("View PR", attachments=(card,)), channel_id="C123", fallback_blocks=False, + ) + assert request["payload"]["blocks"] == [] + assert request["payload"]["attachments"] == [card] + + +@pytest.mark.parametrize("kwargs", [ + {"channel_id": "invalid"}, + {"channel_id": "C123", "thread_ts": "invalid"}, + {"channel_id": "C123", "message_ts": "invalid"}, + {"channel_id": "C123", "metadata": []}, +]) +def test_request_rejects_invalid_provider_inputs(kwargs): + with pytest.raises(ValueError): + message_request(RichReply("Ready"), **kwargs)