fix(telegram): use real forum topic id in unified create_thread

.github/workflows/build-unified-binary.yml

@@ -0,0 +1,60 @@
+# Build a unified openab binary for local install (e.g. arm64 host) without
+# compiling on the developer machine. Triggered on this branch and manually.
+
+name: Build Unified Binary (PR)
+
+on:
+  push:
+    branches:
+      - fix/telegram-unified-create-thread
+  workflow_dispatch:
+  pull_request:
+    paths:
+      - "src/**"
+      - "crates/**"
+      - "Cargo.toml"
+      - "Cargo.lock"
+      - ".github/workflows/build-unified-binary.yml"
+
+jobs:
+  build-unified:
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+          - { target: aarch64-unknown-linux-gnu, runner: ubuntu-24.04-arm, os: linux, arch: arm64 }
+          - { target: x86_64-unknown-linux-gnu, runner: ubuntu-latest, os: linux, arch: x64 }
+    runs-on: ${{ matrix.runner }}
+    permissions:
+      contents: read
+    steps:
+      - uses: actions/checkout@v6
+
+      - name: Install Rust
+        uses: dtolnay/rust-toolchain@stable
+        with:
+          targets: ${{ matrix.target }}
+
+      - uses: Swatinem/rust-cache@v2
+        with:
+          key: unified-${{ matrix.target }}
+
+      - name: Build unified release binary
+        run: cargo build --release --features unified --target ${{ matrix.target }}
+
+      - name: Package
+        shell: bash
+        run: |
+          NAME="openab-unified-${{ matrix.os }}-${{ matrix.arch }}"
+          mkdir -p dist
+          cp "target/${{ matrix.target }}/release/openab" "dist/openab-unified"
+          cp config.toml.example dist/ 2>/dev/null || true
+          tar -C dist -czf "${NAME}.tar.gz" .
+          ls -la "${NAME}.tar.gz"
+
+      - name: Upload artifact
+        uses: actions/upload-artifact@v4
+        with:
+          name: openab-unified-${{ matrix.os }}-${{ matrix.arch }}
+          path: openab-unified-*.tar.gz
+          retention-days: 14

Cargo.toml

@@ -25,6 +25,7 @@ serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 async-trait = "0.1"
 serenity = { version = "0.12", default-features = false, features = ["client", "gateway", "model", "rustls_backend", "cache"], optional = true }
+uuid = { version = "1", features = ["v4"] }
 
 [features]
 # Default: core only (Discord + Slack). Gateway ships as separate binary.

crates/openab-gateway/src/lib.rs

@@ -95,6 +95,16 @@ pub struct AppState {
     /// Optional pre-download identity probe (see [`IngressTrustProbe`]).
     pub trust_probe: Option<IngressTrustProbe>,
     pub client: reqwest::Client,
+    /// In-process oneshot waiters for gateway command responses
+    /// (`create_topic`, etc.). Used by the unified binary so `create_thread`
+    /// can await the real Telegram `message_thread_id` instead of inventing
+    /// one. The standalone gateway WS path keeps its own `pending` map on
+    /// `GatewayAdapter` and does not use this field.
+    pub pending_commands: Arc<
+        tokio::sync::Mutex<
+            HashMap<String, tokio::sync::oneshot::Sender<crate::schema::GatewayResponse>>,
+        >,
+    >,
 }
 
 
@@ -145,6 +155,7 @@ impl AppState {
         lineworks_ingress_queue: Arc::new(Semaphore::new(LINEWORKS_INGRESS_QUEUE_MAX)),
         trust_probe: None,
             client: reqwest::Client::new(),
+            pending_commands: Arc::new(tokio::sync::Mutex::new(HashMap::new())),
         }
     }
 
@@ -267,6 +278,7 @@ impl AppState {
         lineworks_ingress_queue: Arc::new(Semaphore::new(LINEWORKS_INGRESS_QUEUE_MAX)),
         trust_probe: None,
             client,
+            pending_commands: Arc::new(tokio::sync::Mutex::new(HashMap::new())),
         }
     }
 
@@ -849,6 +861,7 @@ pub async fn serve(config: ServeConfig) -> anyhow::Result<()> {
         lineworks_ingress_queue: Arc::new(Semaphore::new(LINEWORKS_INGRESS_QUEUE_MAX)),
         trust_probe: None,
         client,
+        pending_commands: Arc::new(tokio::sync::Mutex::new(HashMap::new())),
     });
 
     // Phase 1 L1 audit (#1356): warn if any active webhook platform has no

src/main.rs

@@ -1481,9 +1481,12 @@ async fn main() -> anyhow::Result<()> {
             let app = app.with_state(gw_state.clone());
 
             // Bridge task: receive events from adapters via event_tx, dispatch to core
-            let unified_adapter: Arc<dyn adapter::ChatAdapter> = Arc::new(
-                unified_adapter::UnifiedGatewayAdapter::new(gw_state.clone()),
-            );
+            // Keep a typed Arc so create_topic responses can complete oneshot waiters
+            // on UnifiedGatewayAdapter before ordinary GatewayEvent processing.
+            let unified_adapter_impl = Arc::new(unified_adapter::UnifiedGatewayAdapter::new(
+                gw_state.clone(),
+            ));
+            let unified_adapter: Arc<dyn adapter::ChatAdapter> = unified_adapter_impl.clone();
 
             // Bot gating still reads env here (structural, not L2/L3):
             // channel/user gating moved to the shared trust registry, seeded
@@ -1515,15 +1518,22 @@ async fn main() -> anyhow::Result<()> {
                 filestore: filestore.clone(),
             });
 
-            // Spawn the event bridge (event_tx → process_gateway_event)
+            // Spawn the event bridge (event_tx → process_gateway_event).
+            // Demux GatewayResponse first so create_topic waiters receive the
+            // real message_thread_id and responses are not mis-parsed as events.
             let mut event_rx = event_tx.subscribe();
             let bridge_ctx = event_ctx.clone();
+            let bridge_adapter = unified_adapter_impl.clone();
             tokio::spawn(async move {
                 loop {
                     match event_rx.recv().await {
                         Ok(event_json) => {
                             let ctx = bridge_ctx.clone();
+                            let adapter = bridge_adapter.clone();
                             tokio::spawn(async move {
+                                if adapter.try_complete_pending(&event_json).await {
+                                    return;
+                                }
                                 if let Err(e) = process_gateway_event(&event_json, &ctx).await {
                                     tracing::warn!(error = %e, "unified bridge: event processing failed");
                                 }

src/unified_adapter.rs

@@ -1,14 +1,19 @@
 //! UnifiedGatewayAdapter — routes ChatAdapter calls through in-process gateway
 //! platform adapters based on the ChannelRef.platform field.
 
-use anyhow::Result;
+use anyhow::{anyhow, Result};
 use async_trait::async_trait;
 use openab_core::adapter::{ChannelRef, ChatAdapter, MessageRef};
-use openab_gateway::schema::{Content, GatewayReply, ReplyChannel};
+use openab_gateway::schema::{Content, GatewayReply, GatewayResponse, ReplyChannel};
 use openab_gateway::AppState;
 use std::collections::HashMap;
 use std::sync::Arc;
 use tokio::sync::Mutex;
+use tracing::warn;
+use uuid::Uuid;
+
+/// How long `create_thread` waits for the platform `create_topic` response.
+const CREATE_TOPIC_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
 
 pub struct UnifiedGatewayAdapter {
     pub gw_state: Arc<AppState>,
@@ -24,6 +29,34 @@ impl UnifiedGatewayAdapter {
         }
     }
 
+    /// If `json` is a gateway command response (`openab.gateway.response.v1`),
+    /// deliver it to any waiter registered via [`Self::create_thread`] and
+    /// return `true`. Returns `false` for ordinary inbound events so the
+    /// caller can keep processing them as `GatewayEvent`s.
+    ///
+    /// Without this demux, `createForumTopic` responses are fed into
+    /// `process_gateway_event`, which fails with
+    /// `invalid type: null, expected a string` (response fields do not match
+    /// the event schema) and the real `message_thread_id` is discarded.
+    pub async fn try_complete_pending(&self, json: &str) -> bool {
+        let Ok(resp) = serde_json::from_str::<GatewayResponse>(json) else {
+            return false;
+        };
+        if resp.schema != "openab.gateway.response.v1" {
+            return false;
+        }
+        if let Some(tx) = self
+            .gw_state
+            .pending_commands
+            .lock()
+            .await
+            .remove(&resp.request_id)
+        {
+            let _ = tx.send(resp);
+        }
+        true
+    }
+
     /// Dispatch a GatewayReply to the correct platform adapter.
     async fn dispatch_reply(&self, reply: &GatewayReply) {
         let client = &self.gw_state.client;
@@ -168,19 +201,69 @@ impl ChatAdapter for UnifiedGatewayAdapter {
     async fn create_thread(
         &self,
         channel: &ChannelRef,
-        trigger_msg: &MessageRef,
+        _trigger_msg: &MessageRef,
         title: &str,
     ) -> Result<ChannelRef> {
-        let reply = self.build_reply(channel, title, Some("create_topic"), None);
+        // Mirror the standalone GatewayAdapter path: register a oneshot, send
+        // create_topic with request_id, wait for GatewayResponse carrying the
+        // real platform thread id (Telegram message_thread_id).
+        //
+        // The previous implementation ignored the response and returned
+        // trigger_msg.message_id as thread_id. For Telegram forums that value
+        // is a *message* id, not a topic id, so replies fail with
+        // "Bad Request: message thread not found" after a topic was created.
+        let req_id = format!("req_{}", Uuid::new_v4());
+        let (tx, rx) = tokio::sync::oneshot::channel();
+        self.gw_state
+            .pending_commands
+            .lock()
+            .await
+            .insert(req_id.clone(), tx);
+
+        let mut reply = self.build_reply(channel, title, Some("create_topic"), None);
+        reply.request_id = Some(req_id.clone());
         self.dispatch_reply(&reply).await;
-        // Return a thread channel ref with the trigger message as thread_id
-        Ok(ChannelRef {
-            platform: channel.platform.clone(),
-            channel_id: channel.channel_id.clone(),
-            thread_id: Some(trigger_msg.message_id.clone()),
-            parent_id: Some(channel.channel_id.clone()),
-            origin_event_id: channel.origin_event_id.clone(),
-        })
+
+        match tokio::time::timeout(CREATE_TOPIC_TIMEOUT, rx).await {
+            Ok(Ok(resp)) if resp.success => {
+                if let Some(thread_id) = resp.thread_id {
+                    Ok(ChannelRef {
+                        platform: channel.platform.clone(),
+                        channel_id: channel.channel_id.clone(),
+                        thread_id: Some(thread_id),
+                        parent_id: Some(channel.channel_id.clone()),
+                        origin_event_id: channel.origin_event_id.clone(),
+                    })
+                } else {
+                    warn!(
+                        request_id = %req_id,
+                        "create_topic succeeded but thread_id missing; falling back to parent channel"
+                    );
+                    Ok(channel.clone())
+                }
+            }
+            Ok(Ok(resp)) => {
+                warn!(
+                    err = ?resp.error,
+                    request_id = %req_id,
+                    "create_topic failed, falling back to same channel"
+                );
+                Ok(channel.clone())
+            }
+            Ok(Err(_)) => {
+                // oneshot dropped — treat as failure
+                self.gw_state.pending_commands.lock().await.remove(&req_id);
+                Err(anyhow!("create_topic response channel closed"))
+            }
+            Err(_) => {
+                warn!(
+                    request_id = %req_id,
+                    "create_topic timeout, falling back to same channel"
+                );
+                self.gw_state.pending_commands.lock().await.remove(&req_id);
+                Ok(channel.clone())
+            }
+        }
     }
 
     async fn add_reaction(&self, msg: &MessageRef, emoji: &str) -> Result<()> {