feat: Telegram Channel

backend/requirements.txt

@@ -1,6 +1,7 @@
 Quart==0.20.0
 quart-auth
 quart-cors
+httpx
 sqlalchemy
 nest_asyncio
 replicate

backend/src/app.py

@@ -22,6 +22,7 @@ from drives import init_drives
 from remote import RemoteConnection, RemoteToolSpec, parse_function_specs
 
 from app_mixins import ChatMixin, AgentMixin
+from channels import Channels
 
 class App(ChatMixin, AgentMixin):
     def __init__(self, data_directory, replicate_token=None, openrouter_token=None):
@@ -34,6 +35,9 @@ class App(ChatMixin, AgentMixin):
         # active websocket tasks, for manual cancelling upon server shutdown
         self.ws_tasks = []
 
+        # Channel bindings, in-memory transcripts, pollers and HTTP.
+        self.channels = Channels(self)
+
         self.tmp = tmp.State()
 
         # Clients
@@ -236,13 +240,16 @@ async def main():
         openrouter_token=os.getenv("OPENROUTER_API_KEY")
     )
 
-    # 3. Start Server
+    # 3. Set up channel listeners (webhook reconciliation or long polling)
+    await horde_app.channels.setup()
+
+    # 4. Start Server
     # manually trigger server shutdown, avoids hypercorn installing signal handlers
     shutdown_event = asyncio.Event()
     # TODO: get port from env or don't pass it (actually server should handle none)
     server_task = asyncio.create_task(start_server(horde_app, port, shutdown_event))
 
-    # 4. Graceful Termination
+    # 5. Graceful Termination
     loop = asyncio.get_running_loop()
     def interrupt():
         print("Interrupt Signal")
@@ -250,6 +257,8 @@ async def main():
         # Unblock websocket handlers parked in websocket.receive()
         for task in list(horde_app.ws_tasks):
             task.cancel()
+        # Stop channel pollers and any in-flight channel runs
+        horde_app.channels.shutdown()
 
     for sig in (signal.SIGINT, signal.SIGTERM):
         loop.add_signal_handler(sig, interrupt)

backend/src/channels.py

@@ -0,0 +1,397 @@
+"""Channel integration, currently Telegram only.
+
+A *channel* is an inbound/outbound pipe between an external messaging service
+and an agent. The binding is stored as an ordinary ``AgentResource`` on the
+responding agent::
+
+    kind  = "channel"
+    key   = "telegram"
+    value = {"token": "<bot token>", "secret": "<webhook secret>", "name": "..."}
+
+Nothing else is persisted: a Telegram conversation never creates ``Chat`` /
+``Thread`` / ``Message`` rows. The rolling transcript lives in memory here
+(``Channels``) and only for the lifetime of the process.
+
+Inbound delivery has two mutually exclusive modes, chosen at startup:
+
+* **Webhook** when ``HUB_HOST`` is set: Telegram POSTs updates to
+  ``/channel/telegram/<agent_id>`` (served by ``server/telegram.py``).
+* **Long polling** otherwise: one background task per binding calls
+  ``getUpdates``.
+
+Both modes converge on :meth:`Channels.handle_update`.
+
+Outbound is a single ``sendMessage`` call, exposed to the model as a
+``ToolSpec`` carrying a ``func`` so the agent's run loop can execute it.
+"""
+
+import asyncio
+import os
+import time
+from collections import deque
+from datetime import datetime
+from functools import partial
+
+import httpx
+
+import database as db
+from tool import ToolSpec, ToolParameter
+
+TELEGRAM_API = "https://api.telegram.org"
+
+# Telegram long-poll timeout (server holds the request open this long).
+POLL_TIMEOUT = 50
+# Rolling transcript cap per conversation, in bytes (oldest trimmed first).
+TRANSCRIPT_BYTE_CAP = 8 * 1024
+# Telegram hard limit for a single text message.
+MAX_MESSAGE_LENGTH = 4096
+
+
+class Channels:
+    """Owns channel bindings, in-memory transcripts, pollers and HTTP."""
+
+    def __init__(self, app):
+        self._app = app
+        self._client = None
+        self._tasks = []
+        self._transcripts = {}     # (agent_id, chat_id) -> {transcript, lock}
+        self._last_update = {}     # (agent_id, kind) -> update_id
+
+    # --- Startup / lifecycle ---
+
+    async def setup(self):
+        """Query channel resources and register listeners where necessary.
+
+        Called once at startup, after ``db.init_database`` and App
+        construction. Webhook mode reconciles the webhook per binding; when
+        ``HUB_HOST`` is absent we warn and start one long-polling task per
+        binding instead.
+        """
+        resources = await db.AgentResource.filter(kind="channel", key="telegram").all()
+        if not resources:
+            print("[channel] no telegram channel bindings found")
+            return
+
+        hub_host = os.getenv("HUB_HOST")
+        if hub_host:
+            for resource in resources:
+                try:
+                    await self._reconcile_webhook(resource, hub_host)
+                except Exception as e:
+                    print(f"[channel] webhook setup failed for agent {resource.agent_id}: {e}")
+        else:
+            print("WARNING: HUB_HOST is not set; telegram channels fall back to long polling")
+            for resource in resources:
+                self._track(asyncio.create_task(self._poll(resource)))
+
+    def shutdown(self):
+        """Cancel every tracked poller / in-flight run."""
+        for task in list(self._tasks):
+            task.cancel()
+
+    # --- Telegram HTTP ---
+
+    def _get_client(self):
+        if self._client is None:
+            self._client = httpx.AsyncClient()
+        return self._client
+
+    async def _api(self, token, method, payload=None, timeout=30):
+        """Call a Telegram Bot API method and return its ``result``.
+
+        Raises on transport errors and on ``ok: false`` responses.
+        """
+        url = f"{TELEGRAM_API}/bot{token}/{method}"
+        response = await self._get_client().post(url, json=payload or {}, timeout=timeout)
+        data = response.json()
+        if not data.get("ok"):
+            raise Exception(f"telegram {method} failed: {data.get('description')}")
+        return data.get("result")
+
+    async def get_binding(self, agent_id):
+        return await db.AgentResource.get_or_none(
+            agent_id=agent_id, kind="channel", key="telegram")
+
+    # --- Webhook mode ---
+
+    async def _reconcile_webhook(self, resource, hub_host):
+        value = resource.value or {}
+        token = value.get("token")
+        if not token:
+            print(f"[channel] telegram binding of agent {resource.agent_id} has no token, skipping")
+            return
+
+        url = f"https://{hub_host}/channel/telegram/{resource.agent_id}"
+        info = await self._api(token, "getWebhookInfo")
+        if info.get("url") == url:
+            print(f"[channel] webhook already registered for agent {resource.agent_id}")
+            return
+
+        payload = {"url": url, "allowed_updates": ["message"]}
+        secret = value.get("secret")
+        if secret:
+            payload["secret_token"] = secret
+        await self._api(token, "setWebhook", payload)
+        print(f"[channel] registered webhook {url} for agent {resource.agent_id}")
+
+    def accept_update(self, agent_id, kind, update_id):
+        """Drop duplicate webhook updates by remembering the last update id."""
+        if update_id is None:
+            return True
+        key = (agent_id, kind)
+        last = self._last_update.get(key)
+        if last is not None and update_id <= last:
+            return False
+        self._last_update[key] = update_id
+        return True
+
+    def dispatch(self, agent_id, kind, payload):
+        """Accept an inbound update and run it in the background.
+
+        Does the duplicate check, spawns the tracked run and returns
+        immediately so the caller (the webhook route) holds no channel
+        bookkeeping.
+        """
+        if not self.accept_update(agent_id, kind, payload.get("update_id")):
+            return
+        self._track(asyncio.create_task(self.handle_update(agent_id, kind, payload)))
+
+    # --- Long-polling mode ---
+
+    async def _poll(self, resource):
+        """Long-poll ``getUpdates`` for one telegram binding until cancelled.
+
+        ``offset`` is recomputed from the last seen ``update_id`` so each
+        update is acknowledged and delivered exactly once. Transport errors
+        back off instead of busy-looping.
+        """
+        value = resource.value or {}
+        token = value.get("token")
+        agent_id = resource.agent_id
+        if not token:
+            print(f"[channel] telegram binding of agent {agent_id} has no token, not polling")
+            return
+
+        # Webhooks and getUpdates are mutually exclusive: clear any webhook
+        # once before the first poll.
+        try:
+            await self._api(token, "deleteWebhook", {"drop_pending_updates": False})
+        except Exception as e:
+            print(f"[channel] deleteWebhook failed for agent {agent_id}: {e}")
+
+        offset = None
+        backoff = 1
+        print(f"[channel] polling telegram updates for agent {agent_id}")
+        while True:
+            try:
+                payload = {
+                    "timeout": POLL_TIMEOUT,
+                    "limit": 100,
+                    "allowed_updates": ["message"],
+                }
+                if offset is not None:
+                    payload["offset"] = offset
+                updates = await self._api(
+                    token, "getUpdates", payload, timeout=POLL_TIMEOUT + 15)
+                backoff = 1
+                for update in updates:
+                    offset = update["update_id"] + 1
+                    await self.handle_update(agent_id, "telegram", update)
+            except asyncio.CancelledError:
+                print(f"[channel] polling stopped for agent {agent_id}")
+                raise
+            except Exception as e:
+                print(f"[channel] telegram polling error for agent {agent_id}: {e}")
+                await asyncio.sleep(backoff)
+                backoff = min(backoff * 2, 60)
+
+    # --- Inbound ---
+
+    async def handle_update(self, agent_id, kind, payload):
+        """Handle one inbound update from a channel.
+
+        Minimal path: only plain text messages from non-bot senders are
+        consumed. Everything else (edited messages, media, callbacks) is
+        ignored.
+        """
+        if kind != "telegram":
+            print(f"[channel] unsupported channel kind '{kind}'")
+            return
+
+        message = payload.get("message")
+        if not message:
+            return
+        text = message.get("text")
+        if not text:
+            return
+
+        sender = message.get("from") or {}
+        if sender.get("is_bot"):
+            return
+
+        chat = message.get("chat") or {}
+        chat_id = chat.get("id")
+        if chat_id is None:
+            return
+
+        binding = await self.get_binding(agent_id)
+        if binding is None:
+            print(f"[channel] no telegram binding for agent {agent_id}, dropping update")
+            return
+
+        sender_name = self._sender_name(sender, chat)
+        state = self._state(agent_id, chat_id)
+        async with state["lock"]:
+            state["transcript"].append({
+                "sender": sender_name,
+                "text": text,
+                "time": time.time(),
+            })
+            self._trim(state["transcript"])
+            transcript = self._render(state["transcript"])
+            await self._run_agent(agent_id, kind, chat_id, chat, sender_name, transcript)
+
+    async def _run_agent(self, agent_id, kind, chat_id, chat, sender_name, transcript):
+        """Fire one prompt run for a conversation event and drain it.
+
+        The run input is flattened to an instruction plus exactly two user
+        messages: a status/info message and the rendered conversation.
+        """
+        instruction = (
+            "You are responding to a Telegram conversation. The messages below "
+            "are the recent conversation; the first user message carries channel "
+            "context. Reply to the latest message by calling the `send_message` "
+            "tool with the text to send. Do not add commentary outside the tool call."
+        )
+
+        title = chat.get("title") or sender_name
+        status = "\n".join([
+            "Channel: telegram",
+            f"Chat id: {chat_id}",
+            f"Chat title: {title}",
+            f"Chat type: {chat.get('type')}",
+            f"Current time: {datetime.now().isoformat(timespec='seconds')}",
+            f"Latest message from: {sender_name}",
+        ])
+
+        tool = ToolSpec(
+            name="send_message",
+            description="Send a text message back to this Telegram conversation.",
+            parameters=[ToolParameter(
+                name="text",
+                type="string",
+                description="The message text to send to the conversation.",
+                required=True,
+            )],
+            func=partial(self.send_message, agent_id, kind, chat_id),
+        )
+
+        messages = [
+            {"role": "user", "content": status},
+            {"role": "user", "content": transcript},
+        ]
+
+        async def drain():
+            async for msg in self._app.prompt_agent(
+                agent_id,
+                instruction,
+                input=None,
+                options={"tools": [tool], "messages": messages},
+            ):
+                self._log_message(kind, chat_id, msg)
+
+        task = self._track(asyncio.create_task(drain()))
+        try:
+            await task
+        except asyncio.CancelledError:
+            raise
+        except Exception as e:
+            print(f"[channel] run failed for agent {agent_id}: {e}")
+
+    def _log_message(self, kind, chat_id, msg):
+        content = msg.content
+        if not isinstance(content, str):
+            content = repr(content)
+        print(f"[channel:{kind}:{chat_id}] {msg.role}: {content}")
+
+    # --- Outbound ---
+
+    async def send_message(self, agent_id, kind, chat_id, text):
+        """Send a text message to a channel conversation.
+
+        Returns a short result string so the model sees the tool outcome.
+        """
+        if kind != "telegram":
+            return f"error: unsupported channel '{kind}'"
+
+        binding = await self.get_binding(agent_id)
+        if binding is None:
+            return "error: no telegram channel binding for this agent"
+
+        token = (binding.value or {}).get("token")
+        if not token:
+            return "error: telegram binding has no token"
+
+        text = (text or "").strip()
+        if not text:
+            return "error: message text is empty"
+        if len(text) > MAX_MESSAGE_LENGTH:
+            text = text[:MAX_MESSAGE_LENGTH]
+
+        try:
+            await self._api(token, "sendMessage", {"chat_id": chat_id, "text": text})
+            return "ok"
+        except Exception as e:
+            print(f"[channel] sendMessage failed for agent {agent_id}: {e}")
+            return f"error: {e}"
+
+    # --- In-memory transcript ---
+
+    def _state(self, agent_id, chat_id):
+        key = (agent_id, chat_id)
+        state = self._transcripts.get(key)
+        if state is None:
+            state = {"transcript": deque(), "lock": asyncio.Lock()}
+            self._transcripts[key] = state
+        return state
+
+    def _trim(self, transcript):
+        """Drop oldest messages until the deque fits the byte cap."""
+        while transcript and self._size(transcript) > TRANSCRIPT_BYTE_CAP:
+            transcript.popleft()
+
+    def _size(self, transcript):
+        return sum(
+            len(m["sender"].encode("utf-8")) + len(m["text"].encode("utf-8")) + 2
+            for m in transcript
+        )
+
+    def _render(self, transcript):
+        if not transcript:
+            return "(no messages)"
+        return "\n".join(f"{m['sender']}: {m['text']}" for m in transcript)
+
+    # --- Parse helpers ---
+
+    def _sender_name(self, user, chat):
+        name = " ".join(filter(None, [user.get("first_name"), user.get("last_name")]))
+        username = user.get("username")
+        if name and username:
+            return f"{name} (@{username})"
+        if name:
+            return name
+        if username:
+            return f"@{username}"
+        return chat.get("title") or "unknown"
+
+    # --- Tracking ---
+
+    def _track(self, task):
+        self._tasks.append(task)
+
+        def discard(t):
+            if t in self._tasks:
+                self._tasks.remove(t)
+
+        task.add_done_callback(discard)
+        return task

backend/src/server/telegram.py

@@ -0,0 +1,36 @@
+"""Telegram webhook endpoint.
+
+Telegram delivers updates to ``POST /channel/telegram/<agent_id>`` once a
+webhook is registered for that agent's binding (see ``channels.py``). The
+route is always served; it is only *useful* when ``HUB_HOST`` is set and
+the webhook has been reconciled, but that is a startup concern, not a
+routing one.
+
+Mirrors ``server/openai.py``'s ``register(server, app)`` pattern.
+"""
+
+from quart import request
+
+
+def register(server, app):
+    @server.post('/channel/telegram/<agent_id>')
+    async def telegram_webhook(agent_id):
+        try:
+            agent_id = int(agent_id)
+        except (TypeError, ValueError):
+            return "invalid agent id", 404
+
+        binding = await app.channels.get_binding(agent_id)
+        if binding is None:
+            return "no telegram channel for this agent", 404
+
+        secret = request.headers.get('X-Telegram-Bot-Api-Secret-Token')
+        expected = (binding.value or {}).get('secret')
+        if expected and secret != expected:
+            return "forbidden", 403
+
+        payload = await request.get_json(silent=True) or {}
+
+        # Never block the response on the run: Telegram retries on timeouts.
+        app.channels.dispatch(agent_id, 'telegram', payload)
+        return {"ok": True}

backend/src/server/web.py

@@ -2,6 +2,7 @@ from quart import Quart, request, send_from_directory, send_file, current_app
 from quart_auth import QuartAuth, AuthUser, login_required, login_user, current_user
 from server.websocket import register as register_websocket
 from server.openai import register as register_openai
+from server.telegram import register as register_telegram
 
 import database as db
 import drives
@@ -175,5 +176,6 @@ async def start_server(app, port, shutdown_event):
 
     register_websocket(server, app, auth_manager)
     register_openai(server, app)
+    register_telegram(server, app)
 
-    await server.run_task(debug=True, shutdown_trigger=shutdown_event.wait, port=port)
+    await server.run_task(debug=True, shutdown_trigger=shutdown_event.wait, host="0.0.0.0", port=port)

backend/src/server/websocket.py

@@ -68,14 +68,54 @@ def register(server, app, auth_manager):
             stop_listening = app.on_event(on_event)
 
             while True:
-                req = json.loads(await websocket.receive())
+                asyncio.create_task(handle_request(
-                if not 'action' in req:
+                    app, user_id, session_connections, send, await websocket.receive()
+                ))
+
+        finally:
+            print("User logged out:", user_id)
+            sender_task.cancel()
+            if ws_task in app.ws_tasks:
+                app.ws_tasks.remove(ws_task)
+            # Unregister the event listener: the send closure would
+            # otherwise keep queueing events for a dead socket forever
+            if stop_listening:
+                stop_listening()
+            # All remotes served by this session are disconnected with it.
+            # disconnectRemote is idempotent, so remotes that were already
+            # disconnected manually (possibly from another session) are safe.
+            for remotename in list(session_connections):
+                app.disconnectRemote(user_id, remotename)
+            app.remove_user_connection(user_id, send_queue)
+
+# Error handling
+async def handle_request(app, user_id, session_connections, send, req):
+    try:
+        req = json.loads(req)
+        await process_request(app, user_id, session_connections, send, req)
+
+    except Exception as err:
+        action = req.get('action') if isinstance(req, dict) else None
+        nonce  = req.get('nonce') if isinstance(req, dict) else None
+
+        # Maintainers
+        print(f"[ws] req error @{user_id} {action!r}: {err!r}")
+        print(traceback.format_exc())
+        # Client — nonce makes this routable to the caller's promise/callback
+        try:
+            await send({'error': str(err), 'nonce': nonce})
+        except Exception:
+            print(traceback.format_exc())
+
+# Actual processing
+async def process_request(app, user_id, session_connections, send, req):
+    if 'action' not in req:
         await send({'error': 'no action'})
-                    continue
+        return
     action = req['action']
     body = req.get('body', {})
     nonce = req.get('nonce', None)
-                print(f"[ws] action @{user_id} {action}")
+    print(f"[ws] req @{user_id} {action}")
 
     async def agent_id_from_body (allow_none=False):
         if 'agentId' in body:
@@ -88,7 +128,7 @@ def register(server, app, auth_manager):
             raise Exception("field 'agentId' or 'agentname' required")
 
     if action == "ping":
-                    await send({'message': "pong"})
+        await send({'message': "pong", 'nonce': nonce})
 
     elif action == "create-agent":
         name = body['name']
@@ -300,7 +340,7 @@ def register(server, app, auth_manager):
     elif action == "prompt-agent":
         if 'instruction' not in body and 'input' not in body:
             await send({'error': "instruction or input required", 'nonce': nonce})
-                        continue
+            return
 
         # Multiple-result action, same shape as read-stream: every
         # generated message is its own packet carrying the request
@@ -337,18 +377,3 @@ def register(server, app, auth_manager):
 
     else:
         await send({'error': 'unknown action: ' + action, 'nonce': nonce})
-        finally:
-            print("User logged out:", user_id)
-            sender_task.cancel()
-            if ws_task in app.ws_tasks:
-                app.ws_tasks.remove(ws_task)
-            # Unregister the event listener: the send closure would
-            # otherwise keep queueing events for a dead socket forever
-            if stop_listening:
-                stop_listening()
-            # All remotes served by this session are disconnected with it.
-            # disconnectRemote is idempotent, so remotes that were already
-            # disconnected manually (possibly from another session) are safe.
-            for remotename in list(session_connections):
-                app.disconnectRemote(user_id, remotename)
-            app.remove_user_connection(user_id, send_queue)

docs/telegram.md

@@ -0,0 +1,91 @@
+# Telegram Channel
+
+Connect a Telegram bot to an agent so the agent can read incoming messages and
+reply. Telegram is the only channel implemented today; the machinery lives in
+`backend/src/app_mixins/channel.py` (logic) and `backend/src/server/telegram.py`
+(webhook route).
+
+An agent that owns a `channel`/`telegram` resource *is* the agent that answers.
+You can bind the same bot token to several agents, and one agent can be bound to
+several bots.
+
+## 1. Create a bot
+
+Talk to [@BotFather](https://t.me/BotFather), create a bot, and copy its token
+(looks like `123456:ABC-DEF...`). Treat the token as a secret.
+
+## 2. Choose a delivery mode
+
+Telegram can push updates to a webhook, or the hub can pull them with long
+polling. The choice is driven entirely by the `HUB_HOST` environment variable:
+
+- **Webhook** — set `HUB_HOST` to a publicly reachable host. Telegram requires
+  HTTPS on ports 443, 80, 88 or 8443, so the hub must be reachable there.
+- **Long polling** — leave `HUB_HOST` unset. The hub prints a warning and polls
+  `getUpdates` instead. No public endpoint or TLS certificate is needed.
+
+The `POST /channel/telegram/<agent_id>` route is always served regardless of
+mode; in polling mode it simply never receives traffic.
+
+Webhooks and polling are mutually exclusive. A webhook is never registered when
+`HUB_HOST` is unset, and the poller clears any existing webhook (`deleteWebhook`)
+before it starts.
+
+## 3. Bind the bot to an agent
+
+Add a resource to the responding agent (via the resources UI or the
+`save-agent-resource` action):
+
+```
+kind  = "channel"
+key   = "telegram"
+value = {"token": "<bot token>", "secret": "<random webhook secret>", "name": "<display>"}
+```
+
+- `token` — the BotFather token. Required.
+- `secret` — a random string of `A-Z a-z 0-9 _ -`, 1–256 characters. Optional,
+  and only meaningful in webhook mode. When present, Telegram sends it back on
+  every request in the `X-Telegram-Bot-Api-Secret-Token` header and the route
+  rejects mismatches with `403`.
+- `name` — a display label. Optional.
+
+This is the only state the channel persists. Conversations are **not** stored —
+no `Chat`, `Thread` or `Message` rows are created. The rolling transcript is
+held in memory on the hub and discarded when the process exits.
+
+## 4. Restart the hub
+
+On startup the hub queries all `channel`/`telegram` resources and reconciles the
+provider:
+
+- **Webhook mode**: it calls `getWebhookInfo` for each binding and, when the
+  registered URL does not match `https://<HUB_HOST>/channel/telegram/<agent_id>`,
+  calls `setWebhook` with that URL, the `secret_token`, and
+  `allowed_updates=["message"]`. Verify with `getWebhookInfo`.
+- **Polling mode**: it starts one background task per binding. Confirm the bot is
+  reachable with `getMe`, then send a test message.
+
+## 5. Talk to the bot
+
+Message the bot. On each text message the hub:
+
+1. Appends the message to the conversation's in-memory transcript.
+2. Runs the agent with a flattened input — an instruction, a status message
+   (channel, chat id/title/type, current time, sender), and the rendered
+   conversation.
+3. The agent replies by calling the `send_message` tool, which issues a
+   `sendMessage` call to Telegram.
+
+Non-text updates (edited messages, media, callbacks) are ignored. Messages
+authored by other bots are ignored.
+
+## 6. Troubleshooting
+
+| Symptom | Cause |
+|---|---|
+| Webhook not set | `HUB_HOST` unset, or the host is not reachable over HTTPS. Check the startup log. |
+| Polling active when you expected a webhook | `HUB_HOST` is unset. |
+| `403` from the webhook route | `X-Telegram-Bot-Api-Secret-Token` does not match the resource's `secret`. |
+| `404` from the webhook route | No `channel`/`telegram` resource on that agent. |
+| No reply | The agent has no model/tool configuration, or the run errored — check the hub log for `[channel]` lines. |
+| `Conflict: can't use getUpdates method while webhook is active` | A stale webhook from a previous deployment. Remove it (`deleteWebhook`) or set `HUB_HOST` so webhook mode takes over. |