wip: responsiveness

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,445 @@
+"""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.mark_as_read(agent_id, chat_id)
+            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}",
+        ])
+
+        # Signal presence for the lifetime of this run.
+        await self.set_online_status(agent_id, chat_id, online=True)
+
+        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]
+
+        # Typing indicator immediately before the message so the user sees
+        # the three-dot animation right as the reply is ready.
+        try:
+            await self._api(token, "sendAction", {"chat_id": chat_id, "action": "typing"})
+        except Exception as e:
+            print(f"[channel] sendAction typing failed for agent {agent_id}: {e}")
+
+        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}"
+
+    async def send_typing(self, agent_id, chat_id, typing=True):
+        """Send a typing action indicator to the chat.
+
+        Telegram exposes this via ``sendAction`` with ``typing`` (start) or
+        ``cancel`` (stop). Failures are swallowed: an indicator is cosmetic
+        and must never break a run.
+        """
+        binding = await self.get_binding(agent_id)
+        if binding is None:
+            return
+        token = (binding.value or {}).get("token")
+        if not token:
+            return
+        action = "typing" if typing else "cancel"
+        try:
+            await self._api(token, "sendAction", {"chat_id": chat_id, "action": action})
+        except Exception as e:
+            print(f"[channel] sendAction failed: {e}")
+
+    async def mark_as_read(self, agent_id, chat_id):
+        """Mark an inbound message as read.
+
+        Telegram bots have no native read receipt; acceptance is already
+        handled by :meth:`accept_update`. This stays an explicit hook on the
+        inbound path so a provider that does support receipts can be wired in
+        here later.
+        """
+        pass
+
+    async def set_online_status(self, agent_id, chat_id, online=True):
+        """Signal presence for the conversation.
+
+        A single ``typing`` action marks the run as online; ``cancel`` clears
+        it. Delegates to :meth:`send_typing`.
+        """
+        await self.send_typing(agent_id, chat_id, typing=online)
+
+    # --- 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,275 +68,10 @@ def register(server, app, auth_manager):
             stop_listening = app.on_event(on_event)
 
             while True:
-                req = json.loads(await websocket.receive())
-                if not 'action' in req:
-                    await send({'error': 'no action'})
-                    continue
-                action = req['action']
-                body = req.get('body', {})
-                nonce = req.get('nonce', None)
-                print(f"[ws] action @{user_id} {action}")
+                asyncio.create_task(handle_request(
+                    app, user_id, session_connections, send, await websocket.receive()
+                ))
 
-                async def agent_id_from_body (allow_none=False):
-                    if 'agentId' in body:
-                        return body['agentId']
-                    elif 'agentname' in body:
-                        return await app.agent_id_from_name(body['agentname'])
-                    elif allow_none:
-                        return None
-                    else:
-                        raise Exception("field 'agentId' or 'agentname' required")
-
-                if action == "ping":
-                    await send({'message': "pong"})
-
-                elif action == "create-agent":
-                    name = body['name']
-                    agent_id = await app.create_agent(
-                        user_id=user_id,
-                        name=name,
-                        agentname=body.get('agentname'),
-                        description=body.get('description'),
-                        picture=body.get('picture'),
-                        kind=body['kind'],
-                        data=body.get('data'),
-                        temporary=body.get('temporary'),
-                    )
-                    await send({'message': "agent created", 'agentId': agent_id, 'nonce': nonce})
-
-                elif action == "update-agent":
-                    agent_id = await agent_id_from_body()
-                    await app.update_agent(
-                        agent_id=agent_id,
-                        name=body.get('name'),
-                        agentname=body.get('agentname'),
-                        description=body.get('description'),
-                        picture=body.get('picture'),
-                        data=body.get('data', {})
-                    )
-                    await send({'message': "agent updated", 'agentId': agent_id, 'nonce': nonce})
-
-                elif action == "get-models":
-                    await send({
-                        'models': [{"id": model.id, "name": model.name, "value": model.value, "kind": model.kind} for model in await app.get_models()],
-                        'nonce': nonce
-                    })
-
-                elif action == "save-model":
-                    model = await app.save_model(
-                        model_id=body.get("modelId"),
-                        name=body.get("name"),
-                        kind=body.get("kind"),
-                        value=body.get("value"),
-                    )
-                    if model:
-                        await send({'message': "model saved", 'modelId': model.id, 'nonce': nonce})
-                    elif "modelId" in body:
-                        await send({'message': "model deleted", 'modelId': body["modelId"], 'nonce': nonce})
-
-                elif action == "get-agent-config-fields":
-                    await send({'fields': await app.get_agent_config_fields(await agent_id_from_body()), 'nonce': nonce})
-
-                elif action == "get-agent-kinds":
-                    await send({'message': "agents retrieved", 'kinds': app.get_agent_kinds(), 'nonce': nonce})
-
-                elif action == "get-agent-kind-info":
-                    await send({'message': "agent kind retrieved", 'info': app.get_agent_kind_info(body['kind']), 'nonce': nonce})
-
-                elif action == "get-agents":
-                    await send({'message': "agents retrieved", 'agents': await app.get_agents(user_id), 'nonce': nonce})
-
-                elif action == "get-agent-resources":
-                    values = await app.get_agent_resources(await agent_id_from_body(), body.get('kind'), body.get('key'))
-                    await send({'resources': [{'kind': r.kind, 'key': r.key, 'value': r.value} for r in values], 'nonce': nonce})
-
-                elif action == "save-agent-resource":
-                    await app.save_agent_resource(await agent_id_from_body(), body['kind'], body['key'], body['value'])
-                    await send({'message': "resource saved", 'nonce': nonce})
-
-                elif action == "delete-agent-resource":
-                    await app.delete_agent_resource(await agent_id_from_body(), body['kind'], body['key'])
-                    await send({'message': "resource deleted", 'nonce': nonce})
-
-                elif action == "get-agent-runs":
-                    await send({
-                        'runs': await app.get_agent_runs(agent_id=await agent_id_from_body(allow_none=True), user_id=user_id),
-                        'nonce': nonce
-                    })
-
-                elif action == "cancel-agent-run":
-                    await app.cancel_agent_run(body['runId'])
-
-                elif action == "remote-list":
-                    await send({'remotes': app.listRemotes(user_id), 'nonce': nonce})
-
-                elif action == "remote-connect":
-                    # Connect a remote served by THIS session's channel.
-                    # Expected failures (duplicate name) are replied, not
-                    # raised, so the session stays alive.
-                    try:
-                        connection = app.connectRemote(user_id, body['remotename'], body.get('functions', []), send)
-                        session_connections[body['remotename']] = connection
-                        await send({'message': "remote connected", 'remotename': body['remotename'], 'nonce': nonce})
-                    except Exception as e:
-                        await send({'error': str(e), 'nonce': nonce})
-
-                elif action == "remote-disconnect":
-                    # Manual disconnect, allowed from any session of the user
-                    connection = app.disconnectRemote(user_id, body['remotename'])
-                    session_connections.pop(body['remotename'], None)
-                    if connection is None:
-                        await send({'error': "no connected remote named '" + str(body['remotename']) + "'", 'nonce': nonce})
-                    else:
-                        await send({'message': "remote disconnected", 'remotename': body['remotename'], 'nonce': nonce})
-
-                elif action == "remote-call":
-                    # Invoke a function of one of the user's connected remotes.
-                    # Expected failures (unknown remote/function, timeout,
-                    # remote-side error) are replied, not raised.
-
-
-                    async def _handle_remote_call(user_id, remotename, func_name, args, send, nonce):
-                        try:
-                            result = await app.callRemote(user_id, remotename, func_name, args)
-                            await send({'result': result, 'nonce': nonce})
-                        except Exception as e:
-                            await send({'error': str(e), 'nonce': nonce})
-                    asyncio.create_task(_handle_remote_call(
-                        user_id, body['remotename'], body['function'], body.get('args', {}), send, nonce
-                    ))
-
-                elif action == "remote-call-result":
-                    # A session serving remotes replies to a downstream call.
-                    # Fields live at the message top level, not in 'body'.
-                    # requestIds are '<remotename>#<n>' and resolve/reject
-                    # are no-ops for unknown ids, so fanning out over this
-                    # session's connections routes the reply correctly.
-                    app.resolve_remote_call(body['requestId'], body.get('result'), error=body.get('error'))
-
-                elif action == "create-chat":
-                    chat_id = await app.create_chat(
-                        user_id=user_id,
-                        name=body['name'],
-                        description=body.get('description'),
-                        picture=body.get('picture'),
-                        members=body.get('members', []),
-                        temporary=body.get('temporary'),
-                    )
-                    await send({'message': "chat created", 'chatId': chat_id, 'nonce': nonce})
-
-                elif action == "get-chats":
-                    chats_data = await app.get_chats(user_id)
-                    await send({'message': "chats retrieved", 'chats': chats_data, 'nonce': nonce})
-
-                elif action == "delete-chat":
-                    chat_id = await app.delete_chat(
-                        chat_id=body['chatId']
-                    )
-                    await send({'message': "chat deleted", 'chatId': chat_id, 'nonce': nonce})
-
-                elif action == "get-chat-messages":
-                    messages_data = await app.get_chat_messages(
-                        body['chatId'],
-                        earliest_index=body.get('earliestIndex', 0)
-                    )
-                    await send({'message': "chat messages retrieved", 'messages': messages_data, 'nonce': nonce})
-
-                elif action == "get-chat-threads":
-                    await send({'message': "threads retrieved", 'nonce': nonce,
-                        'threads': await app.get_chat_threads(body['chatId'])
-                    })
-
-                elif action == "get-chat-thread-heads":
-                    messages_data = await app.get_chat_thread_heads(
-                        body['chatId'],
-                        limit_message_id=body.get('limitMessageId', None)
-                    )
-                    await send({'message': "chat messages retrieved", 'messages': messages_data, 'nonce': nonce})
-
-                elif action == "get-thread-messages":
-                    messages_data = await app.get_thread_messages(
-                        body['chatId'],
-                        body['threadId'],
-                        limit_message_id=body.get('limitMessageId', None)
-                    )
-                    await send({'message': "chat messages retrieved", 'messages': messages_data, 'nonce': nonce})
-
-                elif action == "send-chat-message":
-                    message = await app.send_chat_message(
-                        chat_id=body['chatId'],
-                        thread_id=body.get('threadId'),
-                        content=body['content'],
-                        kind=body.get('kind'),
-                        subkind=body.get('subkind'),
-                        format=body.get('format'),
-                    )
-                    await send({'message': "message sent", 'messageId': message['id'], 'nonce': nonce})
-
-                elif action == "update-chat-message":
-                    message = await app.update_chat_message(
-                        message_id=body['messageId'],
-                        content=body.get('content'),
-                        kind=body.get('kind'),
-                        subkind=body.get('subkind'),
-                        format=body.get('format'),
-                    )
-                    await send({'message': "message sent", 'messageId': body['messageId'], 'nonce': nonce})
-
-                elif action == "create-chat-thread":
-                    thread_id = await app.create_chat_thread(
-                        chat_id=body['chatId'],
-                        name=body.get('name'),
-                    )
-                    await send({'message': "chat thread created", 'threadId': thread_id, 'nonce': nonce})
-
-                elif action == "fork-chat-thread":
-                    thread_id = await app.fork_chat_thread(
-                        thread_id=body['threadId'],
-                        message_id=body.get('messageId'),
-                    )
-                    await send({'message': "chat thread created", 'threadId': thread_id, 'nonce': nonce})
-
-                elif action == "prompt-agent":
-                    if 'instruction' not in body and 'input' not in body:
-                        await send({'error': "instruction or input required", 'nonce': nonce})
-                        continue
-
-                    # Multiple-result action, same shape as read-stream: every
-                    # generated message is its own packet carrying the request
-                    # nonce, so the client callback runs once per message. The
-                    # terminal eof packet resolves the action. Streams referenced
-                    # by open assistant/reasoning messages are read by the
-                    # client via the existing read-stream action.
-                    generator = app.prompt_agent(
-                        await agent_id_from_body(),
-                        body.get('instruction'),
-                        body.get('input'),
-                        body.get('options', {})
-                    )
-                    try:
-                        async for msg in generator:
-                            await send({'message': 'agent-message', 'data': msg, 'nonce': nonce})
-                    finally:
-                        # Disconnect / cancellation point: closes the run task.
-                        await generator.aclose()
-                    await send({'eof': True, 'nonce': nonce})
-
-                elif action == "read-stream":
-                    stream_id = body['streamId']
-                    print("Read Stream" + str(stream_id))
-                    stream = app.streams.get(stream_id)
-                    if stream is None:
-                        await send({'error': 'bad stream id: ' + str(stream_id), 'nonce': nonce})
-                    else:
-                        async def chunk_fn(chunk):
-                            await send({'data': chunk, 'eof': False, 'nonce': nonce})
-                        async def eof_fn():
-                            await send({'eof': True, 'fullData': stream.content, 'nonce': nonce})
-                        stream.listen(chunk_fn, eof_fn)
-
-                else:
-                    await send({'error': 'unknown action: ' + action, 'nonce': nonce})
         finally:
             print("User logged out:", user_id)
             sender_task.cancel()
@@ -352,3 +87,293 @@ def register(server, app, auth_manager):
             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'})
+        return
+    action = req['action']
+    body = req.get('body', {})
+    nonce = req.get('nonce', None)
+    print(f"[ws] req @{user_id} {action}")
+
+    async def agent_id_from_body (allow_none=False):
+        if 'agentId' in body:
+            return body['agentId']
+        elif 'agentname' in body:
+            return await app.agent_id_from_name(body['agentname'])
+        elif allow_none:
+            return None
+        else:
+            raise Exception("field 'agentId' or 'agentname' required")
+
+    if action == "ping":
+        await send({'message': "pong", 'nonce': nonce})
+
+    elif action == "create-agent":
+        name = body['name']
+        agent_id = await app.create_agent(
+            user_id=user_id,
+            name=name,
+            agentname=body.get('agentname'),
+            description=body.get('description'),
+            picture=body.get('picture'),
+            kind=body['kind'],
+            data=body.get('data'),
+            temporary=body.get('temporary'),
+        )
+        await send({'message': "agent created", 'agentId': agent_id, 'nonce': nonce})
+
+    elif action == "update-agent":
+        agent_id = await agent_id_from_body()
+        await app.update_agent(
+            agent_id=agent_id,
+            name=body.get('name'),
+            agentname=body.get('agentname'),
+            description=body.get('description'),
+            picture=body.get('picture'),
+            data=body.get('data', {})
+        )
+        await send({'message': "agent updated", 'agentId': agent_id, 'nonce': nonce})
+
+    elif action == "get-models":
+        await send({
+            'models': [{"id": model.id, "name": model.name, "value": model.value, "kind": model.kind} for model in await app.get_models()],
+            'nonce': nonce
+        })
+
+    elif action == "save-model":
+        model = await app.save_model(
+            model_id=body.get("modelId"),
+            name=body.get("name"),
+            kind=body.get("kind"),
+            value=body.get("value"),
+        )
+        if model:
+            await send({'message': "model saved", 'modelId': model.id, 'nonce': nonce})
+        elif "modelId" in body:
+            await send({'message': "model deleted", 'modelId': body["modelId"], 'nonce': nonce})
+
+    elif action == "get-agent-config-fields":
+        await send({'fields': await app.get_agent_config_fields(await agent_id_from_body()), 'nonce': nonce})
+
+    elif action == "get-agent-kinds":
+        await send({'message': "agents retrieved", 'kinds': app.get_agent_kinds(), 'nonce': nonce})
+
+    elif action == "get-agent-kind-info":
+        await send({'message': "agent kind retrieved", 'info': app.get_agent_kind_info(body['kind']), 'nonce': nonce})
+
+    elif action == "get-agents":
+        await send({'message': "agents retrieved", 'agents': await app.get_agents(user_id), 'nonce': nonce})
+
+    elif action == "get-agent-resources":
+        values = await app.get_agent_resources(await agent_id_from_body(), body.get('kind'), body.get('key'))
+        await send({'resources': [{'kind': r.kind, 'key': r.key, 'value': r.value} for r in values], 'nonce': nonce})
+
+    elif action == "save-agent-resource":
+        await app.save_agent_resource(await agent_id_from_body(), body['kind'], body['key'], body['value'])
+        await send({'message': "resource saved", 'nonce': nonce})
+
+    elif action == "delete-agent-resource":
+        await app.delete_agent_resource(await agent_id_from_body(), body['kind'], body['key'])
+        await send({'message': "resource deleted", 'nonce': nonce})
+
+    elif action == "get-agent-runs":
+        await send({
+            'runs': await app.get_agent_runs(agent_id=await agent_id_from_body(allow_none=True), user_id=user_id),
+            'nonce': nonce
+        })
+
+    elif action == "cancel-agent-run":
+        await app.cancel_agent_run(body['runId'])
+
+    elif action == "remote-list":
+        await send({'remotes': app.listRemotes(user_id), 'nonce': nonce})
+
+    elif action == "remote-connect":
+        # Connect a remote served by THIS session's channel.
+        # Expected failures (duplicate name) are replied, not
+        # raised, so the session stays alive.
+        try:
+            connection = app.connectRemote(user_id, body['remotename'], body.get('functions', []), send)
+            session_connections[body['remotename']] = connection
+            await send({'message': "remote connected", 'remotename': body['remotename'], 'nonce': nonce})
+        except Exception as e:
+            await send({'error': str(e), 'nonce': nonce})
+
+    elif action == "remote-disconnect":
+        # Manual disconnect, allowed from any session of the user
+        connection = app.disconnectRemote(user_id, body['remotename'])
+        session_connections.pop(body['remotename'], None)
+        if connection is None:
+            await send({'error': "no connected remote named '" + str(body['remotename']) + "'", 'nonce': nonce})
+        else:
+            await send({'message': "remote disconnected", 'remotename': body['remotename'], 'nonce': nonce})
+
+    elif action == "remote-call":
+        # Invoke a function of one of the user's connected remotes.
+        # Expected failures (unknown remote/function, timeout,
+        # remote-side error) are replied, not raised.
+
+
+        async def _handle_remote_call(user_id, remotename, func_name, args, send, nonce):
+            try:
+                result = await app.callRemote(user_id, remotename, func_name, args)
+                await send({'result': result, 'nonce': nonce})
+            except Exception as e:
+                await send({'error': str(e), 'nonce': nonce})
+        asyncio.create_task(_handle_remote_call(
+            user_id, body['remotename'], body['function'], body.get('args', {}), send, nonce
+        ))
+
+    elif action == "remote-call-result":
+        # A session serving remotes replies to a downstream call.
+        # Fields live at the message top level, not in 'body'.
+        # requestIds are '<remotename>#<n>' and resolve/reject
+        # are no-ops for unknown ids, so fanning out over this
+        # session's connections routes the reply correctly.
+        app.resolve_remote_call(body['requestId'], body.get('result'), error=body.get('error'))
+
+    elif action == "create-chat":
+        chat_id = await app.create_chat(
+            user_id=user_id,
+            name=body['name'],
+            description=body.get('description'),
+            picture=body.get('picture'),
+            members=body.get('members', []),
+            temporary=body.get('temporary'),
+        )
+        await send({'message': "chat created", 'chatId': chat_id, 'nonce': nonce})
+
+    elif action == "get-chats":
+        chats_data = await app.get_chats(user_id)
+        await send({'message': "chats retrieved", 'chats': chats_data, 'nonce': nonce})
+
+    elif action == "delete-chat":
+        chat_id = await app.delete_chat(
+            chat_id=body['chatId']
+        )
+        await send({'message': "chat deleted", 'chatId': chat_id, 'nonce': nonce})
+
+    elif action == "get-chat-messages":
+        messages_data = await app.get_chat_messages(
+            body['chatId'],
+            earliest_index=body.get('earliestIndex', 0)
+        )
+        await send({'message': "chat messages retrieved", 'messages': messages_data, 'nonce': nonce})
+
+    elif action == "get-chat-threads":
+        await send({'message': "threads retrieved", 'nonce': nonce,
+            'threads': await app.get_chat_threads(body['chatId'])
+        })
+
+    elif action == "get-chat-thread-heads":
+        messages_data = await app.get_chat_thread_heads(
+            body['chatId'],
+            limit_message_id=body.get('limitMessageId', None)
+        )
+        await send({'message': "chat messages retrieved", 'messages': messages_data, 'nonce': nonce})
+
+    elif action == "get-thread-messages":
+        messages_data = await app.get_thread_messages(
+            body['chatId'],
+            body['threadId'],
+            limit_message_id=body.get('limitMessageId', None)
+        )
+        await send({'message': "chat messages retrieved", 'messages': messages_data, 'nonce': nonce})
+
+    elif action == "send-chat-message":
+        message = await app.send_chat_message(
+            chat_id=body['chatId'],
+            thread_id=body.get('threadId'),
+            content=body['content'],
+            kind=body.get('kind'),
+            subkind=body.get('subkind'),
+            format=body.get('format'),
+        )
+        await send({'message': "message sent", 'messageId': message['id'], 'nonce': nonce})
+
+    elif action == "update-chat-message":
+        message = await app.update_chat_message(
+            message_id=body['messageId'],
+            content=body.get('content'),
+            kind=body.get('kind'),
+            subkind=body.get('subkind'),
+            format=body.get('format'),
+        )
+        await send({'message': "message sent", 'messageId': body['messageId'], 'nonce': nonce})
+
+    elif action == "create-chat-thread":
+        thread_id = await app.create_chat_thread(
+            chat_id=body['chatId'],
+            name=body.get('name'),
+        )
+        await send({'message': "chat thread created", 'threadId': thread_id, 'nonce': nonce})
+
+    elif action == "fork-chat-thread":
+        thread_id = await app.fork_chat_thread(
+            thread_id=body['threadId'],
+            message_id=body.get('messageId'),
+        )
+        await send({'message': "chat thread created", 'threadId': thread_id, 'nonce': nonce})
+
+    elif action == "prompt-agent":
+        if 'instruction' not in body and 'input' not in body:
+            await send({'error': "instruction or input required", 'nonce': nonce})
+            return
+
+        # Multiple-result action, same shape as read-stream: every
+        # generated message is its own packet carrying the request
+        # nonce, so the client callback runs once per message. The
+        # terminal eof packet resolves the action. Streams referenced
+        # by open assistant/reasoning messages are read by the
+        # client via the existing read-stream action.
+        generator = app.prompt_agent(
+            await agent_id_from_body(),
+            body.get('instruction'),
+            body.get('input'),
+            body.get('options', {})
+        )
+        try:
+            async for msg in generator:
+                await send({'message': 'agent-message', 'data': msg, 'nonce': nonce})
+        finally:
+            # Disconnect / cancellation point: closes the run task.
+            await generator.aclose()
+        await send({'eof': True, 'nonce': nonce})
+
+    elif action == "read-stream":
+        stream_id = body['streamId']
+        print("Read Stream" + str(stream_id))
+        stream = app.streams.get(stream_id)
+        if stream is None:
+            await send({'error': 'bad stream id: ' + str(stream_id), 'nonce': nonce})
+        else:
+            async def chunk_fn(chunk):
+                await send({'data': chunk, 'eof': False, 'nonce': nonce})
+            async def eof_fn():
+                await send({'eof': True, 'fullData': stream.content, 'nonce': nonce})
+            stream.listen(chunk_fn, eof_fn)
+
+    else:
+        await send({'error': 'unknown action: ' + action, 'nonce': nonce})

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. |