fix: drives

feat(backend): refactor websocket handler
feat(chat): collapse log messages
2026-10-14 02:02:31 -05:00 · 2026-10-14 01:44:34 -05:00 · 2026-10-14 03:31:05 +00:00
12 changed files with 240 additions and 560 deletions
@@ -1,7 +1,6 @@
 Quart==0.20.0
 quart-auth
 quart-cors
 httpx
 sqlalchemy
 nest_asyncio
 replicate
@@ -22,7 +22,6 @@ 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):
@@ -35,9 +34,6 @@ 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
@@ -240,16 +236,13 @@ async def main():
        openrouter_token=os.getenv("OPENROUTER_API_KEY")
    )
-    # 3. Set up channel listeners (webhook reconciliation or long polling)
+    # 3. Start Server
    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))
-    # 5. Graceful Termination
+    # 4. Graceful Termination
    loop = asyncio.get_running_loop()
    def interrupt():
        print("Interrupt Signal")
@@ -257,8 +250,6 @@ 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)
@@ -1,2 +1,2 @@
 from app_mixins.chat import ChatMixin
-from app_mixins.agent import AgentMixin
+from app_mixins.agent import AgentMixin
@@ -1,397 +0,0 @@
 """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
@@ -27,7 +27,7 @@ async def upload_file(user_id, drive_name, filename, stream, mime_type=None):
        drive = await db.Drive.create(user_id=user_id, name=drive_name)
    # Check collision
-    if await db.Drive.get_or_none(drive_id=drive.id, filename=filename):
+    if await db.DriveFile.get_or_none(drive_id=drive.id, filename=filename):
        raise Exception("Duplicated file")
    # 2. Prepare disk path
@@ -82,7 +82,7 @@ async def delete_file(file_id):
    return abs_path
 async def resolve_file(user_id, drive_name, filename):
-    """Resolve file to absolute disk path after permission check. Returns (path, mime)."""
+    """Resolve file to absolute disk path after permission check. Returns (abs_path, file.id)."""
    # Eager load drive and verify ownership
    drive = await db.Drive.get_or_none(user_id=user_id, name=drive_name)
    if drive.user_id != user_id:
@@ -1,36 +0,0 @@
 """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}
@@ -2,7 +2,6 @@ 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
@@ -176,6 +175,5 @@ 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, host="0.0.0.0", port=port)
+    await server.run_task(debug=True, shutdown_trigger=shutdown_event.wait, port=port)
@@ -1,27 +1,29 @@
 # Drives
-Per-user file storage: a **drive** is a named container owned by a user, and each **file** is a database record plus a blob on disk. Everything is addressed by *drive name + filename* — the random file id never appears in URLs. Drives bypass the WebSocket action protocol entirely: all endpoints are plain REST (routes in `backend/src/server/web.py`, logic in `backend/src/drives.py`), so downloads can be plain `<a>` links.
+Per-user file storage: a **drive** is a named container owned by a user, and each **file** is a database record plus a blob on disk. Client-facing routes are addressed by *drive name + filename* — the random file id never appears in URLs — though several `drives.py` helpers below are id-addressed internally. Drives bypass the WebSocket action protocol entirely: all endpoints are plain REST (routes in `backend/src/server/web.py`, logic in `backend/src/drives.py`), so downloads can be plain `<a>` links.
 ## Logical model
 - **Drive** — user-scoped, addressed by name. Created implicitly on first upload to a new name; there is no create/rename/delete-drive endpoint.
 - **File** — readable `filename`, mime type, size, `created`/`accessed` timestamps.
 - **Folders are virtual** — a filename may contain slashes (`docs/api/readme.md`); the Drives view splits names to render a tree, but storage stays flat.
- **Identity** — `DriveFile.id` is a random hex string (`secrets.token_hex(8)`) that doubles as the on-disk filename, so user-facing names never touch the filesystem.
+- **Identity** — `DriveFile.id` is a random hex string (`secrets.token_hex(8)`, 16 hex chars) that doubles as the on-disk filename, so user-facing names never touch the filesystem.
 ## Backend interface
 Download links can't set headers, so the GET/DELETE file routes additionally accept the token as a `?token=` query parameter.
 Client-facing file routes are drive+filename-addressed, but the `drives.py` helpers behind them are mixed: `upload_file` accepts drive+filename directly, while `get_metadata(file_id)` and `delete_file(file_id)` are id-addressed, so their routes must first resolve filename → id (via the drive's file records) before calling them. The two listing routes are not implemented in `drives.py` at all — their logic lives elsewhere.
 | Route | Behavior |
 | --- | --- |
-| `GET /files` | List the user's drives: `[{id, name}]` |
+| `GET /files` | List the user's drives: `[{id, name}]`. Not implemented in `drives.py`. |
-| `GET /files/<drive>` | List files: `[{id, name, size, mime, created}]` |
+| `GET /files/<drive>` | List files: `[{id, name, size, mime, created}]`. Not implemented in `drives.py`. |
 | `POST /files/<drive>/<filename>` | Multipart upload (field `file`) in a single request. The URL tail — `<path:filename>`, slashes allowed — becomes the stored filename; the part's own name is the fallback. Returns `{fileId}`. |
-| `GET /files/<drive>/<filename>` | Serve the bytes with their stored mime type, or return `{id, name, size, mime, created, accessed}` when `?metadata` is present. |
+| `GET /files/<drive>/<filename>` | Serve the bytes, with the mime type looked up from the file record (not from `resolve_file`), or return `{id, name, size, mime, created, accessed}` when `?metadata` is present. |
-| `DELETE /files/<drive>/<filename>` | Delete the record, then the blob. Returns `{fileId}`. |
+| `DELETE /files/<drive>/<filename>` | Delete the record, then the blob. Returns the removed absolute disk path (`abs_path`). |
-Every resolve (download, metadata, delete) stamps `accessed_at`.
+Only `resolve_file` stamps `accessed_at`; `get_metadata` and `delete_file` never touch it.
 ## Frontend interface
@@ -29,20 +31,20 @@ In `frontend/src/backend.js`:
 - `webFetch(path, options)` — generic REST helper: sets `Authorization: Bearer` (waiting for login if needed), parses JSON, throws on non-2xx. Downloads bypass it.
 - `getDrives()`, `getDriveFiles(driveName)`, `deleteDriveFile(driveName, filename)`
- `uploadDriveFile(driveName, file)` — multipart POST of the whole file.
+- `uploadDriveFile(driveName, file, path)` — multipart-POSTs the whole file to `/files/<driveName>/<path>`, where `path` is the destination filename including any virtual folder prefix (`docs/api/readme.md`). The composite path belongs in the URL, since the upload route only exists at `/files/<drive>/<path:filename>`.
 - `driveFileUrl(driveName, filename)` — GET URL carrying `?token=`, used for download anchors and for opening a file in a tab.
-`frontend/src/views/Drives.vue` lazy-loads a drive's files on first expand, rebuilds the folder tree from the flat file list on every load (open folders live on the drive object, so they survive reloads), uploads through a file picker, and deletes with confirmation.
+`frontend/src/views/Drives.vue` lazy-loads a drive's files on first expand, rebuilds the folder tree from the flat file list on every load (open folders live on the drive object, so they survive reloads), uploads through a file picker (building the target path from the picked folder and passing it as `uploadDriveFile`'s third argument), and deletes with confirmation.
 ## Implementation notes
- Models in `backend/src/database.py` (migration `0002_auto_20260921_1840`): `Drive(id, user_id, name, created_at)` and `DriveFile(id char(64) pk, drive_id, filename, mime_type, size_bytes, accessed_at, created_at)`, table `drive_files`. Ownership rides the `User → Drive → File` chain and is checked before any disk access.
+- Models in `backend/src/database.py` (migration `0002_auto_20260921_1840`): `Drive(id, user_id, name, created_at)` and `DriveFile(id, drive_id, filename, mime_type, size_bytes, accessed_at, created_at)`, table `drive_files`. The `id` is a 16-char hex string, not `char(64)`. Ownership rides the `User → Drive → File` chain; it is checked in `resolve_file` before disk access, but `delete_file` and `get_metadata` take a bare `file_id` and perform no ownership check of their own — their routes must guard them.
 - Disk layout: `{data_dir}/drives/{drive_id}/{file_id}` — keyed by drive id, not by user or name. `init_drives(data_directory)`, called from `main()`, creates the root.
 - Upload streams to disk in 8 KiB chunks and only then writes the DB record; delete removes the record first, then the blob.
 ## Known issues
 - `uploadDriveFile` POSTs to `/files/<drive>`, where no POST route exists, and drops the folder-path argument `Drives.vue` passes it — UI uploads fail, and folder-targeted upload is not wired up.
 - Drives are only created implicitly, and the view only uploads into existing drives, so the first drive cannot be created from the frontend.
 - Uploads are meant to refuse a filename already used in the drive, but the check queries the `Drive` model with `DriveFile`'s fields, so it errors instead of detecting duplicates.
- `resolve_file` dereferences the drive without a None check, so a nonexistent drive yields a 500 instead of a 404.
+- `resolve_file` dereferences the drive without a None check, so a nonexistent drive yields a 500 instead of a 404.
 - `delete_file` and `get_metadata` take a bare `file_id` with no ownership check; if their routes don't resolve and authorize the file against the owning user, any authenticated user could delete or inspect another user's file by id.
@@ -1,91 +0,0 @@
 # 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. |
@@ -379,6 +379,8 @@ export function invalidateCache({ chats=false, agents=false } = {}) {
 // Drive endpoints are plain HTTP routes instead of WebSocket actions,
 // so they authenticate with the bearer token in the Authorization header.
 // TODO: Next time these functions need updating, move them directly to views/Drives.vue
 export function getDrives () {
  return webFetch('/files');
 }
@@ -387,10 +389,10 @@ export function getDriveFiles (driveName) {
  return webFetch(`/files/${driveName}`);
 }
-export function uploadDriveFile (driveName, file) {
+export function uploadDriveFile (driveName, file, path) {
  const body = new FormData();
  body.append('file', file, file.name);
-  return webFetch(`/files/${driveName}`, {
+  return webFetch(`/files/${driveName}/${path}`, {
    method: 'POST',
    body,
  });
@@ -2,6 +2,7 @@
 import { ref, computed, watch } from 'vue';
 import { marked } from "marked";
 import * as backend from '../backend';
 import * as iconoir from '@iconoir/vue';
 marked.use({
  renderer: {
    code ({lang, text}) {
@@ -33,6 +34,12 @@ const props = defineProps({
  logExpanded: {
    type: Boolean,
    required: false,
  },
  // Toggle set of expanded message ids, shared with the parent view.
  // Used by log-groups to expand/collapse the previewed log.
  expandedLogs: {
    type: Object,
    required: false,
  }
 });
@@ -45,6 +52,8 @@ function chatEntryClass(msg) {
    return ['resource', 'resource-' + (msg.subkind || 'generic')];
  } else if (msg.type == 'log') {
    return ['log'];
  } else if (msg.type == 'log-group') {
    return ['log-group'];
  } else if (msg.type == 'thread-header') {
    let cls = ['thread'];
    if (!msg.active) cls.push(msg.isFork ? 'thread-fork' : 'thread-closed')
@@ -108,10 +117,106 @@ watch(() => [props.message.format, props.message.content], ([newFormat, newConte
  }
 }, {immediate: true})
 // === Log groups === //
 // A log-group collects consecutive log messages. Each one is shown as a square
 // icon (which selects it as the preview); the selected log is previewed below.
 const logs = computed(() =>
  props.message.type == 'log-group' ? (props.message.logs || []) : []
 );
 // Icon used for a log's square button, chosen by log subkind
 function logIcon(subkind) {
  if (subkind == 'reasoning') return iconoir.Brain;
  if (subkind == 'tool-call') return iconoir.Wrench;
  return iconoir.Terminal;
 }
 // Null means "default to the last log in the group"
 const selectedLogId = ref(null);
 const selectedLog = computed(() => {
  const arr = logs.value;
  if (!arr.length) return null;
  if (selectedLogId.value != null) {
    const found = arr.find(l => l.id == selectedLogId.value);
    if (found) return found;
  }
  return arr[arr.length - 1];
 });
 // Streamed logs resolve their content through a shared per-stream cache, so
 // switching the preview between logs still shows up-to-date content.
 const boundStreams = new Set();
 const streamContents = ref({});
 function bindLogStreams() {
  for (const log of logs.value) {
    if (log.format != 'stream') continue;
    const streamId = parseInt(log.content);
    if (boundStreams.has(streamId)) continue;
    boundStreams.add(streamId);
    const stream = backend.readStream(streamId);
    const update = () => {
      streamContents.value = { ...streamContents.value, [streamId]: stream.partialContent };
    };
    update();
    stream.on('data', update);
  }
 }
 watch(logs, bindLogStreams, { immediate: true });
 const selectedLogText = computed(() => {
  const log = selectedLog.value;
  if (!log) return '';
  if (log.format == 'stream') return streamContents.value[parseInt(log.content)] ?? '';
  return log.content ?? '';
 });
 const groupLogLong = computed(() => {
  if (!selectedLog.value) return false;
  const lines = String(selectedLogText.value ?? '').split('\n');
  return lines.length > LOG_MAX_LINES || lines.some(line => line.length > LOG_MAX_LENGTH);
 });
 const groupExpanded = computed(() =>
  selectedLog.value != null && !!props.expandedLogs?.has(selectedLog.value.id)
 );
 // Collapsed preview keeps the *last* N characters (the tail), not the head;
 // CSS then clips the top so only the bottom lines show.
 const groupLogContent = computed(() => {
  const text = String(selectedLogText.value ?? '');
  if (groupLogLong.value && !groupExpanded.value) {
    return text.slice(-(LOG_MAX_LENGTH * LOG_MAX_LINES));
  }
  return text;
 });
 const groupLogMeta = computed(() => groupLogLong.value ? {expanded: groupExpanded.value} : null);
 function toggleGroupLog() {
  if (selectedLog.value) props.expandedLogs?.toggle(selectedLog.value.id);
 }
 function onContextMenu() {
  if (props.message.type == 'log-group') {
    if (selectedLog.value) emit('open-menu', selectedLog.value, groupLogMeta.value);
  } else {
    emit('open-menu', props.message, logMeta.value);
  }
 }
 </script>
 <template>
-  <div :class="chatEntryClass(message)" @contextmenu.prevent="$emit('open-menu', message, logMeta)" @click="handleClick(message)">
+  <div :class="chatEntryClass(message)" @contextmenu.prevent="onContextMenu" @click="handleClick(message)">
    <template v-if="message.type == 'message' || message.type == null">
      <div v-if="message.sender != 'user'" class="sender">{{message.sender}}</div>
@@ -135,6 +240,28 @@ watch(() => [props.message.format, props.message.content], ([newFormat, newConte
      <a v-if="logLong" class="log-toggle" @click.prevent="$emit('toggle-log')">{{logExpanded ? '(collapse)' : '(expand)'}}</a>
    </template>
    <template v-if="message.type == 'log-group'">
      <div class="log-icons">
        <button
          v-for="log in logs"
          :key="log.id"
          type="button"
          class="icon"
          :class="{ highlight: selectedLog && log.id == selectedLog.id }"
          :title="(log.sender || '') + (log.subkind ? ' [' + log.subkind + ']' : '')"
          @click="selectedLogId = log.id"
        >
          <component :is="logIcon(log.subkind)" />
        </button>
      </div>
      <div v-if="selectedLog" class="log-preview">
        <div class="log-sender">{{selectedLog.sender}}<span v-if="selectedLog.subkind"> [{{selectedLog.subkind}}]</span></div>
        <div class="content" :class="{ clipped: !groupExpanded }">{{groupLogContent}}</div>
        <a v-if="groupLogLong" class="log-toggle" @click.prevent="toggleGroupLog()">{{groupExpanded ? '(collapse)' : '(expand)'}}</a>
      </div>
    </template>
    <template v-if="message.type == 'thread-header'">
      <div class="thread-title">
        {{ message.threadId == null ? 'New Thread' : `Thread ${message.threadId}` }}
@@ -219,6 +346,67 @@ watch(() => [props.message.format, props.message.content], ([newFormat, newConte
  user-select: none;
 }
 /* === Log groups === */
 /* Completely transparent container; stacks icons above the preview and
   stretches to the full width of the chat */
 .log-group {
  align-self: stretch;
  width: 100%;
  padding: 0.6em 1.4em;
  background: transparent;
  display: flex;
  flex-direction: column;
  gap: 0.4em;
 }
 /* Square buttons (sized by the global `button.icon`) that wrap to the width,
   each line centered within the container */
 .log-icons {
  display: flex;
  flex-wrap: wrap;
  justify-content: center;
  gap: 0.3em;
 }
 .log-icons .highlight {
  color: var(--color-highlight);
  border-color: var(--color-highlight);
 }
 .log-preview {
  font-family: monospace;
  font-size: 80%;
  max-width: 70em;
 }
 .log-preview .log-sender {
  font-weight: bold;
 }
 .log-preview .content {
  white-space: pre-wrap;
  word-wrap: anywhere;
  width: fit-content;
  max-width: 70em;
  line-height: 1.3;
 }
 /* Clipping shows only the bottom lines: the content is bottom-aligned, so any
   vertical overflow is clipped off the top. ~6 lines at line-height 1.3. */
 .log-preview .content.clipped {
  display: flex;
  flex-direction: column;
  justify-content: flex-end;
  max-height: 7.8em;
  overflow: hidden;
 }
 .log-preview .log-toggle {
  cursor: pointer;
  user-select: none;
 }
 .resource {
  border-radius: var(--border-radius);
  border: 1px dashed var(--color-border);
@@ -1,6 +1,6 @@
 <script setup>
  import {
-    ref, watch,
+    ref, computed, watch,
    onMounted, onUnmounted,
  } from "vue";
@@ -30,6 +30,29 @@
    creatingThread,
  } = session;
  // Consecutive log messages collapse into a single "log-group" entry, which
  // Message.vue renders as a container of icons plus a preview.
  const displayEntries = computed(() => {
    const entries = chatEntries.value || [];
    const out = [];
    let group = null;
    for (const entry of entries) {
      if (entry.type == 'log') {
        if (!group) {
          group = { id: 'log-group-' + entry.id, type: 'log-group', logs: [] };
          out.push(group);
        }
        group.logs.push(entry);
      } else {
        group = null;
        out.push(entry);
      }
    }
    return out;
  });
  // UI state coupled to the document itself
  // Message ids of logs currently expanded
@@ -216,7 +239,7 @@
    <div class="row" style="overflow: auto;">
      <div class="column">
        <div class="messages-container" ref="chatContainerElem">
-          <div class="message-container" v-for="msg in chatEntries">
+          <div class="message-container" v-for="msg in displayEntries">
            <Message 
              :key="msg.id" 
              :message="msg" 
@@ -225,6 +248,7 @@
              :log-expanded="expandedLogs.has(msg.id)"
              @toggle-log="expandedLogs.toggle(msg.id)"
              :expanded-logs="expandedLogs"
              :cancellable="messageRunMap[msg.id]"
              @cancel="session.cancelMessageRun(msg.id)"
@@ -353,4 +377,4 @@
    height: 4em;
    cursor: pointer;
  }
-</style>
+</style>
Author	SHA1	Message	Date
Shivering Ocelot	ea938a1248	fix: drives	2026-10-14 02:02:31 -05:00
Shivering Ocelot	0d79b1f706	feat(backend): refactor websocket handler	2026-10-14 01:44:34 -05:00
hubdev	4e7be79926	feat(chat): collapse log messages	2026-10-14 03:31:05 +00:00
`@@ -1,2 +1,2 @@`
	`from app_mixins.chat import ChatMixin`	`from app_mixins.chat import ChatMixin`
	`from app_mixins.agent import AgentMixin`	`from app_mixins.agent import AgentMixin`