feat(acp): ACP server over WebSocket (revives #1260)

.github/workflows/ci.yml

@@ -50,9 +50,14 @@ jobs:
         run: cargo clippy --workspace --features unified -- -D warnings
       - name: cargo test
         run: cargo test --workspace
-
-  # gateway tests are now covered by `cargo test --workspace` in the check job above
-  # (openab-gateway is a workspace member in crates/openab-gateway/)
+      # `cargo test --workspace` builds with default features, so openab-gateway's `acp`-gated
+      # conformance/handler/streaming tests never compile or run. Exercise them explicitly, and
+      # build the unified binary so the embedded /acp endpoint is compiled in CI. Scoped to
+      # `-p openab-gateway` to avoid the workspace hooks::tests parallel flake.
+      - name: cargo test (acp gateway)
+        run: cargo test -p openab-gateway --features acp
+      - name: cargo build (unified)
+        run: cargo build --features unified
 
   operator:
     needs: changes

Cargo.toml

@@ -28,7 +28,7 @@ serenity = { version = "0.12", default-features = false, features = ["client", "
 default = ["discord", "slack", "secrets-aws", "agentcore", "config-s3", "pre-seed", "filestore"]
 
 # Opt-in: compile all gateway adapters into a single unified binary
-unified = ["telegram", "line", "feishu", "googlechat", "wecom", "teams"]
+unified = ["telegram", "line", "feishu", "googlechat", "wecom", "teams", "acp"]
 
 # Core adapters
 discord = ["dep:serenity", "openab-core/discord"]
@@ -48,6 +48,7 @@ feishu = ["dep:openab-gateway", "dep:axum", "openab-gateway/feishu"]
 googlechat = ["dep:openab-gateway", "dep:axum", "openab-gateway/googlechat"]
 wecom = ["dep:openab-gateway", "dep:axum", "openab-gateway/wecom"]
 teams = ["dep:openab-gateway", "dep:axum", "openab-gateway/teams"]
+acp = ["dep:openab-gateway", "dep:axum", "openab-gateway/acp"]
 
 [dev-dependencies]
 tempfile = "3.27.0"

crates/openab-core/Cargo.toml

@@ -26,6 +26,7 @@ bytes = "1"
 base64 = "0.22"
 image = { version = "0.25", default-features = false, features = ["jpeg", "png", "gif", "webp"] }
 unicode-width = "0.2"
+unicode-segmentation = "1"
 pulldown-cmark = { version = "0.13", default-features = false }
 tokio-tungstenite = { version = "0.21", features = ["rustls-tls-webpki-roots"] }
 rustls = { version = "0.22", optional = true }

crates/openab-core/src/adapter.rs

@@ -21,6 +21,20 @@ pub struct OutputDirectives {
     pub reply_to: Option<String>,
 }
 
+/// Chunk limit for delivering a reply on `platform`. ACP is a WebSocket transport with
+/// no small per-message limit, and its reply route is closed after the first delivered
+/// message — so splitting a long reply into multiple messages truncates it over ACP
+/// (review F2). ACP therefore delivers whole (`usize::MAX` → a single chunk); every
+/// other platform keeps the adapter's chunk limit. Overflow-safe: the only arithmetic on
+/// the result is `saturating_sub` (mention-footer reserve).
+fn reply_message_limit(platform: &str, adapter_limit: usize) -> usize {
+    if platform == "acp" {
+        usize::MAX
+    } else {
+        adapter_limit
+    }
+}
+
 /// Parse `[[key:value]]` directives from the beginning of agent output.
 /// Returns parsed directives and the remaining content (directives stripped).
 pub fn parse_output_directives(content: &str) -> (OutputDirectives, String) {
@@ -686,8 +700,16 @@ impl AdapterRouter {
     ) -> Result<()> {
         let adapter = adapter.clone();
         let thread_channel = thread_channel.clone();
-        let message_limit = adapter.message_limit();
-        let streaming = adapter.use_streaming(other_bot_present);
+        let message_limit = reply_message_limit(&thread_channel.platform, adapter.message_limit());
+        // ACP must not inherit the unified adapter's Telegram streaming flag (wrong
+        // coupling): it streams append-only `agent_message_chunk` deltas built from the
+        // post+edit (`edit_message` snapshot) path, i.e. streaming=false. Decide it
+        // explicitly by platform rather than by whatever Telegram happens to be set to.
+        let streaming = if thread_channel.platform == "acp" {
+            false
+        } else {
+            adapter.use_streaming(other_bot_present)
+        };
         // Keep the full turn text (incl. inter-tool narration) when streaming
         // (it was already shown live) OR when `[reactions] narration_display` is
         // set. Otherwise a send-once turn delivers only the final answer block.
@@ -706,6 +728,10 @@ impl AdapterRouter {
             self.table_mode
         };
         let tool_display = self.reactions_config.tool_display;
+        // ACP streams over an append-only `agent_message_chunk`; a re-rendered tool-status
+        // prefix from `compose_display` would corrupt the deltas, so ACP streams the raw
+        // append-only answer text and surfaces tools separately (review F2 / roadmap).
+        let platform_is_acp = thread_channel.platform == "acp";
         let prompt_hard_timeout = self.prompt_hard_timeout;
         let liveness_check_interval = self.liveness_check_interval;
 
@@ -933,7 +959,8 @@ impl AdapterRouter {
                                             }
                                         }
                                     } else if let Some(tx) = &buf_tx {
-                                        let _ = tx.send(compose_display(
+                                        let _ = tx.send(display_for(
+                                            platform_is_acp,
                                             &tool_lines,
                                             &text_buf,
                                             true,
@@ -982,7 +1009,8 @@ impl AdapterRouter {
                                     }
                                     // Post+edit live update (no-op under native streaming: buf_tx is None).
                                     if let Some(tx) = &buf_tx {
-                                        let _ = tx.send(compose_display(
+                                        let _ = tx.send(display_for(
+                                            platform_is_acp,
                                             &tool_lines,
                                             &text_buf,
                                             true,
@@ -1027,7 +1055,8 @@ impl AdapterRouter {
                                         });
                                     }
                                     if let Some(tx) = &buf_tx {
-                                        let _ = tx.send(compose_display(
+                                        let _ = tx.send(display_for(
+                                            platform_is_acp,
                                             &tool_lines,
                                             &text_buf,
                                             true,
@@ -1086,7 +1115,7 @@ impl AdapterRouter {
 
                     // Build final content
                     let final_content =
-                        compose_display(&tool_lines, &text_buf, false, tool_display);
+                        display_for(platform_is_acp, &tool_lines, &text_buf, false, tool_display);
                     let final_content = if final_content.is_empty() {
                         if turn_result.is_silent_failure() {
                             warn!(
@@ -1474,6 +1503,25 @@ pub(crate) fn classify_empty_turn(
     }
 }
 
+/// Content to stream/deliver for a reply. ACP gets the raw append-only answer `text`
+/// (its `agent_message_chunk` stream is append-only, so a re-rendered `compose_display`
+/// tool-status prefix would corrupt the deltas — review F2); tool activity is surfaced
+/// separately as structured `tool_call` updates (roadmap). Every other platform gets the
+/// tool-merged display.
+fn display_for(
+    platform_is_acp: bool,
+    tool_lines: &[ToolEntry],
+    text: &str,
+    streaming: bool,
+    tool_display: ToolDisplay,
+) -> String {
+    if platform_is_acp {
+        text.to_string()
+    } else {
+        compose_display(tool_lines, text, streaming, tool_display)
+    }
+}
+
 fn compose_display(
     tool_lines: &[ToolEntry],
     text: &str,
@@ -1686,6 +1734,18 @@ fn propagate_mentions_to_chunks(
 mod tests {
     use super::*;
 
+    #[test]
+    fn acp_reply_limit_is_unbounded_others_use_adapter_limit() {
+        // ACP delivers whole (no chunking → no truncation, review F2); other platforms
+        // keep the adapter's limit.
+        assert_eq!(reply_message_limit("acp", 4096), usize::MAX);
+        assert_eq!(reply_message_limit("discord", 2000), 2000);
+        assert_eq!(reply_message_limit("slack", 4096), 4096);
+        // and a long reply under the ACP limit is a single chunk (delivered whole)
+        let long = "x".repeat(50_000);
+        assert_eq!(crate::format::split_message(&long, reply_message_limit("acp", 4096)).len(), 1);
+    }
+
     #[test]
     fn select_delivery_text_send_once_keeps_only_final_block() {
         // Simulates: narration "n1" → tool (answer_start→2) → narration "n2"