wip: responsiveness

backend/requirements.txt

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

backend/src/agent.py

@@ -5,7 +5,6 @@ import asyncio
 import traceback
 
 from temporary import generate_tmp_id
-from stream import Stream
 
 # Tool related classes must be moved to a different module
 
@@ -36,8 +35,9 @@ class Agent:
 
         # should be in expert init?
         self.tools = []
+        self.remotes = []
         for resource in self.resources:
-            if resource.kind != 'tool': continue
+            if resource.kind == 'tool':
                 if resource.value.get('type') == 'native' and 'name' in resource.value:
                     tool_cls = next((t for t in self.app.tool_classes if t.tool_name() == resource.value['name']), None)
                     if tool_cls:
@@ -45,6 +45,15 @@ class Agent:
                         spec = tool.get_spec().clone()
                         spec.name = resource.key
                         self.tools.append(spec)
+            elif resource.kind == 'remote':
+                value = resource.value or {}
+                if isinstance(value, str):
+                    value = {'prefix': value}
+                self.remotes.append({
+                    'name': resource.key,
+                    'prefix': value.get('prefix'),
+                })
+
     
     def _set_basic_info(self, id, name, user_id):
         self.id = id
@@ -56,6 +65,10 @@ class Agent:
         agent_run.close()
         raise NotImplementedError("Subclasses must implement run()")
 
+    # Async generator: constructs a run, executes it, and yields every message
+    # the run emits (raw inference.Message objects: role + content) in arrival
+    # order. No projection: no delta collapsing, no tool filtering, no dropping
+    # of reasoning/usage. Consumers route or accumulate as they see fit.
     async def prompt(self, instruction, input=None, **kwargs):
         run = AgentRun.create(
             self.app,
@@ -67,19 +80,48 @@ class Agent:
                 **kwargs,
             },
         )
-        out_stream = Stream(self.app)
 
-        def on_output_event(stream):
-            def on_chunk(chunk):
-                out_stream.feed(chunk)
-            stream.listen(on_chunk, lambda: None)
-        def on_done(data):
-            out_stream.close()
-        run.on('output', on_output_event)
+        queue = asyncio.Queue()
+
+        def on_inference(msg):
+            queue.put_nowait(('message', msg))
+
+        def on_done(_data):
+            queue.put_nowait(('done', None))
+
+        # Register listeners before scheduling the task. AgentRun.on replays
+        # past events of that name, so this both avoids a race and guarantees
+        # no early messages are lost.
+        run.on('inference', on_inference)
         run.on('done', on_done)
-        await self.run(run)
 
-        return out_stream
+        run.task = asyncio.create_task(self.run(run))
+
+        def on_task_done(task):
+            # Safety net: an agent that raises (or forgets run.close()) would
+            # otherwise leave the consumer awaiting forever.
+            queue.put_nowait(('task', task))
+
+        run.task.add_done_callback(on_task_done)
+
+        try:
+            while True:
+                kind, payload = await queue.get()
+                if kind == 'message':
+                    yield payload
+                elif kind == 'done':
+                    break
+                else:  # task
+                    if payload.cancelled():
+                        break
+                    exc = payload.exception()
+                    if exc:
+                        raise exc
+                    break
+        finally:
+            if not run.task.done():
+                run.task.cancel()
+
 
     async def received_message(self, message):
         chat = await self.app.get_chat(message.chat_id)
@@ -113,13 +155,28 @@ class Agent:
             },
         )
 
-        async def on_output_event(stream):
-            print("[agent] run output " + repr(stream))
+        async def on_inference(msg):
+            kwargs = {'content': msg.content}
+
+            if msg.role == 'assistant':
+                # kind and subkind are default
+                pass
+            elif msg.role == 'reasoning':
+                kwargs['kind'] = "log"
+                kwargs['subkind'] = "reasoning"
+            elif msg.role == 'tool':
+                kwargs['kind'] = "log"
+                kwargs['subkind'] = "tool-call"
+                kwargs['content'] = repr(msg.content)
+
+            # Unknown message
+            else: return
+
             new_message = await self.app.send_chat_message(
                 sender=self.id,
                 chat_id=message.chat_id,
                 thread_id=message.thread_id,
-                content=stream,
+                **kwargs
             )
             run.emit_event('chat-message', {
                 'messageId': new_message['id'],
@@ -127,41 +184,7 @@ class Agent:
                 'threadId': message.thread_id,
             })
 
-        async def on_reasoning_event(stream):
-            print("[agent] run output " + repr(stream))
-            new_message = await self.app.send_chat_message(
-                sender=self.id,
-                chat_id=message.chat_id,
-                thread_id=message.thread_id,
-                content=stream,
-                kind="log",
-                subkind="reasoning",
-            )
-            run.emit_event('chat-message', {
-                'messageId': new_message['id'],
-                'chatId': message.chat_id,
-                'threadId': message.thread_id,
-            })
-
-        async def on_toolcall_event(stream):
-            print("[agent] run output " + repr(stream))
-            new_message = await self.app.send_chat_message(
-                sender=self.id,
-                chat_id=message.chat_id,
-                thread_id=message.thread_id,
-                content=stream,
-                kind="log",
-                subkind="tool-call",
-            )
-            run.emit_event('chat-message', {
-                'messageId': new_message['id'],
-                'chatId': message.chat_id,
-                'threadId': message.thread_id,
-            })
-
-        await run.on_async('reasoning', on_reasoning_event)
-        await run.on_async('output', on_output_event)
-        await run.on_async('tool-call', on_toolcall_event)
+        await run.on_async('inference', on_inference)
 
         await self.run(run)
         return run
@@ -184,16 +207,21 @@ class Agent:
                 if r.kind == kind and (key is None or r.key == key)]
 
     def get_tools(self, run=None):
-        """Collect the tools available to this agent.
-
-        Returns the agent-configured tools (derived from its resources) plus
-        any remote tools referenced by the run input. When a run is given,
-        remote names in ``run.input['remotes']`` are resolved and appended.
-        """
-        tools = list(self.tools)
+        remotes = list(self.remotes)
         if run is not None and 'remotes' in run.input:
-            tools += self.app.getRemoteTools(self.user_id, run.input['remotes'])
-        return tools
+            remotes += list(run.input['remotes'])
+
+        tools = list(self.tools) + self.app.getRemoteTools(self.user_id, remotes)
+
+        # Dedupe by tool name (agent-configured vs run-input vs native collisions)
+        seen, result = set(), []
+        for tool in tools:
+            if tool.name in seen:
+                continue
+            seen.add(tool.name)
+            result.append(tool)
+
+        return result
 
 
 class DebugAgent(Agent):

backend/src/agents/expert.py

@@ -1,5 +1,6 @@
 import inference
 from agent import Agent, AgentField
+from tool import ToolSpec
 
 import asyncio
 
@@ -20,62 +21,47 @@ async def chat_completion_loop(app, run, messages, model_id=None, tools=None):
     tools = list(tools or [])
     execute_tools = run.input.get('tool_policy', 'auto') == 'auto'
 
-    tool_calls = []
-    output_stream = None
-    reasoning_stream = None
-    inference_end = asyncio.Event()
-
-    def on_output(stream):
-        nonlocal output_stream
-        output_stream = stream
-
-    def on_reasoning(stream):
-        nonlocal reasoning_stream
-        reasoning_stream = stream
-
-    run.on('output', on_output)
-    run.on('reasoning', on_reasoning)
-    run.on('tool-call', tool_calls.append)
-    run.on('inference-end', lambda _: inference_end.set())
-
     while True:
-        # Reset per-iteration state
-        output_stream = None
-        reasoning_stream = None
-        inference_end.clear()
-
-        await inference.chat_completion(
+        tool_calls = []
+        turn_response = False
+        generator = inference.chat_completion(
             app,
             model_id=model_id,
             messages=messages,
             tools=tools,
-            agent_run=run,
+            opts=run.input,
         )
 
-        await inference_end.wait()
-
-        content = output_stream and await output_stream.content_wait()
-        reasoning = reasoning_stream and await reasoning_stream.content_wait()
-
-        if content or reasoning:
-            msg = inference.LlamaMessage('assistant', content or "")
-            if reasoning:
-                msg.reasoning = reasoning
+        pre_len = len(messages)
+        async for msg in generator:
             messages.append(msg)
+            run.emit_event('inference', msg)
+            if msg.role == 'tool':
+                tool_calls.append(msg.content)
+            if msg.role == 'assistant':
+                turn_response = True
 
-        if execute_tools and tool_calls:
+        # Avoid infinite loops from server failures
+        if len(messages) == pre_len: break
+
+        cancel_tools = False
         for tc in tool_calls:
             tool = next((t for t in tools if t.name == tc.tool_name), None)
             if tool is None:
                 print(f"[agent] unknown tool name: {tc.tool_name}")
                 continue
+            if execute_tools:
+                # Only known 'non-callable' tool spec
+                if isinstance(tool, ToolSpec) and not tool.func:
+                    cancel_tools = True
+                    tc.tool = tool
+                else:
                     tc.result = await tool.execute_async(**tc.arguments)
-                messages.append(inference.LlamaMessage('tool', tc))
-            tool_calls.clear()
-            # Buffer time to let the provider rest before the next call
-            await asyncio.sleep(2)
-        elif content or not reasoning:
-            break
+            else:
+                tc.tool = tool
+
+        # If not executing tools, respond to end the turn
+        if not (tool_calls and execute_tools) and turn_response or cancel_tools: break
 
     run.close()
 
@@ -105,13 +91,9 @@ class ExpertAgent(Agent):
         # Provide the agent's default tools unless the run supplies its own.
         # Only agent-provided tools are executable within the loop.
         tools = run.input.get("tools", [])
-        if tools:
-            run.input['tool_policy'] = 'ignore'
-        else:
+        if run.input.get('tool_policy') != 'caller':
             tools += self.get_tools(run)
 
-        print("Final tool object: " + repr(run.input.get('tools', [])))
-
         # Build the initial message list
         messages = inference.build_messages(
             system_prompt=self.config.get("system_prompt", ""),

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
@@ -47,7 +51,8 @@ class App(ChatMixin, AgentMixin):
         ]
 
         from features.calc import Calc
-        self.tool_classes = [Calc]
+        from features.web import WebSearch, WebFetch
+        self.tool_classes = [Calc, WebSearch, WebFetch]
 
     # --- Connection Management (Synchronous) ---
 
@@ -235,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")
@@ -249,6 +257,8 @@ async def main():
         # Unblock websocket handlers parked in websocket.receive()
         for task in list(horde_app.ws_tasks):
             task.cancel()
+        # Stop channel pollers and any in-flight channel runs
+        horde_app.channels.shutdown()
 
     for sig in (signal.SIGINT, signal.SIGTERM):
         loop.add_signal_handler(sig, interrupt)

backend/src/app_mixins/agent.py

@@ -126,13 +126,21 @@ class AgentMixin:
 
     async def save_agent_resource(self, agent_id, kind, key, value):
         await db.AgentResource.update_or_create(
+            {'value': value},
             agent_id=agent_id,
             kind=kind,
             key=key,
-            value=value,
         )
         print(f"Saved agent resource: {agent_id} - {kind} - {key}")
 
+    async def delete_agent_resource(self, agent_id, kind, key):
+        await db.AgentResource.filter(
+            agent_id=agent_id,
+            kind=kind,
+            key=key,
+        ).delete()
+        print(f"Deleted agent resource: {agent_id} - {kind} - {key}")
+
     # --- Control & Actions ---
 
     async def get_agent_control(self, agent_id):
@@ -172,10 +180,47 @@ class AgentMixin:
             runs = [r for r in runs if r.user_id == user_id]
         return [r.serialize() for r in runs]
 
-    async def send_agent_prompt(self, agent_id, instruction, input=None, options={}):
+    async def prompt_agent(self, agent_id, instruction, input=None, options={}):
+        # Mirror agent.prompt's generator so run ownership and lifecycle stay
+        # inside the mixin: callers just iterate and receive the same raw
+        # inference.Message objects. get_agent_control is awaited lazily on first
+        # iteration, keeping the control object alive for the generator's
+        # lifetime.
+
+        def normalize_tool(t):
+            # this function makes a lot of sense in ToolSpec itself...
+            from tool import ToolSpec, ToolParameter
+            if isinstance(t, dict):
+                params = []
+
+                d = t.get('parameters', {})
+                required = d.get('required', [])
+                props = d.get('properties', {})
+
+                for key, p in props.items():
+                    params.append(ToolParameter(
+                        name=key,
+                        type=p.get('type'),
+                        description=p.get('description'),
+                        required=key in required,
+                    ))
+                return ToolSpec(
+                    name=t['name'],
+                    description=t['description'],
+                    parameters=params,
+                )
+            else:
+                return t
+
+        if 'tools' in options:
+            options = {**options}
+            options['tools'] = [
+                normalize_tool(t) for t in options['tools']
+            ]
+
         control = await self.get_agent_control(agent_id)
-        response = await control.prompt(instruction, input, **options)
-        return response
+        async for msg in control.prompt(instruction, input, **options):
+            yield msg
 
     async def call_agent_action(self, agent_id, action, args):
         control = await self.get_agent_control(agent_id)

backend/src/app_mixins/chat.py

@@ -13,6 +13,7 @@ def message_json(msg):
         "sender": msg.sender,
         "kind": msg.kind,
         "subkind": msg.subkind,
+        "format": msg.format,
         "content": msg.content,
         "threadId": msg.thread_id,
     }

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/features/web.py

@@ -0,0 +1,106 @@
+"""Web tools: online search and page fetching.
+
+Both tools subclass :class:`~backend.src.tool.Tool` and rely on the base
+class's automatic ``ToolSpec`` generation (docstring -> description,
+``run`` signature -> parameters).
+"""
+
+import re
+import urllib.parse
+import urllib.request
+from html import unescape
+
+from tool import Tool
+
+
+def _strip_html(html: str) -> str:
+    """Light HTML filter: drop scripts/styles, strip tags, collapse whitespace."""
+    html = re.sub(
+        r"<(script|style|noscript)[^>]*>.*?</\1>",
+        " ",
+        html,
+        flags=re.DOTALL | re.IGNORECASE,
+    )
+    text = re.sub(r"<[^>]+>", " ", html)
+    text = unescape(text)
+    return re.sub(r"\s+", " ", text).strip()
+
+
+class WebSearch(Tool):
+    """Search the web with DuckDuckGo. Returns titles, URLs and snippets."""
+
+    ENDPOINT = "https://html.duckduckgo.com/html/"
+    USER_AGENT = (
+        "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
+        "(KHTML, like Gecko) Chrome/124.0 Safari/537.36"
+    )
+
+    # Result anchors look like:
+    #   <a class="result__a" href="//duckduckgo.com/l/?uddg=...">Title</a>
+    _RESULT_RE = re.compile(
+        r'<a[^>]*class="[^"]*\bresult__a\b[^"]*"[^>]*href="(?P<href>[^"]+)"[^>]*>'
+        r"(?P<title>.*?)</a>",
+        re.DOTALL,
+    )
+    _SNIPPET_RE = re.compile(
+        r'<a[^>]*class="[^"]*\bresult__snippet\b[^"]*"[^>]*>(?P<snippet>.*?)</a>',
+        re.DOTALL,
+    )
+
+    def run(self, query: str, max_results: int = 5):
+        body = urllib.parse.urlencode({"q": query}).encode("utf-8")
+        req = urllib.request.Request(
+            self.ENDPOINT,
+            data=body,
+            method="POST",
+            headers={
+                "User-Agent": self.USER_AGENT,
+                "Content-Type": "application/x-www-form-urlencoded",
+                "Accept": "text/html",
+            },
+        )
+        with urllib.request.urlopen(req, timeout=15) as resp:
+            page = resp.read().decode("utf-8", errors="replace")
+
+        snippets = [m.group("snippet") for m in self._SNIPPET_RE.finditer(page)]
+
+        results = []
+        for i, match in enumerate(self._RESULT_RE.finditer(page)):
+            if len(results) >= max_results:
+                break
+            results.append({
+                "title": self._clean(match.group("title")),
+                "url": self._unwrap(match.group("href")),
+                "snippet": self._clean(snippets[i]) if i < len(snippets) else "",
+            })
+        return results
+
+    @staticmethod
+    def _clean(value: str) -> str:
+        return unescape(re.sub(r"<[^>]+>", "", value)).strip()
+
+    @staticmethod
+    def _unwrap(href: str) -> str:
+        # DDG wraps result links as //duckduckgo.com/l/?uddg=<url-encoded-target>
+        if "uddg=" in href:
+            target = urllib.parse.parse_qs(urllib.parse.urlparse(href).query).get(
+                "uddg", [""]
+            )[0]
+            if target:
+                return target
+        if href.startswith("//"):
+            return "https:" + href
+        return href
+
+
+class WebFetch(Tool):
+    """Fetch a web page and return its text content, lightly filtered."""
+
+    def run(self, url: str):
+        req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
+        with urllib.request.urlopen(req, timeout=15) as resp:
+            html = resp.read().decode("utf-8", errors="replace")
+        return _strip_html(html)
+
+
+__all__ = ["WebSearch", "WebFetch"]

backend/src/inference.py

@@ -4,70 +4,121 @@ import json
 import os
 import traceback
 import random
+import inspect
 from stream import Stream
-from tool import format_tools_for_engine, build_tool_prompt_prefix, ToolSpec, ToolCall
+from tool import Tool, ToolSpec, ToolCall
 
 #from llama_cpp import Llama
 from openrouter import OpenRouter
 
-VALID_ROLES = ['user', 'assistant', 'system', 'tool']
+# TODO: Collect usage data
+
+TEST_MOCK = False
+VALID_ROLES = ['user', 'assistant', 'system', 'tool', 'reasoning', 'usage']
 CONTEXT_WINDOW = 10_000
 
-# TODO: Rename to ChatMessage. Add methods for constructing from instruct/openai/etc formatted objects.
-class LlamaMessage:
-    def __init__(self, role, content, reasoning=None):
+# Background inference tasks are kept referenced here so they aren't GC'd mid-flight.
+_background_tasks = set()
+
+def spawn(coro):
+    task = asyncio.create_task(coro)
+    _background_tasks.add(task)
+    task.add_done_callback(_background_tasks.discard)
+    return task
+
+# These reflect "emitted messages" from any actor, individual turn parts, rather than entire turns.
+# role acts as a type and a sender simultaneously, and content is dynamic
+class Message:
+    def __init__(self, role, content):
         if role not in VALID_ROLES:
             raise Exception("message role must be one of: " + ",".join(VALID_ROLES))
         self.role = role
         self.content = content
-        self.reasoning = reasoning
 
     def to_dict(self):
-        """Serialize to the dict format expected by llama-cpp / ollama."""
-        d = {"role": self.role, "content": self.content}
-        if self.reasoning:
-            d["reasoning"] = self.reasoning
-        return d
+        return {"role": self.role, "content": self.content}
 
     def __repr__(self):
-        return f"LlamaMessage(role={self.role!r}, content={self.content!r}, reasoning={self.reasoning!r})"
+        return f"Message(role={self.role!r}, content={self.content!r})"
 
-class Models:
-    llama_8b = "meta/meta-llama-3-8b-instruct"
-    llama_70b = "meta/meta-llama-3-70b-instruct"
-    qwen3_8b = "lucataco/qwen3-vl-8b-instruct:39e893666996acf464cff75688ad49ac95ef54e9f1c688fbc677330acc478e11"
-    qwen3_235b = "qwen/qwen3-235b-a22b-instruct-2507"
+def format_tools(tools):
+    specs = []
+    for t in tools:
+        if isinstance(t, dict):
+            specs.append(t)
+        elif isinstance(t, ToolSpec):
+            specs.append(t.to_openai_dict())
+        elif isinstance(t, Tool):
+            specs.append(t.get_spec().to_openai_dict())
+    return specs
 
 def format_openai(messages):
     formatted = []
-    last_assistant_message = None
+    def last_assistant_message():
+        if formatted and formatted[-1]['role'] == 'assistant':
+            return formatted[-1]
+        else:
+            msg = {'role': 'assistant', 'content': ''}
+            formatted.append(msg)
+            return msg
+
+    # For debugging
+    _stream_count = 0
+    _badrole_count = 0
+    _badroles = set()
+
     for msg in messages:
+        content = msg.content
+        if isinstance(content, Stream):
+            _stream_count += 1
+            content = content.content
+
         if msg.role == 'tool':
-            if not last_assistant_message:
-                # Create one implicitly
-                last_assistant_message = {'role': 'assistant', 'content': ''}
-            tool_calls = last_assistant_message.get('tool_calls')
+            last = last_assistant_message()
+            tool_calls = last.get('tool_calls')
             if tool_calls == None:
                 tool_calls = []
-                last_assistant_message['tool_calls'] = tool_calls
+                last['tool_calls'] = tool_calls
+            tc = content
+            if isinstance(tc, dict):
+                tc = ToolCall(
+                    tool_name=tc['tool_name'],
+                    arguments=tc['arguments'],
+                    result=tc['result'],
+                    id=tc['id']
+                )
+                # Skip non-resolved calls
+                if not tc.result: continue
+                # TODO: Fix call id if missing
             tool_calls.append({
-                'id': msg.content.id,
+                'id': tc.id,
                 'type': 'function',
                 'function': {
-                    'name': msg.content.tool_name,
-                    'arguments': json.dumps(msg.content.arguments),
+                    'name': tc.tool_name,
+                    'arguments': json.dumps(tc.arguments),
                 },
             })
             formatted.append({
                 'role': 'tool',
-                'tool_call_id': msg.content.id,
-                'content': json.dumps(msg.content.result),
+                'tool_call_id': tc.id,
+                'content': json.dumps(tc.result),
             })
-        else:
-            fmsg = {'role': msg.role, 'content': msg.content}
+        elif msg.role == 'reasoning':
+            last = last_assistant_message()
+            reasoning = last.get('reasoning', '')
+            last['reasoning'] = reasoning + content
+        elif msg.role in ['user', 'assistant', 'system']:
+            fmsg = {'role': msg.role, 'content': content}
             formatted.append(fmsg)
-            if msg.role == 'assistant':
-                last_assistant_message = fmsg
+        else:
+            _badrole_count += 1
+            _badroles.add(msg.role)
+
+    _warnings = []
+    if _stream_count: _warnings.append(f"{_stream_count} streams in message list")
+    if _badrole_count: _warnings.append(f"{_badrole_count} messages with bad roles {_badroles}")
+    if _warnings: print(f"[inference] WARNING: " + ", ".join(_warnings))
+
     return formatted
 
 
@@ -81,9 +132,12 @@ def safe_queue(callback):
     queue = asyncio.Queue()
     loop = asyncio.get_running_loop()
     async def read_loop():
+        nonlocal callback
         while True:
             args = await queue.get()
-            await callback(*args)
+            task = callback(*args)
+            if inspect.isawaitable(task):
+                await task
     loop.create_task(read_loop())
 
     def send(*args):
@@ -116,10 +170,10 @@ def get_llama_cpp_model(app, model_file):
 # Generate a conversation based on agent inputs
 def build_messages(system_prompt="", instruction=None, input=None, messages=None):
     def normalize(message):
-        if isinstance(message, LlamaMessage):
+        if isinstance(message, Message):
             return message
         elif isinstance(message, dict):
-            return LlamaMessage(message['role'], message['content'])
+            return Message(message['role'], message['content'])
         else:
             raise Exception("Unsupported message type: " + repr(type(message)))
 
@@ -139,9 +193,9 @@ def build_messages(system_prompt="", instruction=None, input=None, messages=None
 
     # Add default messages to list
     if system_prompt:
-        messages.insert(0, LlamaMessage("system", system_prompt))
+        messages.insert(0, Message("system", system_prompt))
     if prompt:
-        messages.append(LlamaMessage("user", prompt))
+        messages.append(Message("user", prompt))
 
     return messages
 
@@ -151,47 +205,38 @@ async def chat_completion(app,
     messages=None,
     tools=[],
     opts={},
-    agent_run=None
 ):
-    # TESTING OVERRIDE
-    # return await mock_completion(app, agent_run)
     db_model = (await app.get_model(model_id)).value
 
     model_engine = db_model['engine']
     model_name = db_model['model']
 
-    if model_engine == "llama-cpp":
-        result = await local_chat_completion(app, messages, model=model_name, tools=tools, opts=opts, agent_run=agent_run)
-    elif model_engine == "openrouter":
-        result = await openrouter_chat_completion(app, messages, model=model_name, tools=tools, opts=opts, agent_run=agent_run)
-    else:
-        raise Exception("Unsupported model: " + repr(model))
+    print(f"[inference] chat completion: {model_engine} '{model_name}', {len(messages)} messages, {len(tools)} tools")
 
-    if agent_run:
-        return None
-    return result
+    if TEST_MOCK:
+        generator = mock_completion(app)
+    elif model_engine == "llama-cpp":
+        generator = local_chat_completion(app, messages, model=model_name, tools=tools, opts=opts)
+    elif model_engine == "openrouter":
+        generator = openrouter_chat_completion(app, messages, model=model_name, tools=tools, opts=opts)
+    else:
+        raise Exception("Unsupported model engine: " + repr(model_engine))
+
+    async for message in generator:
+        yield message
+
+    print(f"[inference] chat completion end: {model_engine} '{model_name}'")
 
 
 # llama.cpp
-async def local_chat_completion(app, messages, model, tools=[], opts={}, agent_run=None):
-    if not agent_run:
-        raise Exception("agent_run required")
+async def local_chat_completion(app, messages, model, tools=[], opts={}):
+    queue = asyncio.Queue()
+    loop = asyncio.get_running_loop()
+    emit = safe_queue(queue.put_nowait)
 
-    content_stream = None
-    async def message_loop(name, value=None):
-        nonlocal content_stream, agent_run
-        if name == 'output':
-            if content_stream == None:
-                content_stream = Stream(app)
-                agent_run.emit_event('output', content_stream)
-            content_stream.feed(value)
-        elif name == 'close':
-            if content_stream:
-                content_stream.close()
-            agent_run.emit_event('inference-end', {})
-    send_message = safe_queue(message_loop)
-
-    def worker():
+    async def worker():
+        stream = None
+        try:
             llm = get_llama_cpp_model(app, model)
 
             kwargs = {}
@@ -205,56 +250,59 @@ async def local_chat_completion(app, messages, model, tools=[], opts={}, agent_r
                 kwargs['repeat_penalty'] = repeat
 
             # Format tools for llama-cpp (OpenAI-compatible function format)
-        formatted_tools = format_tools_for_engine(tools, "llama-cpp") if tools else None
+            formatted_tools = format_tools(tools) if tools else None
             if formatted_tools:
                 kwargs['tools'] = formatted_tools
                 kwargs['tool_choice'] = 'auto'
 
             # With stream=True, returns an iterator that infers tokens on demand (openai compatible)
-        llm_output = llm.create_chat_completion(
+            iterator = await loop.run_in_executor(None, lambda: llm.create_chat_completion(
                 messages=[msg.to_dict() for msg in messages],
                 stream=True,
                 **kwargs
-        )
+            ))
 
-        for chunk in llm_output:
-            # Check for cancellation
-            if agent_run and agent_run.cancelled:
-                print(f"Cancelling local inference for run {agent_run.id}")
+            sentinel = object()
+            while True:
+                chunk = await loop.run_in_executor(None, next, iterator, sentinel)
+                if chunk is sentinel:
                     break
 
                 delta = chunk['choices'][0]['delta']
                 if 'content' in delta:
-                send_message('output', delta['content'])
-            # Handle tool_call deltas if the model emits them
+                    if stream is None:
+                        stream = Stream(app)
+                        emit(Message('assistant', stream))
+                    stream.feed(delta['content'])
                 if 'tool_calls' in delta:
-                # Forward tool call chunks as JSON strings for the consumer
-                send_message("tool-call", {
-                    "id": tc.get("id"),
-                    "function": tc.get("function"),
-                })
+                    for tc in delta['tool_calls']:
+                        emit(Message('tool', tc))
+        except Exception:
+            print(traceback.format_exc())
+        finally:
+            if stream: stream.close()
+            emit(None)
 
-        send_message('close')
+    spawn(worker())
 
-    thread = threading.Thread(target=worker)
-    if agent_run:
-        agent_run.thread = thread
-    thread.start()
-
-    if agent_run:
-        return None
-    return content_stream
+    while True:
+        message = await queue.get()
+        if message is None:
+            break
+        yield message
 
 
-async def openrouter_chat_completion(app, messages, model, tools=[], opts={}, agent_run=None):
-    if not agent_run:
-        raise Exception("agent_run required")
+async def openrouter_chat_completion(app, messages, model, tools=[], opts={}):
+    if not app.openrouter_client:
+        raise Exception("OPENROUTER_API_KEY environment variable is not set")
 
-    # TODO: Separate as InferenceRun class
+    queue = asyncio.Queue()
+
+    async def worker():
         content_stream = None
         reasoning_stream = None
-
         last_channel = None
+
         def open_channel(channel):
             nonlocal last_channel, content_stream, reasoning_stream
             if channel != last_channel:
@@ -264,33 +312,7 @@ async def openrouter_chat_completion(app, messages, model, tools=[], opts={}, ag
                 reasoning_stream = None
             last_channel = channel
 
-
-    async def message_loop(name, value=None):
-        nonlocal content_stream, reasoning_stream, agent_run
-        open_channel(name) # Hack: use the message as the channel name
-        if name == 'output':
-            if content_stream == None:
-                content_stream = Stream(app)
-                agent_run.emit_event('output', content_stream)
-            content_stream.feed(value)
-        if name == 'reasoning':
-            if reasoning_stream == None:
-                reasoning_stream = Stream(app)
-                agent_run.emit_event('reasoning', reasoning_stream)
-            reasoning_stream.feed(value)
-        if name == 'tool-call':
-            agent_run.emit_event('tool-call', value)
-        elif name == 'close':
-            if content_stream: content_stream.close()
-            if reasoning_stream: reasoning_stream.close()
-            agent_run.emit_event('inference-end', {})
-    emit = safe_queue(message_loop)
-
-    async def worker():
-        print("START OPENROUTER INFERENCE")
-        if not app.openrouter_client:
-            raise Exception("OPENROUTER_API_KEY environment variable is not set")
-
+        try:
             kwargs = {
                 'model': model,
                 'messages': format_openai(messages),
@@ -301,15 +323,15 @@ async def openrouter_chat_completion(app, messages, model, tools=[], opts={}, ag
             }
 
             # Pass tools in OpenAI-compatible format if provided
-        formatted_tools = format_tools_for_engine(tools, "openrouter") if tools else None
+            formatted_tools = format_tools(tools) if tools else None
             if formatted_tools:
                 kwargs['tools'] = formatted_tools
 
-        temp = agent_run.input.get("temperature")
+            temp = opts.get("temperature")
             if temp is not None:
                 kwargs['temperature'] = temp
 
-        reasoning = agent_run.input.get("reasoning_effort")
+            reasoning = opts.get("reasoning_effort")
             if reasoning is not None:
                 kwargs['reasoning'] = {"effort": reasoning}
 
@@ -320,19 +342,22 @@ async def openrouter_chat_completion(app, messages, model, tools=[], opts={}, ag
             tool_calls = []
 
             async for event in stream_response:
-            if agent_run and agent_run.cancelled:
-                stream_response.cancel()
-                break
-                
                 if event.choices and len(event.choices) > 0:
                     delta = event.choices[0].delta
 
-                if delta.content:
-                    emit('output', delta.content)
                     if delta.reasoning:
-                    emit('reasoning', delta.reasoning)
+                        open_channel('reasoning')
+                        if reasoning_stream is None:
+                            reasoning_stream = Stream(app)
+                            queue.put_nowait(Message('reasoning', reasoning_stream))
+                        reasoning_stream.feed(delta.reasoning)
+                    if delta.content:
+                        open_channel('output')
+                        if content_stream is None:
+                            content_stream = Stream(app)
+                            queue.put_nowait(Message('assistant', content_stream))
+                        content_stream.feed(delta.content)
                     if delta.tool_calls:
-                    print("[openrouter] tool call:" + repr(delta.tool_calls)) 
                         for tc in delta.tool_calls:
                             if tc.index >= len(tool_calls):
                                 tool_calls.append(ToolCall(tool_name="", arguments="", id=""))
@@ -346,36 +371,40 @@ async def openrouter_chat_completion(app, messages, model, tools=[], opts={}, ag
                                     tool_call.tool_name += tc.function.name
                                 if tc.function.arguments:
                                     tool_call.arguments += tc.function.arguments
-            else:
-                print("[openrouter] no event choices: " + repr(event))
+                if event.usage:
+                    usage = {}
+                    for key in ['completion_tokens', 'prompt_tokens', 'total_tokens', 'cost']:
+                        if hasattr(event.usage, key):
+                            value = getattr(event.usage, key)
+                            if value != None:
+                                usage[key] = value
+                    queue.put_nowait(Message('usage', usage))
+
+            if content_stream: content_stream.close()
+            if reasoning_stream: reasoning_stream.close()
 
             for tc in tool_calls:
                 try:
-                arguments = json.loads(tc.arguments)
-                tc.arguments = arguments
-                emit('tool-call', tc)
+                    tc.arguments = json.loads(tc.arguments)
+                    queue.put_nowait(Message('tool', tc))
                 except Exception:
                     print("[openrouter] Bad tool call arguments for " + repr(tc))
                     print(traceback.format_exc())
-
-        emit('close')
-        print("END OPENROUTER INFERENCE")
-
-    def handle_done(task):
-        try:
-            task.result()
-        except asyncio.CancelledError:
-            print("[openrouter] inference cancelled")
-        except Exception as e:
-            print(f"[openrouter] Worker error: {e}")
+        except Exception:
             print(traceback.format_exc())
+        finally:
+            queue.put_nowait(None)
 
-    task = asyncio.create_task(worker())
-    task.add_done_callback(handle_done)
-    agent_run.task = task
+    spawn(worker())
+
+    while True:
+        message = await queue.get()
+        if message is None:
+            break
+        yield message
 
 
-async def mock_completion(app, agent_run=None):
+async def mock_completion(app):
     VERBS = ['fix','scan','sync','deploy','parse','render','cache','trace',
         'merge','purge','index','probe','flush','bind','warm','drain']
     OBJECTS = ['nodes','packets','tokens','frames','shards','clusters',
@@ -383,20 +412,28 @@ async def mock_completion(app, agent_run=None):
        'handles','slots','cursors','segments']
     DELAY = 0.3
 
-    content_stream = Stream(app)
-    if agent_run:
-        agent_run.emit_event('output', content_stream)
+    queue = asyncio.Queue()
+    emit = safe_queue(queue.put_nowait)
 
-    async def content_task ():
+    async def worker():
+        stream = Stream(app)
+        emit(Message('assistant', stream))
+        try:
             await asyncio.sleep(DELAY)
-        content_stream.feed(random.choice(VERBS))
+            stream.feed(random.choice(VERBS))
             await asyncio.sleep(DELAY)
-        content_stream.feed(f" {random.randint(1, 99)} ")
+            stream.feed(f" {random.randint(1, 99)} ")
             await asyncio.sleep(DELAY)
-        content_stream.feed(random.choice(VERBS))
+            stream.feed(random.choice(VERBS))
             await asyncio.sleep(DELAY)
-        content_stream.close()
-        agent_run.emit_event('inference-end', {})
+        finally:
+            stream.close()
+            emit(None)
 
-    asyncio.create_task(content_task())
-    if not agent_run: return content_stream
+    spawn(worker())
+
+    while True:
+        message = await queue.get()
+        if message is None:
+            break
+        yield message

backend/src/server/openai.py

@@ -1,8 +1,7 @@
 import json, time, traceback, asyncio, uuid
 from quart import Response, request
 from quart_auth import login_required, current_user
-from tool import ToolSpec, ToolParameter
-from agent import AgentRun
+from tool import ToolSpec, ToolParameter, ToolCall
 from utils import encode_json
 
 # ---------------------------------------------------------------------------
@@ -49,66 +48,72 @@ def route_tool_call(call, tool_names):
     return deferred
 
 
-async def execute_deferred(function_call):
+async def execute_deferred(tc):
     """Execute a deferred (native) tool call.
 
-    The output must be stored on the call itself (``function_call.result``).
+    The output must be stored on the call itself (``tc.result``).
 
-    TODO: resolve the ToolSpec for ``function_call.tool_name`` from the
+    TODO: resolve the ToolSpec for ``tc.tool_name`` from the
     agent's native tools and dispatch through the existing tool
     infrastructure (ToolSpec.execute_async), recording the result or error on
     the call. Deferred execution needs more design work; stubbed for now.
     """
-    pass
+    tc.result = await tc.tool.execute_async(**tc.arguments)
 
 
-def attach_run_handlers(run, tool_names):
-    """Wire the run's output events into a single queue.
+def messages_to_queue(generator, tool_names):
+    """Drive a prompt generator into a single queue.
 
     The queue receives ('content'|'reasoning'|'tool_call'|'deferred_call', data)
     tuples and terminates with ('eof', None). Tool calls are routed (client vs
-    native/deferred) here so both response modes share the same logic.
+    native/deferred) here so both response modes share the same logic. The
+    returned ``state`` dict tracks whether any tool call was observed, which
+    drives ``finish_reason``.
     """
     queue = asyncio.Queue()
     loop = asyncio.get_event_loop()
+    state = {'tool_call': False, 'error': None}
 
-    def on_output(stream):
-        async def on_chunk(chunk):
+    async def on_content(chunk):
         await queue.put(('content', chunk))
-        async def on_eof():
-            # on_done doesn't work?
-            await queue.put(('eof', None))
-        stream.listen(on_chunk, on_eof)
-
-    def on_reasoning(stream):
-        async def on_chunk(chunk):
+    async def on_reasoning(chunk):
         await queue.put(('reasoning', chunk))
-        async def on_eof(): pass
-        stream.listen(on_chunk, on_eof)
+    def on_eof(): pass
 
-    def on_tool_call(tool_call):
-        deferred = route_tool_call(tool_call, tool_names)
+    async def drain():
+        try:
+            async for msg in generator:
+                # assistant and reasoning are assumed to have a Stream
+                # tool is assumed to have a ToolCall
+                if msg.role == 'assistant':
+                    msg.content.listen(on_content, on_eof)
+                elif msg.role == 'reasoning':
+                    msg.content.listen(on_reasoning, on_eof)
+                elif msg.role == 'tool':
+                    deferred = route_tool_call(msg.content, tool_names)
                     msg_type = 'deferred_call' if deferred else 'tool_call'
-        loop.create_task(queue.put((msg_type, tool_call)))
+                    state['tool_call'] = True
+                    await queue.put((msg_type, msg.content))
+                elif msg.role == 'usage':
+                    await queue.put(('usage', msg.content))
+        except Exception as e:
+            # Surface the failure to the consumer via the eof drain.
+            state['error'] = e
+        finally:
+            await queue.put(('eof', None))
 
-    def on_done(output):
-        loop.create_task(queue.put(('eof', None)))
-
-    run.on('output', on_output)
-    run.on('reasoning', on_reasoning)
-    run.on('tool-call', on_tool_call)
-    run.on('done', on_done)
-
-    return queue
+    loop.create_task(drain())
+    return queue, state
 
 
-def format_chunk(agent_id, delta, finish_reason=None):
+def format_chunk(agent_id, delta, finish_reason=None, usage=None):
     """Build one OpenAI-format streaming chunk body."""
     return {
         "id": "chatcmpl-" + str(time.time()),
         "object": "chat.completion.chunk",
         "created": int(time.time()),
         "model": agent_id,
+        "usage": usage,
         "choices": [{"delta": delta, "index": 0, "finish_reason": finish_reason}]
     }
 
@@ -124,8 +129,8 @@ def register(server, app):
 
         # Extract required fields
         agent_id = data.get("model")
-        openai_messages = data.get("messages", [])
-        openai_tools = data.get("tools", [])
+        openai_messages = data.get("messages") or []
+        openai_tools = data.get("tools") or []
         stream_mode = data.get("stream", False)
 
         if not agent_id or not openai_messages:
@@ -164,20 +169,36 @@ def register(server, app):
                         # message carries the output.
                         call.result = content
                     entry['resolved'] = True
-                messages.append({'role': 'tool', 'content': call.result, 'tool_call_id': call.id})
+                messages.append({'role': 'tool', 'content': call})
                 continue
 
             if role == 'assistant' and msg.get('tool_calls'):
-                # Rebuild the assistant's tool call turn from the cache so
-                # unknown or expired calls are silently dropped and the agent
-                # receives the original call objects, which carry their own
-                # output once resolved.
+                # Rebuild the assistant's tool call turn from the request.
+                # Calls already in the cache (persistent conversation) are
+                # reused as-is. Client-declared function calls carry their
+                # full payload on the wire, so they are reconstructed from
+                # the message when the cache is gone. Server-originated
+                # (deferred) calls only ever sent an id + type, so they can
+                # only be recovered from the cache and are dropped otherwise.
                 calls = []
                 for tc in msg['tool_calls']:
-                    entry = tool_call_cache.get(tc.get('id', ''))
-                    if entry is None:
-                        continue
+                    call_id = tc.get('id', '')
+                    entry = tool_call_cache.get(call_id)
+                    if entry is not None:
                         calls.append(entry['call'])
+                        continue
+
+                    func = tc.get('function')
+                    if tc.get('type') == 'function' and func:
+                        call = ToolCall(
+                            id=call_id,
+                            tool_name=func.get('name', ''),
+                            arguments=json.loads(func.get('arguments') or '{}'),
+                        )
+                        # Seeding the cache lets the following role='tool'
+                        # message resolve the call and attach its result.
+                        cache_tool_call(call, deferred=False)
+                        calls.append(call)
                 messages.append({
                     'role': 'assistant',
                     'content': content if type(content) == str else None,
@@ -226,30 +247,30 @@ def register(server, app):
         # tool and gets deferred.
         tool_names = {spec.name for spec in tools}
 
-        # --- Get agent control and create AgentRun directly ---
+        # --- Get agent control and drive the run through the shared prompt generator ---
         control = await app.get_agent_control(agent_id)
 
-        run = AgentRun.create(
-            app,
-            agent_id=agent_id,
-            user_id=user_id,
-            input={
-                "messages": messages,
-                "tools": tools,
+        # The OpenAI inputs ride as kwargs so they land in run.input, which
+        # ExpertAgent.run reads ("tools", "tool_policy", "messages").
+        generator = control.prompt(
+            None,
+            input=None,
+            messages=messages,
+            tools=tools,
             # Never execute tools inside the run: emit them instead so
             # the handler can send client tools back and defer native
             # ones.
-                "defer_tool_execution": True,
-            },
+            tool_policy="ignore",
         )
-        print("run input: " + repr(run.input))
 
-        queue = attach_run_handlers(run, tool_names)
-        await control.run(run)
+        queue, state = messages_to_queue(generator, tool_names)
+
+        usage = {'completion_tokens': 0, 'prompt_tokens': 0, 'total_tokens': 0, 'cost': 0}
 
         # --- Streaming ---
         if stream_mode:
             async def generate():
+                nonlocal usage
                 try:
                     tool_call_count = 0
                     while True:
@@ -281,19 +302,26 @@ def register(server, app):
                                 "type": "tool",
                             }]}
 
+                        elif msg_type == 'usage':
+                            usage = data
+
                         elif msg_type == 'eof':
+                            if state['error']:
+                                raise state['error']
                             # Determine finish_reason — was there a tool call?
-                            finish_reason = "tool_calls" if any(
-                                e['name'] == 'tool_call' for e in run.events
-                            ) else "stop"
+                            finish_reason = "tool_calls" if state['tool_call'] else "stop"
                             yield f"data: {json.dumps(format_chunk(agent_id, {}, finish_reason))}\n\n"
                             break
 
-                        if delta:
-                            yield f"data: {json.dumps(format_chunk(agent_id, delta))}\n\n"
                         else:
                             raise Exception(f"Unrecognized msg_type: {msg_type}. {repr(data)}")
 
+                        if delta:
+                            yield f"data: {json.dumps(format_chunk(agent_id, delta))}\n\n"
+
+                    # Usage
+                    yield f"data: {json.dumps(format_chunk(agent_id, {}, finish_reason, usage))}\n\n"
+
                 except Exception as e:
                     print("[openai] ERROR: " + str(e))
                     print(traceback.format_exc())
@@ -309,6 +337,8 @@ def register(server, app):
                 while True:
                     msg_type, data = await queue.get()
                     if msg_type == 'eof':
+                        if state['error']:
+                            raise state['error']
                         break
                     elif msg_type == 'content':
                         content += data
@@ -327,6 +357,8 @@ def register(server, app):
                         # Deferred calls stay opaque to the client: only the
                         # call id and type go on the wire.
                         tool_calls.append({"id": data.id, "type": "tool"})
+                    elif msg_type == 'usage':
+                        usage = data
                     else:
                         raise Exception(f"Unrecognized msg_type: {msg_type}. {repr(data)}")
             except Exception as e:
@@ -346,6 +378,7 @@ def register(server, app):
                 "object": "chat.completion",
                 "created": int(time.time()),
                 "model": agent_id,
+                "usage": usage,
                 "choices": [{
                     "message": message,
                     "finish_reason": finish_reason,

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

@@ -1,7 +1,6 @@
-import json, asyncio
+import json, asyncio, traceback
 from quart import websocket
 from utils import encode_json
-from stream import Stream      # isinstance check in send-agent-prompt
 
 def register(server, app, auth_manager):
     @server.websocket('/ws')
@@ -40,10 +39,14 @@ def register(server, app, auth_manager):
 
         send_queue = asyncio.Queue()
         async def sender_impl():
+            # This thread MUST NOT crash
             nonlocal send_queue
             while True:
+                try:
                     msg = await send_queue.get()
                     await websocket.send(encode_json(msg))
+                except Exception as err:
+                    print(traceback.format_exc())
         async def send(msg):
             await send_queue.put(msg)
 
@@ -65,13 +68,54 @@ 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:
+                asyncio.create_task(handle_request(
+                    app, user_id, session_connections, send, await websocket.receive()
+                ))
+
+        finally:
+            print("User logged out:", user_id)
+            sender_task.cancel()
+            if ws_task in app.ws_tasks:
+                app.ws_tasks.remove(ws_task)
+            # Unregister the event listener: the send closure would
+            # otherwise keep queueing events for a dead socket forever
+            if stop_listening:
+                stop_listening()
+            # All remotes served by this session are disconnected with it.
+            # disconnectRemote is idempotent, so remotes that were already
+            # disconnected manually (possibly from another session) are safe.
+            for remotename in list(session_connections):
+                app.disconnectRemote(user_id, remotename)
+            app.remove_user_connection(user_id, send_queue)
+
+# Error handling
+async def handle_request(app, user_id, session_connections, send, req):
+    try:
+        req = json.loads(req)
+        await process_request(app, user_id, session_connections, send, req)
+
+    except Exception as err:
+        action = req.get('action') if isinstance(req, dict) else None
+        nonce  = req.get('nonce') if isinstance(req, dict) else None
+
+        # Maintainers
+        print(f"[ws] req error @{user_id} {action!r}: {err!r}")
+        print(traceback.format_exc())
+        # Client — nonce makes this routable to the caller's promise/callback
+        try:
+            await send({'error': str(err), 'nonce': nonce})
+        except Exception:
+            print(traceback.format_exc())
+
+# Actual processing
+async def process_request(app, user_id, session_connections, send, req):
+    if 'action' not in req:
         await send({'error': 'no action'})
-                    continue
+        return
     action = req['action']
     body = req.get('body', {})
     nonce = req.get('nonce', None)
+    print(f"[ws] req @{user_id} {action}")
 
     async def agent_id_from_body (allow_none=False):
         if 'agentId' in body:
@@ -84,7 +128,7 @@ def register(server, app, auth_manager):
             raise Exception("field 'agentId' or 'agentname' required")
 
     if action == "ping":
-                    await send({'message': "pong"})
+        await send({'message': "pong", 'nonce': nonce})
 
     elif action == "create-agent":
         name = body['name']
@@ -150,6 +194,10 @@ def register(server, app, auth_manager):
         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),
@@ -268,7 +316,7 @@ def register(server, app, auth_manager):
     elif action == "update-chat-message":
         message = await app.update_chat_message(
             message_id=body['messageId'],
-                        content=body['content'],
+            content=body.get('content'),
             kind=body.get('kind'),
             subkind=body.get('subkind'),
             format=body.get('format'),
@@ -289,23 +337,30 @@ def register(server, app, auth_manager):
         )
         await send({'message': "chat thread created", 'threadId': thread_id, 'nonce': nonce})
 
-                elif action == "send-agent-prompt":
+    elif action == "prompt-agent":
         if 'instruction' not in body and 'input' not in body:
-                        await send({'error': "instruction or input required"})
-                        continue
+            await send({'error': "instruction or input required", 'nonce': nonce})
+            return
 
-                    response = await app.send_agent_prompt(
+        # 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', {})
         )
-                    if isinstance(response, Stream):
-                        if body.get('stream'):
-                            response = {"streamId": response.id}
-                        else:
-                            response = await response.content_wait()
-                    await send({'message': "prompt sent", 'response': response, 'nonce': nonce})
+        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']
@@ -322,18 +377,3 @@ def register(server, app, auth_manager):
 
     else:
         await send({'error': 'unknown action: ' + action, 'nonce': nonce})
-        finally:
-            print("User logged out:", user_id)
-            sender_task.cancel()
-            if ws_task in app.ws_tasks:
-                app.ws_tasks.remove(ws_task)
-            # Unregister the event listener: the send closure would
-            # otherwise keep queueing events for a dead socket forever
-            if stop_listening:
-                stop_listening()
-            # All remotes served by this session are disconnected with it.
-            # disconnectRemote is idempotent, so remotes that were already
-            # disconnected manually (possibly from another session) are safe.
-            for remotename in list(session_connections):
-                app.disconnectRemote(user_id, remotename)
-            app.remove_user_connection(user_id, send_queue)

backend/src/tool.py

@@ -149,7 +149,7 @@ class Tool:
     
     @classmethod
     def tool_name(cls):
-        return ''.join(['_'+c.lower() if c.isupper() else c for c in cls.__name__]).lstrip('_')
+        return ''.join(['-'+c.lower() if c.isupper() else c for c in cls.__name__]).lstrip('-')
 
 
     def _generate_spec(self):
@@ -218,24 +218,17 @@ class ToolCall:
         is_error: Whether the result represents an error.
     """
 
-    def __init__(self, tool_name, arguments={}, id=None):
+    def __init__(self, tool_name, arguments={}, id=None, result=None):
         self.id = id or ""
         self.tool_name = tool_name
         self.arguments = arguments
 
-        self.result = None
+        self.result = result
 
     def __repr__(self):
         id_prefix = self.id + ', ' if self.id else ''
         return f"ToolCall({id_prefix}tool={self.tool_name!r}, args={self.arguments!r})"
 
-    def call_message(self):
-        raise Exception("TODO: Llama message for tool call")
-
-    def result_message(self):
-        from inference import LlamaMessage
-        return LlamaMessage("tool", self.result)
-
 
 # ---------------------------------------------------------------------------
 # Helpers to format tool specs per engine

backend/src/utils.py

@@ -1,14 +1,39 @@
 import json
 
 from stream import Stream
-from tool import ToolSpec
+from tool import ToolSpec, ToolCall
+from inference import Message
+
+# TODO: Move to module 'format'
+def encode_value(value):
+    if isinstance(value, ToolSpec):
+        return value.to_openai_dict()
+    elif isinstance(value, Message):
+        return {'role': value.role, 'content': encode_value(value.content)}
+    elif isinstance(value, ToolCall):
+        json = {
+            'id': value.id,
+            'tool_name': value.tool_name,
+            'arguments': encode_value(value.arguments),
+        }
+        if value.result != None:
+            json['result'] = encode_value(value.result),
+        return json
+    elif isinstance(value, Stream):
+        if value.open:
+            return {'streamId': value.id}
+        else:
+            return value.content
+    else:
+        return value
 
 def encode_json(value):
     def encode_unknown(value):
-        if isinstance(value, ToolSpec):
-            return value.to_openai_dict()
-        elif isinstance(value, Stream):
-            return {'streamId': value.id}
+        # All complex values are transformed to a dict
+        value = encode_value(value)
+        # otherwise simply show the repr
+        if isinstance(value, dict):
+            return value
         else:
             return repr(value)
     return json.dumps(value, default=encode_unknown)

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

frontend/src/assets/main.css

@@ -95,3 +95,13 @@ input:disabled, textarea:disabled, select:disabled {
 .with-sidebar-expanded .when-sidebar-collapsed {
   display: none;
 }
+
+/* Content */
+h1, h2, h3, h4, h5, h6 {
+  margin: 1em 0 0.5em;
+  font-weight: bold;
+}
+
+p, ul, il {
+  margin: 0.5em 0;
+}

frontend/src/chat.js

@@ -545,6 +545,33 @@ export class ChatSession {
     this.creatingThread.value = Boolean(fork);
   }
 
+  // Sets the render format of a message ("markdown", or "" for plain text).
+  // Omit `content` so the backend leaves the existing content untouched.
+  setMessageFormat (message, format) {
+    backend.sendAction("update-chat-message", {
+      messageId: message.id, format,
+    }).then(response => {
+      if (response && response.error) {
+        ui.alert("Error updating message format: " + response.error);
+      }
+    });
+
+    // No update event is broadcast (matching the content-edit path above), so
+    // patch the source message and rebuild the entries locally. Other sessions
+    // and stale views won't reflect this until reload — which is correct, since
+    // the format is persisted. If we ever broadcast edits, flip this to rely on
+    // the incoming update instead.
+    let thread = this.getThread(message.threadId);
+    let existing = thread.messages.find(msg => msg.id == message.id);
+    if (existing) {
+      existing.format = format;
+    } else {
+      console.error("Can't set format locally?", {messageId: message.id, thread});
+    }
+
+    this.computeChatEntries();
+  }
+
   // Adds a bulk of messages to the message list, in their correct
   // chronological position and replacing duplicates
   addMessageBulk (messages) {

frontend/src/components/JsonEditor.vue

@@ -1,5 +1,6 @@
 <script setup>
   import { ref, computed, watch } from 'vue'
+  import * as ui from '../ui'
 
   const props = defineProps({
     /** JSON schema object describing the input structure */
@@ -55,6 +56,17 @@
     };
   }
 
+  async function addField(field) {
+    let key = await ui.input("Enter new property name");
+    if (!key) return;
+
+    // Never mutate the field object in place.
+    currentValue.value = {
+      ...(currentValue.value || {}),
+      [key]: null
+    };
+  }
+
   const computedFields = computed(() => {
     let fields = [];
     let skip = {};

frontend/src/components/Message.vue

@@ -4,13 +4,17 @@ import { marked } from "marked";
 import * as backend from '../backend';
 marked.use({
   renderer: {
-    code (code, infostring, escaped) {
-      let langElem = infostring ? 
-        `<div class="language">${infostring}</div>` : "";
+    code ({lang, text}) {
+      let langElem = lang ? 
+        `<div class="code-lang">${lang}</div>` : "";
+
+      const div = document.createElement('div');
+      div.textContent = text;
+      const escaped = div.innerHTML;
 
       return `<div class="code">
         ${langElem}
-        <div class="content">${code}</div>
+        <div class="code-content">${escaped}</div>
       </div>`;
     }
   }
@@ -111,7 +115,7 @@ watch(() => [props.message.format, props.message.content], ([newFormat, newConte
     <template v-if="message.type == 'message' || message.type == null">
       <div v-if="message.sender != 'user'" class="sender">{{message.sender}}</div>
 
-      <div v-if="format == 'markdown'" v-html="marked.parse(message.content)" class="content"></div>
+      <div v-if="format == 'markdown'" v-html="marked.parse(message.content)" class="md-content" style="white-space: collapse;"></div>
       <div v-else class="content">{{content}}</div>
     </template>
 
@@ -189,14 +193,6 @@ watch(() => [props.message.format, props.message.content], ([newFormat, newConte
   margin-right: 2em;
 }
 
-.message .content {
-  white-space: preserve;
-}
-
-.message ul, .message ol {
-  white-space: normal;
-}
-
 .message .sender,
 .note .sender {
   font-weight: bold;
@@ -288,4 +284,38 @@ watch(() => [props.message.format, props.message.content], ([newFormat, newConte
   font-style: normal;
   font-weight: bold;
 }
+
+.md-content {
+  white-space: collapse;
+  width: fit-content;
+  max-width: 50em;
+  line-height: normal;
+  word-wrap: anywhere;
+}
+
+</style>
+
+<style>
+/* Code blocks are generated at runtime, these styles cannot be scoped */
+
+.code {
+  background-color: var(--color-surface-input);
+  border-radius: var(--border-radius);
+  margin: .5em 0;
+  padding: .5em;
+}
+
+.code-lang {
+  padding-bottom: .5em;
+  margin-top: -.5em;
+  font-weight: bold;
+  user-select: none;
+}
+
+.code-content {
+  white-space: pre;
+  font-family: monospace;
+  overflow-x: auto;
+  line-height: 1.6rem;
+}
 </style>

frontend/src/components/TitleBar.vue

@@ -23,8 +23,8 @@
 <style scoped>
   :slotted(h1) {
     font-size: 160%;
-    font-weight: bold;
     padding: 0 0.4em;
+    margin: 0;
   }
 
   .main-header {

frontend/src/views/Agent.vue

@@ -10,6 +10,8 @@
   import * as backend from '../backend';
   import * as ui from '../ui';
 
+  import * as iconoir from '@iconoir/vue';
+
   const props = defineProps({
     id: {
       type: String,
@@ -32,8 +34,7 @@
   });
 
   // UI State
-  const addingTool = ref(null);
-  const addingResource = ref(null);
+  const targetResource = ref(null);
 
   // === State Management === //
 
@@ -92,8 +93,7 @@
 
     // Replace with a backend call
     backend.sendAction('get-agent-resources', {agentname}).then(res => {
-      tools.value = res.resources.filter(r => r.kind == 'tool');
-      resources.value = res.resources.filter(r => !['tool', 'config'].includes(r.kind));
+      resources.value = res.resources;
     });
   }
 
@@ -157,15 +157,36 @@
     }
 
     promptHistory.value.push(historyEntry);
-    let res = await backend.sendAction("send-agent-prompt", payload);
 
-    if (res.error) {
-      historyEntry.error = res.error;
+    // prompt-agent is a multiple-result action (like read-stream): one packet
+    // per generated inference.Message, followed by a terminal eof packet. The
+    // callback runs once per packet. Assistant/reasoning content is an open
+    // stream ({streamId}); we read it live via read-stream.
+    historyEntry.output = "";
+    try {
+      await backend.sendAction("prompt-agent", payload, (packet, deferred) => {
+        if (packet.eof) {
+          deferred.resolve(packet);
         } else {
-      historyEntry.output = res.response;
+          const msg = packet.data;
+          if (msg.role !== 'assistant') return;
+          const content = msg.content;
+          if (content && content.streamId != null) {
+            const stream = backend.readStream(content.streamId);
+            stream.on('data', () => {
+              historyEntry.output = stream.partialContent;
+            });
+          } else if (typeof content === 'string') {
+            historyEntry.output += content;
           }
+        }
+      });
+    } catch (e) {
+      historyEntry.error = e.message || String(e);
+    } finally {
       historyEntry.loading = false;
     }
+  }
 
   function promptKeyDown (event, stage) {
     if (event.key == "Enter" && event.ctrlKey) {
@@ -237,34 +258,66 @@
     });
   }
 
-  async function saveAddingTool() {
-    let resource = {
-      kind: 'tool',
-      key: addingTool.value.name,
-      value: JSON.parse(addingTool.value.value),
-    };
-    addingTool.value = null;
+  async function updateResourceKey () {
+    let newKey = await ui.input("Enter new resource key");
+    if (!newKey || newKey == targetResource.value.key) return;
 
-    let res = await backend.sendAction("save-agent-resource", {
-      agentId: agent.value.id,
-      ...resource
-    });
-    tools.value.push(resource);
+    // Guard collisions, backend won't
+    for (let r of resources.value) {
+      if (r.kind == targetResource.value.kind && r.key == newKey) {
+        ui.alert(`Conflict: ${targetResource.value.kind} '${newKey}'`);
+        return;
+      }
     }
 
-  async function saveAddingResource() {
-    let resource = {
-      kind: addingResource.value.kind,
-      key: addingResource.value.key,
-      value: JSON.parse(addingResource.value.value),
-    };
-    addingResource.value = null;
+    // Keys can't be updated directly, we move the resource
+    await backend.sendAction("delete-agent-resource", {
+      agentId: agent.value.id,
+      ...targetResource.value
+    });
 
+    targetResource.value.key = newKey;
+    await backend.sendAction("save-agent-resource", {
+      agentId: agent.value.id,
+      ...targetResource.value,
+    });
+  }
+
+  async function saveResource () {
     let res = await backend.sendAction("save-agent-resource", {
       agentId: agent.value.id,
-      ...resource
+      ...targetResource.value
     });
-    resources.value.push(resource);
+
+    ui.alert(`Saved ${targetResource.value.kind} '${targetResource.value.key}'`);
+
+    if (targetResource.value.isNew) {
+      delete targetResource.value.isNew;
+      resources.value.push(targetResource.value);
+    }
+  }
+
+  async function deleteResource () {
+    if (!await ui.confirm(`Delete ${targetResource.value.kind} '${targetResource.value.key}'?`)) return;
+
+    let res = await backend.sendAction("delete-agent-resource", {
+      agentId: agent.value.id,
+      ...targetResource.value
+    });
+
+    let index = resources.value.indexOf(targetResource.value);
+    if (index >= 0) resources.value.splice(index, 1);
+    targetResource.value = null;
+  }
+
+  async function createResource () {
+    let kind = await ui.input("Enter resource kind");
+    if (!kind) return ui.alert('Cancel new resource');
+
+    let key = await ui.input("Enter resource key");
+    if (!kind) return ui.alert('Cancel new resource');
+
+    targetResource.value = {kind, key, value: {}, isNew: true};
   }
 </script>
 
@@ -327,36 +380,31 @@
       <JsonEditor :schema="config.schema" v-model="config.object" />
       <button type="submit" @click="saveConfig()">Save</button>
     </section>
-    <section>
-      <h3>Tools</h3>
-      <div v-for="(resource, index) in tools">
-        <div class="label">{{ resource.key }}</div>
-        <pre>{{ JSON.stringify(resource.value, null, 2) }}</pre>
-      </div>
-      <section v-if="addingTool">
-        <input v-model="addingTool.name" placeholder="name"><br>
-        <textarea v-model="addingTool.value" placeholder="{}"></textarea>
-        <br>
-        <button @click="saveAddingTool()">Save</button>
-        <button @click="addingTool=null">Cancel</button>
-      </section>
-      <div v-else><button @click="addingTool={}">Add Tool</button></div>
-    </section>
     <section>
       <h3>Resources</h3>
-      <div v-for="resource in resources">
-        <div class="label">{{ resource.kind }}:{{ resource.key }}</div>
-        <pre>{{ JSON.stringify(resource.value, null, 2) }}</pre>
+      <div class="row">
+        <section class="resource-detail column" v-if="targetResource">
+          <div>
+            <span class="resource-type">{{ targetResource.kind }}</span>
+            {{ targetResource.key }}
           </div>
-      <section v-if="addingResource">
-        <input v-model="addingResource.kind" placeholder="kind/domain"><br>
-        <input v-model="addingResource.key" placeholder="key"><br>
-        <textarea v-model="addingResource.value" placeholder="value"></textarea>
-        <br>
-        <button @click="saveAddingResource()">Save</button>
-        <button @click="addingResource=null">Cancel</button>
+          <JsonEditor :schema="{type:'object'}" v-model="targetResource.value" />
         </section>
-      <div v-else><button @click="addingResource={}">Add Resource</button></div>
+        <section class="resource-table">
+          <div class="resource-entry" v-for="resource in resources" @click="targetResource=resource">
+            <span class="label">{{ resource.kind }}</span> {{ resource.key }}
+          </div>
+        </section>
+      </div>
+      <div>
+        <template v-if="targetResource">
+          <button @click="updateResourceKey">Update Key</button>
+          <button @click="deleteResource">Delete</button>
+          <button @click="saveResource">Save</button>
+          <button @click="targetResource=null">Cancel</button>
+        </template>
+        <button v-else @click="createResource">Add Resource</button>
+      </div>
     </section>
   </main>
 </template>
@@ -398,4 +446,37 @@
     resize: vertical;
     height: 5em;
   }
+
+  .resource-table {
+    display: grid;
+    grid-template-columns: repeat(auto-fill, minmax(6em, 1fr));
+    height: fit-content;
+    flex: 1;
+
+    /*align-items: center;
+    gap: 0.5rem;
+    padding: 0.5rem 0.75rem;
+    border-bottom: 1px solid var(--color-border);*/
+  }
+  .resource-entry {
+    border: 1px solid var(--color-border);
+    border-radius: var(--border-radius);
+    background-color: var(--color-surface-strong);
+    padding: .4rem .6rem;
+    width: max-content;
+  }
+  .resource-entry:hover {
+    border-color: var(--color-highlight);
+  }
+
+  .resource-type {
+    font-family: monospace;
+    border: 1px solid var(--color-border);
+    border-radius: 3px;
+    padding: .1rem .3rem;
+  }
+
+  /* style fix */
+  .row > section { margin-left: 0; }
+  .row > section:last-child { margin-right: 0; }
 </style>

frontend/src/views/Chat.vue

@@ -130,6 +130,12 @@
       {value: 'pin', title: "Pin message"},
       {value: 'edit', title: "Edit in-place"},
       {value: 'edit-fork', title: "Edit in new thread"},
+      // ContextMenu.vue uses null for "dismissed", so the "None" entry can't
+      // carry a null value; "format:" (empty suffix) is the sentinel for "".
+      {title: "Set format", entries: [
+        {value: 'format:markdown', title: "Markdown"},
+        {value: 'format:', title: "None"},
+      ]},
     ];
 
     // Long logs can be expanded/collapsed
@@ -164,6 +170,12 @@
         }
         break;
       }
+      default: {
+        if (selected?.startsWith('format:')) {
+          session.setMessageFormat(message, selected.slice('format:'.length));
+        }
+        break;
+      }
 
     }
   }