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/app_mixins/__init__.py

@@ -1,2 +1,2 @@
 from app_mixins.chat import ChatMixin
-from app_mixins.agent import AgentMixin
+from app_mixins.agent import AgentMixin

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)