Compare commits
3
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Shivering Ocelot
|
279861bdb8 | ||
|
Shivering Ocelot
|
9a85c61781 | ||
|
Shivering Ocelot
|
3238c059f6 |
No files matched your search
@@ -1,169 +0,0 @@
|
||||
# Adding Channel Connections (Telegram, WhatsApp, Discord, Slack) to the Hub Backend
|
||||
|
||||
## Purpose & Scope
|
||||
|
||||
This report maps the *design seams* in the current Hub backend against the shape of four chat-service APIs, and lays out the realistic integration options. It is deliberately analysis-first: we do not yet have a design for channels, and the goal is to decide where such a feature can live without distorting the existing agent/chat/run model.
|
||||
|
||||
Sourced from `docs/agent.md`, `docs/api.md`, `docs/chat.md`, `docs/remotes.md`, `backend/src/app.py`, `backend/src/database.py`, `backend/src/agent.py`, `backend/src/agents/expert.py`, `backend/src/app_mixins/agent.py`, `backend/src/app_mixins/chat.py`.
|
||||
|
||||
---
|
||||
|
||||
## 1. The Hub model we must fit into
|
||||
|
||||
The backend today has four concrete concepts that a channel feature must interact with. Understanding their contracts is what determines the integration options.
|
||||
|
||||
### 1.1 Agents
|
||||
|
||||
- An agent is created by a user, has a `kind` resolved to a Python class (`agents/expert.py`), and holds config in `data` (schema fields) plus resources.
|
||||
- Two entry points drive an agent (`docs/agent.md`):
|
||||
- **`prompt(instruction, input=None, **options)`** — a one-shot; returns an async generator of raw `inference.Message` objects (role + content). Implemented in `agent.py` and mirrored in `AgentMixin.prompt_agent`.
|
||||
- **`received_message(message)`** — the chat-driven path. It builds a `ChatView` from the message's thread, converts thread messages to LLM messages (roles derived from senders via `_role_for_message`), collects `remote` resources from the thread, creates an `AgentRun`, and streams output/reasoning/tool-call messages *back into the same thread* as new chat messages.
|
||||
- The agent object is explicitly ephemeral: "Everything durable (config, resources, chat membership) lives in the database or in resources." This matters — a channel feature must not assume the connection survives in an agent object.
|
||||
|
||||
### 1.2 Chats, Threads, Messages
|
||||
|
||||
- `Chat` is owned by a user (`database.Chat.user`), has messages and threads. `Thread` can be parented (forking) and carries denormalized summary fields (`first_message`, `last_message`, `message_count`).
|
||||
- `Message.sender` is either the literal string `'user'` or an **agent id**. Content may be a *stream reference* updated in place when the stream finishes.
|
||||
- Message `kind` ∈ {`message`, `log`, `note`, `resource`}; `resource` messages are declarations (e.g. remotes) and are **not** delivered to agents.
|
||||
- Chats may be **temporary** (in-memory, via `temporary.py`) or persistent. Temporary ids are distinguished by `tmp.is_tmp_id(...)` throughout the mixins — a branching pattern repeated everywhere (`get_agent`, `get_chat_messages`, `create_chat_thread`, etc.).
|
||||
|
||||
### 1.3 Runs
|
||||
|
||||
- `AgentRun` is the envelope for all agent activity: input dict (instruction/messages/tools/chat_id/thread_id/remotes), an event log with replay for late subscribers, output/reasoning streams, and cancellation. Runs are always in-memory (`self.tmp.runs`).
|
||||
- Events are broadcast to websocket clients as `run-event`.
|
||||
|
||||
### 1.4 Transports already present
|
||||
|
||||
- **Inbound user input** arrives solely through websocket actions (`send-chat-message`, `send-agent-prompt` — see `docs/api.md`), handled in `server/websocket.py` and dispatched into the mixins.
|
||||
- **Outbound** to the user is the websocket fan-out: `App.user_connections[user_id]` is a list of channels, and the chat mixin pushes packets (`notify_message_to_user`, `thread-update`, etc.) to them.
|
||||
- **Remotes** (`remote.py`, `docs/remotes.md`) are the closest existing analog to an external integration: connection-scoped, keyed by `(user_id, name)`, in-memory, torn down with the serving session, expanded into tools at run time by `App.getRemoteTools`. They are **inbound-less**: a remote exposes functions the *agent calls out to*; nothing external ever initiates a run.
|
||||
|
||||
### 1.5 Key structural observation
|
||||
|
||||
There is exactly **one private, user-scoped, session-bound external-integration mechanism today (remotes)**, and it is strictly *agent-tool* shaped. There is **no** notion of an external party initiating a conversation, no external-identity mapping, and no persistent outbound destination. Those three gaps are the whole problem.
|
||||
|
||||
---
|
||||
|
||||
## 2. What the four chat APIs actually look like
|
||||
|
||||
Grouped by the mechanics that matter for integration (inbound delivery, outbound send, credentials/scoping, threading model).
|
||||
|
||||
### 2.1 Telegram (Bot API)
|
||||
|
||||
- **Transport:** HTTPS REST to `api.telegram.org/bot<token>/…`. Inbound is either **long polling** (`getUpdates`) or a **webhook** (`setWebhook`) — either way we receive `update` JSON objects.
|
||||
- **Inbound shape:** `update.message.from.{id,username}`, `message.chat.{id,type}`, `message.text`, `message.message_id`, plus `callback_query` for inline buttons.
|
||||
- **Outbound:** `sendMessage` with `chat_id` + `text`; richer payloads (`sendPhoto`, etc.). Replies can be threaded via `reply_to_message_id`.
|
||||
- **Credentials:** a single **bot token**; one bot identity per token. No per-workspace OAuth — the mapping unit is *bot ↔ (many) chats*.
|
||||
- **Threading:** chats/groups are the container; topics exist in forum groups.
|
||||
- **Server model:** works from a normal always-on backend; long polling needs no public URL, webhooks need one.
|
||||
|
||||
### 2.2 WhatsApp (Meta Cloud API / Graph API)
|
||||
|
||||
- **Transport:** Graph API `POST /{phone-number-id}/messages` to send. Inbound is **webhooks only** — a GET verification handshake (hub.challenge) then POST payloads.
|
||||
- **Inbound shape:** `entry[].changes[].value.messages[]` (`from`, `id`, `text.body`, `timestamp`) plus `contacts[]` for profile names; status updates share the envelope.
|
||||
- **Outbound:** session messages within a **24-hour customer-service window**; outside it, pre-approved **template** messages are required. This is a real product constraint, not a transport detail.
|
||||
- **Credentials:** business phone number id + access token; per-number scoping. Third-party BSPs common.
|
||||
- **Threading:** 1:1 conversation keyed by the customer's phone number; no native threads.
|
||||
- **Server model:** requires a public webhook endpoint.
|
||||
|
||||
### 2.3 Discord
|
||||
|
||||
- **Transport:** **Gateway websocket** for events (`MESSAGE_CREATE`, `INTERACTION_CREATE`) + REST for sending (`POST /channels/{channel_id}/messages`). Interaction endpoints can also be HTTP webhooks.
|
||||
- **Inbound shape:** message object with `id`, `channel_id`, `guild_id`, `author.{id,username,bot}`, `content`, `referenced_message` for replies.
|
||||
- **Credentials:** **bot token**; requires declaring **Gateway intents**, and reading message *content* requires the privileged **MESSAGE_CONTENT intent**. A bot joins a guild/channel by invite.
|
||||
- **Threading:** channels and threads are first-class (Guild → channel → thread); DMs are a special channel type.
|
||||
- **Server model:** a *persistent outbound websocket* connection is mandatory (or an interaction webhook). This is a long-lived background task, not request/response.
|
||||
|
||||
### 2.4 Slack
|
||||
|
||||
- **Transport:** Inbound via **Events API** (HTTP webhook with `url_verification` challenge and HMAC **signing-secret** verification) or **Socket Mode** (websocket, no public URL). Outbound via **Web API** `chat.postMessage` with a bot token (`xoxb-…`).
|
||||
- **Inbound shape:** event envelope with `team_id`, `event.{type,channel,user,text,ts,thread_ts}`; events include `message`, `app_mention`.
|
||||
- **Credentials:** per-workspace **OAuth install** producing a bot token scoped by `team_id`. This is a *multi-tenant* model unlike Telegram/Discord's single-bot model.
|
||||
- **Threading:** messages keyed by `(channel, ts)`; `thread_ts` groups replies.
|
||||
- **Server model:** webhook (public URL + signature verification) or Socket Mode websocket.
|
||||
|
||||
### 2.5 Cross-cutting pattern
|
||||
|
||||
Every one of the four reduces to the same three primitives, with different transport details:
|
||||
|
||||
1. **An inbound event stream** (webhook or long-lived socket) that yields "someone said something in an external thread."
|
||||
2. **An outbound send** addressed by a per-service *destination id* (chat_id / phone / channel_id / channel+ts).
|
||||
3. **A credential/identity scope** (bot token, phone-number id, bot token, workspace token) that a user authorizes once and which the server holds.
|
||||
|
||||
The variance is in **transport lifecycle** (poll vs webhook vs persistent socket), **tenancy** (single bot vs per-workspace OAuth), and **product constraints** (WhatsApp's 24h window and templates, Discord's privileged intent). None of these are agent-model questions; they are a new *transport/connection layer*.
|
||||
|
||||
---
|
||||
|
||||
## 3. The gap between the two
|
||||
|
||||
Mapping external concepts onto Hub concepts exposes the missing pieces:
|
||||
|
||||
| External concept | Hub equivalent today | Gap |
|
||||
|---|---|---|
|
||||
| External user identity (tg id, discord id, slack user, phone) | `Message.sender` = `'user'` or agent id | No external-identity ↔ Hub-user table |
|
||||
| External thread/channel (tg chat, WS number, dc channel, slack channel+ts) | `Chat` / `Thread` | No persistent binding external-destination ↔ Hub thread |
|
||||
| Inbound "says something" event | websocket `send-chat-message` | No non-websocket entry point into the chat/agent pipeline |
|
||||
| Outbound "reply" | websocket fan-out to `user_connections` | No outbound destination; replies only reach live browser sessions |
|
||||
| Credential per service | remotes `(user_id, name)` in-memory | Must be persisted (bots/workspaces outlive sessions) |
|
||||
|
||||
Two of these deserve emphasis because they cut against existing assumptions:
|
||||
|
||||
- **Persistence of connections.** Remotes are explicitly session-scoped and torn down with the serving session. Channels cannot be — a Telegram bot token and the resulting conversations exist whether or not a browser is connected. This means channels must be a *database-backed* resource, not an in-memory one like remotes.
|
||||
- **Bidirectional, unsolicited input.** Every current run originates from an explicit user action (websocket send/prompt). A channel produces runs *on its own*, driven by external events. The run lifecycle (`AgentRun.create` + `received_message`) already supports this, but nothing today *instantiates* it outside a websocket handler.
|
||||
|
||||
---
|
||||
|
||||
## 4. Integration options
|
||||
|
||||
### Option A — Channel as a new App service layer ("channel mixin")
|
||||
|
||||
A `ChannelMixin` (parallel to `ChatMixin`/`AgentMixin`) owns channel connections: connect/disconnect actions, per-service adapters, inbound listeners, and an outbound dispatcher. It translates inbound events into `app.send_chat_message(...)` and subscribes to outgoing chat messages to relay them back out.
|
||||
|
||||
- **Pros:** matches the existing mixin topology; keeps transport concerns out of agents and chats; naturally bidirectional.
|
||||
- **Cons:** the outbound leg has no hook today — the chat path only pushes to websocket `user_connections`. We would need an extension point (e.g. an event emitted on message creation that channels subscribe to), or channels would have to be woven into `notify_message_to_user`. This is the main new plumbing either way.
|
||||
- **Fits:** Telegram, WhatsApp, Slack, Discord all equally; credentials persisted in new tables.
|
||||
|
||||
### Option B — Channel as a specialized agent kind
|
||||
|
||||
Model each external channel as an agent (kind `telegram`, etc.) that is a member of a chat. Inbound events become `received_message`; the agent's `run` sends output back out via the channel API instead of (or in addition to) streaming into the thread.
|
||||
|
||||
- **Pros:** reuses the entire agent lifecycle and membership model; the channel "participates" in chats exactly like other agents; no changes to the chat layer.
|
||||
- **Cons:** conflates *transport* with *behavior* — a channel bot is not an LLM-driven agent, and forcing the loop (`chat_completion_loop`) onto it is a poor fit; `_role_for_message` and the run-message projection assume LLM roles. Also identity: the external sender must map to a Hub user independently of the agent.
|
||||
- **Verdict:** attractive for symmetry, but the agent contract (`prompt`, inference loop, tool resolution) is the wrong abstraction for a raw transport adapter.
|
||||
|
||||
### Option C — Channel as an extended "remote"
|
||||
|
||||
Generalize remotes to include an *inbound* direction: a remote that both exposes functions (outbound tools) and can push external events in.
|
||||
|
||||
- **Pros:** reuses the connection registry, `(user_id, name)` scoping, and `getRemoteTools` expansion.
|
||||
- **Cons:** remotes are deliberately in-memory and session-bound — the exact opposite of what channels need. Extending them would either break their lifecycle guarantees or require persisting them, muddling two different concepts.
|
||||
|
||||
### Option D — Channel as a standalone gateway process
|
||||
|
||||
A separate service that speaks each chat API and talks to Hub through the existing websocket API (acting as a synthetic user session that sends `send-chat-message` and listens for `run-event`/message broadcasts).
|
||||
|
||||
- **Pros:** **zero backend changes**; channels become just another websocket client; deployment-isolated and independently restartable; each service's idiosyncrasies (WhatsApp templates, Discord intents) live outside Hub.
|
||||
- **Cons:** forces an external identity model into the websocket auth (a synthetic user per channel or shared user); cannot easily correlate an incoming external thread to an existing Hub thread without additional storage it would have to own; and it inherits the websocket API's user-scoped shape awkwardly.
|
||||
- **Fits:** good as a *stepping stone* and for prototyping; less clean as the long-term home for identity/binding storage that the rest of Hub should know about.
|
||||
|
||||
---
|
||||
|
||||
## 5. Recommendation
|
||||
|
||||
The cleanest structure is **Option A, with Option D's containment discipline**:
|
||||
|
||||
1. **Add a database-backed channel layer.** New models (sketch): `ChannelConnection` (user, service kind, credentials, display name), `ChannelIdentity` (connection, external sender id → Hub user), and `ChannelBinding` (connection + external destination id ↔ Hub `chat`/`thread`). This is the piece remotes cannot provide because they are in-memory.
|
||||
2. **Define one small adapter interface** — e.g. `start()/stop()` for the inbound listener, `send(destination, content)` for outbound, `normalize(event) → {external_destination, external_sender, content}` — and implement it four times. This absorbs all the per-service variance listed in §2.
|
||||
3. **Inbound** adapters call the existing `ChatMixin`/message-creation path (translating an inbound event into `send_chat_message`), so agents are notified exactly as they are for a websocket message. No agent changes required — `received_message` already turns a thread into a run.
|
||||
4. **Outbound** requires the *one genuinely new seam*: a message-created/settled event that channels can subscribe to, oriented by `ChatBinding`, so agent replies and streamed message updates relay back to the external service. Today only `App.emit_event`/`user_connections` exist for fan-out; channels should hook the same event stream rather than being spliced into `notify_message_to_user`.
|
||||
|
||||
Defer the outbound seam decision only if we choose Option D for a first cut (a channel *is* a websocket client), accepting the identity/binding cost.
|
||||
|
||||
## 6. Open questions to resolve before implementation
|
||||
|
||||
- **Identity:** do external senders become Hub users (1:1), or do external identities hang off a single owning Hub user? The `Message.sender` contract (`'user'` or agent id) does not currently express a *third-party* user.
|
||||
- **Threading semantics:** how do Telegram chats / WhatsApp numbers / Discord channels / Slack `(channel, ts)` map onto Hub `Chat` vs `Thread`? One chat with many threads, or many chats?
|
||||
- **Message kinds:** external messages are plain conversation — confirm they are `kind=message` and that `resource`/`note` (the non-delivered kinds) never leak outward.
|
||||
- **Temporary vs persistent:** must channel-bound chats be forbidden from being temporary, given `tmp.is_tmp_id` branching everywhere?
|
||||
- **Transport lifecycle:** Discord/Socket-Mode need persistent background tasks; Telegram/WhatsApp/Slack-webhook need public endpoints. Where does the server lifecycle own these (a `start_server` hook)?
|
||||
- **WhatsApp constraints:** the 24-hour session window and template requirement are product-level; where are they enforced — adapter, or surfaced to the user as configuration?
|
||||
@@ -1,11 +0,0 @@
|
||||
# Final User Notes
|
||||
|
||||
*This document responds to issues raised by docs/channels/method.md and docs/channels/shape.md, and clarifies some misconceptions and corrected decisions from these documents. Anything said here must be taken as the final requirement.*
|
||||
|
||||
Channels are natively integrated in the hub backend, wwith shared responsibility between app_mixins/channel.py and server (may be one server module for channels or one per channel, depends on individual api complexity). We always use a public endpoint for incoming data (if we ever need a hostname, get it from an env. never imply it but also don't require it).
|
||||
|
||||
Channels are stored as `AgentResource` on the agent that responds to the prompt. The application does an initial query to setup the api endpoint listeners (if necessary, if the host is registered on the chat provider this is not necessary). No data should be persisted beyond essential information required for access.
|
||||
|
||||
The shape report mentions "flattening" the data into the prompt message, which is actually still a map but one level lower. We must flatten the channel input one level further: the entire conversation (the most recent messages up to some byte size limit) goes in a single "conversation" message. Other important channel context (time? idk what's important) goes in a separate user message (first user message for status and info), and the instruction discloses the source of the prompt and what the agent is expected to do with it (responding I guess).
|
||||
|
||||
We will first implement the Telegram channel. We will do the minimum required modifications to get it working, adhering to the design contraints refined in these documents. When done, write 'docs/telegram.md' with brief instructions to setup a telegram channel.
|
||||
@@ -1,173 +0,0 @@
|
||||
# Channel Conversations in the Agent Context: Flatten, Don't Map
|
||||
|
||||
## Purpose & the decision examined
|
||||
|
||||
`docs/channels.md` settled the *transport* question (Option A: a builtin channel layer as a new app mixin, database-backed, with a small per-service adapter interface). It explicitly deferred the question this report addresses: **once an external conversation reaches an agent, what shape does that conversation take?**
|
||||
|
||||
Two candidate shapes:
|
||||
|
||||
- **Map** — bind each external conversation (Telegram chat, WhatsApp number, Discord/Slack channel) to a Hub `Chat`/`Thread`, and drive agents through the existing chat path (`received_message`).
|
||||
- **Flatten** — do *not* create a Hub chat. The channel layer owns the transcript and presents it to the agent as a plain `messages` list plus a framing instruction ("this is a conversation on *Telegram*"), with channel-specific tools (starting with simple outbound text).
|
||||
|
||||
This report argues the codebase is already shaped for **flatten**, identifies exactly which seams it uses, and flags the few that need care. Everything below is grounded in `docs/agent.md`, `docs/api.md`, `backend/src/app.py`, `backend/src/database.py`, `backend/src/agent.py`, `backend/src/agents/expert.py`, `backend/src/app_mixins/agent.py`, `backend/src/app_mixins/chat.py`, and `backend/src/tool.py`.
|
||||
|
||||
---
|
||||
|
||||
## 1. Hub has two agent entry points, and only one is chat-shaped
|
||||
|
||||
`docs/agent.md` names them, and `agent.py` confirms the asymmetry:
|
||||
|
||||
| Entry point | Requires a chat/thread? | Produces |
|
||||
|---|---|---|
|
||||
| `prompt(instruction, input=None, **options)` | **No** | async generator of raw `inference.Message` (role + content) |
|
||||
| `received_message(message)` | **Yes** — reads `message.chat_id` / `message.thread_id` | side-effects: streams new messages *into the thread* |
|
||||
|
||||
`Agent.prompt` (agent.py) is the chat-free path: it builds an `AgentRun` with `input={"instruction": instruction, "input": input, **kwargs}` and yields every message the run emits. It never touches a chat. This is the entry point **flatten** should use.
|
||||
|
||||
`Agent.received_message` (agent.py) is the chat-coupled path. It:
|
||||
- loads `chat.get_thread(message.thread_id)` and its page of messages,
|
||||
- derives LLM roles from senders via `_role_for_message` (sender `== self.id` → `assistant`, `'user'` → `user`, `tmp-*` → `assistant`, else `user`),
|
||||
- collects `remote` resources from the thread,
|
||||
- and, crucially, **writes back**: each `inference` event becomes a new message via `app.send_chat_message(...)` in the same thread, emitting a `chat-message` run event.
|
||||
|
||||
That write-back is the heart of "map". If we flatten, we never call it, and none of its chatter — `Chat`, `Thread`, `Message` rows, membership, `tmp.is_tmp_id` branching — is required.
|
||||
|
||||
**The first report said inbound should translate to `send_chat_message` and let `received_message` do the rest. That is precisely the mapped-chat choice, and flattening reverses it:** inbound goes to a run created via `prompt`, and outbound is a tool call, not a message insert.
|
||||
|
||||
---
|
||||
|
||||
## 2. The Expert agent's `run` is already chat-agnostic
|
||||
|
||||
`ExpertAgent.run` (expert.py) consumes only the run *input*, never a chat:
|
||||
|
||||
```python
|
||||
tools = run.input.get("tools", [])
|
||||
if run.input.get('tool_policy') != 'caller':
|
||||
tools += self.get_tools(run)
|
||||
|
||||
messages = inference.build_messages(
|
||||
system_prompt=self.config.get("system_prompt", ""),
|
||||
instruction=run.input.get("instruction"),
|
||||
input=run.input.get("input"),
|
||||
messages=run.input.get("messages"),
|
||||
)
|
||||
await chat_completion_loop(self.app, run, messages,
|
||||
model_id=self.config.get("model"), tools=tools)
|
||||
```
|
||||
|
||||
Three consequences for a flattened channel conversation:
|
||||
|
||||
1. **Conversation history is just `run.input["messages"]`.** A channel layer passes the reconstructed transcript here. No chat object is involved.
|
||||
2. **Framing is `instruction` (and optionally `input`).** "This is a conversation on Telegram" belongs in `instruction` (per-run), leaving the agent's `system_prompt` general — or the channel binds a channel-configured agent whose `system_prompt` already speaks for that service. Both are available; `instruction` is the lighter, per-conversation lever.
|
||||
3. **`chat_completion_loop` is reusable verbatim.** It executes tools when `tool_policy == 'auto'` (the default), feeds results back, and closes the run. A flattened channel run rides this loop unchanged.
|
||||
|
||||
*Note a config precedence quirk:* `ExpertAgent.run` first overwrites `temperature` / `repetition_penalty` / `reasoning_effort` in `run.input` from the agent config whenever those are set. So **agent config beats per-run input** for those three; per-run framing should stay in `instruction`/`input`/`messages`/`tools`.
|
||||
|
||||
---
|
||||
|
||||
## 3. Outbound becomes a channel tool (this is the whole outbound story)
|
||||
|
||||
The first report called the outbound leg "the one genuinely new seam" because the chat path only fans out to websocket `user_connections`. **Flattening dissolves that seam**: outbound is a `ToolSpec` whose `func` calls the channel adapter's `send(destination, content)`.
|
||||
|
||||
`tool.py` supports exactly this. A `ToolSpec` carries an optional `func`, and `chat_completion_loop` executes it in-loop when policy is `auto`:
|
||||
|
||||
```python
|
||||
tc.result = await tool.execute_async(**tc.arguments)
|
||||
```
|
||||
|
||||
So a Telegram/Slack/… adapter constructs, per run, something like `ToolSpec(name="send_message", func=partial(adapter.send, destination))`, and the loop calls it when the model emits a tool call. Starting with "simple outbound text" is therefore a one-parameter tool (`text`), with richer payloads (photos, threads, buttons) added later as more tools — matching the report's adapter-interface idea, now expressed as run tools.
|
||||
|
||||
### 3.1 A caveat in `prompt_agent` that decides *how* to pass tools
|
||||
|
||||
`AgentMixin.prompt_agent` normalizes `options['tools']`:
|
||||
|
||||
```python
|
||||
def normalize_tool(t):
|
||||
if isinstance(t, dict):
|
||||
... return ToolSpec(name=..., description=..., parameters=...) # NOTE: no func
|
||||
else:
|
||||
return t
|
||||
```
|
||||
|
||||
A **dict** tool descriptor is rebuilt into a `ToolSpec` **without `func`** — and `chat_completion_loop` treats a non-callable `ToolSpec` as `cancel_tools = True` (it aborts the turn and hands the tool back instead of executing). So:
|
||||
|
||||
- **Pass real `ToolSpec` objects (with `func`) through `options['tools']`**, not dicts, and the loop executes the outbound send automatically (`tool_policy` default `auto`).
|
||||
|
||||
### 3.2 Per-run tool selection is additive, not exclusive
|
||||
|
||||
Contrary to `docs/agent.md`'s "if the run carries tools, they are used as-is", the code *always* adds the agent's own tools unless `tool_policy == 'caller'`:
|
||||
|
||||
```python
|
||||
tools = run.input.get("tools", [])
|
||||
if run.input.get('tool_policy') != 'caller':
|
||||
tools += self.get_tools(run) # resource tools + remote tools
|
||||
```
|
||||
|
||||
`get_tools` merges resource tools and remotes and dedupes by name. So to give a channel agent a **clean, channel-only toolset**, bind it to an agent with no resource tools and no remotes. Using `tool_policy == 'caller'` is *not* a good way to restrict tools — it also stops in-loop execution (the loop stops after emitting tool calls, giving you one-shot, non-iterated tool calls), which defeats the simple outbound-send tool.
|
||||
|
||||
### 3.3 Observability for free
|
||||
|
||||
Every run event is broadcast: `AgentRun.emit_event` calls `self.app.emit_event("run-event", ...)`, and the global bus (`app.on_event`) plus per-run listeners are available. A channel layer can subscribe to the run's `inference` / `tool_call` / `done` events for logging, metrics, or to observe outbound sends without inventing new plumbing.
|
||||
|
||||
---
|
||||
|
||||
## 4. State, identity, and the transcript
|
||||
|
||||
Flattening moves the *source of truth for the conversation* into the channel layer, which is also where persistence must live (the first report's "Persistence of connections" point). Two facts from the code make this clean:
|
||||
|
||||
- **Agents and runs are ephemeral.** `docs/agent.md` and `agent.py`: runs live in `app.tmp.runs`, are always in-memory, and are closed with the run. So the transcript cannot live in a run — it must be rebuilt each inbound event from channel-owned storage.
|
||||
- **`Message.sender` cannot express a third party.** `database.Message.sender` is a plain `CharField`, set to the literal `'user'` or an agent id; `_role_for_message` only knows self/`user`/`tmp-*`. Flattening sidesteps this entirely: external speakers become ordinary LLM roles in the `messages` list (with names embedded in content, or as role labels), and **no third-party Hub `User` is needed in the chat model**. Identity is a channel-layer concern (which external sender is talking, and whether they are authorized to trigger the agent), not a Hub-chat concern.
|
||||
|
||||
So the channel layer owns, at minimum: the connection/credential, the external destination ↔ Hub-agent binding, and the transcript used to build `run.input["messages"]`. This is a smaller footprint than the mapped-chat option, which additionally needs external-destination ↔ `Chat`/`Thread` bindings and would have to answer "one chat with many threads, or many chats?".
|
||||
|
||||
---
|
||||
|
||||
## 5. Flatten vs map — weighted comparison
|
||||
|
||||
| Dimension | **Flatten (recommended)** | Map to Hub Chat/Thread |
|
||||
|---|---|---|
|
||||
| Entry point | `prompt` (chat-free) | `received_message` (chat-coupled) |
|
||||
| Agent sees | `run.input["messages"]` + `instruction` framing | thread history projected via `_role_for_message` |
|
||||
| Outbound | channel tool (`ToolSpec.func`) executed by the loop | `send_chat_message` + websocket/relay fan-out |
|
||||
| New backend plumbing | channel-owned storage + adapters (needed either way) | *plus* chat/thread bindings and an outbound relay seam |
|
||||
| Identity | external senders are roles; no third-party `User` | needs an external-identity ↔ `User` model |
|
||||
| Threading semantics | channel layer chooses its own conversation model | forced to answer Chat-vs-Thread mapping per service |
|
||||
| `tmp.is_tmp_id` branching | avoided (no Chat) | must handle/handle-forbid temporary chats |
|
||||
| Reuse | `AgentRun`, Expert loop, tool execution, run events | same, *plus* the whole chat write-back path |
|
||||
| Cost of service differences | absorbed by per-channel tools + adapters | leaks service quirks into the chat model |
|
||||
|
||||
The mapped option's only structural advantage — "the channel is just another chat participant" — is largely cosmetic here, because the participant would be a transport adapter, not an LLM agent, and it drags the identity/binding/threading questions into Hub's durable schema.
|
||||
|
||||
---
|
||||
|
||||
## 6. What flattening reuses vs. what must be built
|
||||
|
||||
**Reused unchanged**
|
||||
- `Agent.prompt` + `AgentRun` (events, streams, cancellation).
|
||||
- `ExpertAgent.run` / `chat_completion_loop` (inference, tool execution, multi-turn tool loop).
|
||||
- `AgentMixin.prompt_agent` as the mixin-level entry (it forwards `instruction`, `input`, `options` → run input).
|
||||
- Global event bus for run observability.
|
||||
|
||||
**Must be built (channel layer)**
|
||||
- Adapters with `start()/stop()`, `send(destination, content)`, `normalize(event)` (per `docs/channels.md` §5).
|
||||
- Durable storage: connection/credentials, external destination ↔ Hub-agent binding, transcript.
|
||||
- Per-run tool construction: `ToolSpec` objects (with `func` closures over the adapter + destination), passed via `options['tools']`.
|
||||
- An inbound handler that, per external event, rebuilds `run.input["messages"]`, sets `instruction` framing, and fires `prompt_agent` (ideally as a tracked task, for cancellation/reaping).
|
||||
- Lifecycle hooks for persistent listeners (Discord gateway, Slack Socket Mode) and public endpoints (WhatsApp/Slack webhooks) — a server-start seam, per the first report.
|
||||
|
||||
---
|
||||
|
||||
## 7. Open questions to resolve before implementation
|
||||
|
||||
1. **History window.** How much transcript is rebuilt per inbound event, and does the channel layer or the agent own truncation/summarization? Runs are in-memory and closed, so any cross-event memory must be persisted by the channel layer.
|
||||
2. **Framing channel.** Is "this is a conversation on *Telegram*" carried purely in `instruction`, or does each channel bind a dedicated agent whose `system_prompt` already speaks for it? (Recommend a dedicated, resource-free agent per channel binding, so §3.2 gives a clean toolset.)
|
||||
3. **Tool granularity.** Start with one `send_message(text)` per channel; define how richer features (reply-to, threads, buttons, WhatsApp templates/24h window) surface — as more tools, or as tool parameters.
|
||||
4. **Sender representation.** Embed names in message content (e.g. `Alice: …`) vs. synthetic roles — and where authorization (is this external sender allowed to drive the agent?) is enforced.
|
||||
5. **Run lifetime & cancellation.** Are inbound runs awaited inline or spawned as tasks, and how are they tracked/cancelled (the run's `cancel()` and `app.tmp.runs` are the available hooks)?
|
||||
6. **The `normalize_tool`/`tool_policy` traps.** Confirm the contract that channel tools are always passed as executable `ToolSpec` objects; consider tightening `prompt_agent.normalize_tool` so a dict descriptor can carry a callable, or documenting the object-only rule.
|
||||
|
||||
---
|
||||
|
||||
### One-line summary
|
||||
|
||||
Flatten is the lower-friction shape: it uses `prompt` (not `received_message`), presents the external conversation as `run.input["messages"]` with a per-channel `instruction`, and makes outbound a channel-specific executable `ToolSpec` — reusing the entire run/loop/tool/event machinery while keeping Hub's chat, identity, and threading models out of the channel path.
|
||||
@@ -1,79 +0,0 @@
|
||||
import fs from 'node:fs';
|
||||
import path from 'node:path';
|
||||
|
||||
const READ_TOOLS = ['list-dir', 'read-file', 'find-files', 'search-files'];
|
||||
const WRITE_TOOLS = ['write-file', 'update-file', 'multi-edit'];
|
||||
|
||||
const parentSystem = `You must execute the given task to completion.
|
||||
You don't revise, you don't ideate. You execute and build. All the information has already been gathered and evaluted before presenting it to you, you go straight to the code and update exactly as instructed.
|
||||
If the goal involves a complex plan, split it into independent parts and use the \`delegate\` tool to give each part to a fresh subagent, saving your own context.
|
||||
Do not delegate trivial single-step work; do it yourself. Work iteratively until the task is fully implemented.
|
||||
CONSTRAINTS:
|
||||
- Prefer reading only files present in the prompt.
|
||||
- Do not test anything beyond syntax nor run any command.
|
||||
- Extrapolating requirements or inferring updates is FORBIDDEN. If insufficient context, terminate.
|
||||
`;
|
||||
|
||||
const childSystem = `Your goal is to execute the specific instruction you were given.
|
||||
Your task was asigned autonomously by a coordinator. Focus only on your assigned part.
|
||||
Note: task is terminated after 20 turns, reasoning traces are not persisted after agent turns.
|
||||
CONSTRAINTS:
|
||||
- You must ONLY read files mentioned in your instruction, and only explore to find them.
|
||||
`;
|
||||
|
||||
function resolveInstruction (input) {
|
||||
if (input.prompt) return input.prompt;
|
||||
|
||||
const candidates = ['prompt.md', 'prompt.txt', 'prompt', 'plan.md', 'report.md'];
|
||||
for (const name of candidates) {
|
||||
const file = path.resolve(Gral.path, '.gral', name);
|
||||
if (fs.existsSync(file)) {
|
||||
console.log("Prompt read from: " + name);
|
||||
return fs.readFileSync(file, 'utf8');
|
||||
}
|
||||
}
|
||||
|
||||
throw new Error("Instruction required.");
|
||||
}
|
||||
|
||||
// Delegate tool: spawn a subagent with the same permissions, but no delegate tool.
|
||||
const delegate = {
|
||||
name: 'delegate',
|
||||
description: 'Spawn a subagent to execute a specific part of the plan. Use it to split complex plans into parts and save context. The subagent has the same file permissions but cannot delegate further.',
|
||||
parameters: {
|
||||
type: 'object',
|
||||
properties: {
|
||||
prompt: { type: 'string', description: 'The specific instruction the subagent must execute.' },
|
||||
},
|
||||
required: ['prompt'],
|
||||
},
|
||||
handler: async (args) => {
|
||||
const res = await Gral.run({
|
||||
instruction: childSystem,
|
||||
messages: [{ role: 'user', content: args.prompt }],
|
||||
agentId: 7,
|
||||
read: true,
|
||||
write: true,
|
||||
defaultTools: [...READ_TOOLS, ...WRITE_TOOLS],
|
||||
// no `delegate` tool: the subagent cannot delegate further
|
||||
});
|
||||
return res;
|
||||
},
|
||||
};
|
||||
|
||||
export default async function build (input) {
|
||||
const instruction = resolveInstruction(input);
|
||||
|
||||
let res = await Gral.run({
|
||||
instruction: parentSystem,
|
||||
messages: [{ role: 'user', content: instruction }],
|
||||
agentId: 7,
|
||||
read: true,
|
||||
write: true,
|
||||
defaultTools: [...READ_TOOLS, ...WRITE_TOOLS],
|
||||
tools: [delegate],
|
||||
maxIter: 64,
|
||||
});
|
||||
|
||||
console.log(res);
|
||||
};
|
||||
@@ -1,38 +0,0 @@
|
||||
import fs from 'node:fs';
|
||||
import path from 'node:path';
|
||||
|
||||
const system = `The user is probably asking for either information, a review, or general suggestions.
|
||||
You can only read files, you're forbidden from writing any file, only respond to the user after light research.
|
||||
Only read up to 10 files. If more context is needed, submit your response regardless with all the information available. Your maximum iteration count is 20, you should NEVER reach it, ideally stay under 5 iterations.
|
||||
`;
|
||||
|
||||
function resolvePrompt (input) {
|
||||
if (input.prompt) return input.prompt;
|
||||
|
||||
const candidates = ['prompt.md', 'prompt.txt', 'prompt'];
|
||||
for (const name of candidates) {
|
||||
const file = path.resolve(Gral.path, '.gral', name);
|
||||
if (fs.existsSync(file)) {
|
||||
return fs.readFileSync(file, 'utf8');
|
||||
}
|
||||
}
|
||||
|
||||
throw new Error("Prompt required.");
|
||||
}
|
||||
|
||||
export default async function research (input) {
|
||||
const prompt = resolvePrompt(input);
|
||||
|
||||
let res = await Gral.run({
|
||||
instruction: system,
|
||||
// input gets bumped to the front of the convo every turn
|
||||
// pass as message instead
|
||||
messages: [{role: "user", content: prompt}],
|
||||
agentId: 7,
|
||||
read: true,
|
||||
write: [],
|
||||
defaultTools: ['list-dir', 'read-file', 'find-files', 'search-files'],
|
||||
});
|
||||
|
||||
console.log(res);
|
||||
};
|
||||
@@ -1,62 +0,0 @@
|
||||
import fs from 'node:fs';
|
||||
import path from 'node:path';
|
||||
|
||||
const system = `Your goal is to write a research report based on the given prompt.
|
||||
Use the \`submit-report\` function when you have gathered all information.
|
||||
You can only read from \`docs\` and from the source code directories \`backend/src\` and \`frontend/src\`. Exploration in any other location is forbidden.
|
||||
Only read up to 10 files. If more context is needed, submit the report regardless with all the information available. This is the breakpoint to work iteratively wit hthe user
|
||||
Note: task is terminated after 20 turns, reasoning traces are not persisted after agent turns.
|
||||
`;
|
||||
|
||||
function resolvePrompt (input) {
|
||||
if (input.prompt) return input.prompt;
|
||||
|
||||
const candidates = ['prompt.md', 'prompt.txt', 'prompt'];
|
||||
for (const name of candidates) {
|
||||
const file = path.resolve(Gral.path, '.gral', name);
|
||||
if (fs.existsSync(file)) {
|
||||
return fs.readFileSync(file, 'utf8');
|
||||
}
|
||||
}
|
||||
|
||||
throw new Error("Prompt required.");
|
||||
}
|
||||
|
||||
function writeReport(content) {
|
||||
fs.writeFileSync(path.resolve(Gral.path, '.gral/report.md'), content);
|
||||
}
|
||||
|
||||
export default async function research (input) {
|
||||
const prompt = resolvePrompt(input);
|
||||
|
||||
let done = false;
|
||||
|
||||
let res = await Gral.run({
|
||||
instruction: system,
|
||||
input: prompt,
|
||||
agentId: 7,
|
||||
tools: [{
|
||||
name: 'submit-report',
|
||||
description: 'Submit a markdown report as your task deliverable.',
|
||||
parameters: {
|
||||
type: 'object',
|
||||
properties: {
|
||||
content: { type: 'string' },
|
||||
},
|
||||
required: ['content'],
|
||||
},
|
||||
handler: (args) => {
|
||||
done = true;
|
||||
writeReport(args.content);
|
||||
}
|
||||
}],
|
||||
read: ['docs', 'frontend/src', 'backend/src'],
|
||||
write: [],
|
||||
defaultTools: ['list-dir', 'read-file', 'find-files', 'search-files'],
|
||||
|
||||
maxIter: 99,
|
||||
verify: () => done,
|
||||
});
|
||||
|
||||
console.log(res);
|
||||
};
|
||||
@@ -1,41 +0,0 @@
|
||||
## The core break
|
||||
|
||||
`Drives.vue` builds the right target path and passes it to the upload helper as a third argument:
|
||||
|
||||
```js
|
||||
const path = dirPath ? `${dirPath}/${file.name}` : file.name;
|
||||
await backend.uploadDriveFile(drive.name, file, path);
|
||||
```
|
||||
|
||||
The backend only accepts uploads at `POST /files/<drive>/<path:filename>` (`upload_file(user_id, drive_name, filename, stream, mime_type)`), i.e. the composite name *must* be in the URL. Docs describe the helper as 2-arg (`uploadDriveFile(driveName, file)`) POSTing to `/files/<drive>`, which has no route. So the documented contract and the view disagree, and every upload (root or folder) fails. The view is right; the docs/wrapper are wrong. This is the same failure the doc already lists under "Known issues," but it's written as a bug report rather than as the correct interface.
|
||||
|
||||
Everything else in `Drives.vue` matches the documented REST API and shapes:
|
||||
- `getDrives()` → `{id, name}` ✓
|
||||
- `getDriveFiles(name)` → `{id, name, size, mime, created}`, consumed by `buildTree` ✓
|
||||
- `deleteDriveFile(name, path)` ✓
|
||||
- `driveFileUrl(name, path)` for both the `<a download>` and `window.open` ✓
|
||||
|
||||
So the *view* speaks the right API. The fix belongs in `backend.js` (accept the path, multipart-POST to `/files/<drive>/<path>`), and the "Frontend interface" section should document the 3-arg signature.
|
||||
|
||||
## Docs vs. backend (`drives.py`)
|
||||
|
||||
1. **`accessed_at` claim is false.** Docs: "Every resolve (download, metadata, delete) stamps `accessed_at`." Only `resolve_file` stamps it. `get_metadata` and `delete_file` never touch it.
|
||||
2. **DELETE return value is wrong.** Docs: returns `{fileId}`. `delete_file` returns the removed absolute disk path (`abs_path`).
|
||||
3. **`resolve_file` returns the wrong second value.** Docs say it serves bytes "with their stored mime type" and the docstring says it returns `(path, mime)`. It actually returns `(abs_path, file.id)` — the file id, not the mime. So mime cannot come from `resolve_file`; the route must look it up. Fix the docstring too.
|
||||
4. **Id- vs name-addressing is undocumented.** The route table presents download/delete/metadata as drive+filename-addressed, but the functions are id-addressed: `get_metadata(file_id)` and `delete_file(file_id)`. Only `upload_file` takes drive+filename. The docs should note the extra filename→id resolution step.
|
||||
5. **Ownership checks don't match the claim.** Docs: "Ownership ... is checked before any disk access." `resolve_file` checks it (badly — already noted for the `None` deref). But `delete_file` and `get_metadata` perform **no** ownership check at all; both work off a bare `file_id`. If the route doesn't guard them, that's a real authorization gap, and the doc overstates the protection.
|
||||
6. **Listing routes aren't in `drives.py`.** The table lists `GET /files` and `GET /files/<drive>` as backend routes, but no `list_drives`/`list_files` exists here — the doc says "logic in `backend/src/drives.py`," which is therefore inaccurate for those two.
|
||||
7. **Minor identity mismatch in the doc's own text.** It describes `DriveFile(id char(64) pk ...)` while `_generate_file_id()` is `secrets.token_hex(8)` → 16 hex chars. Harmless but the doc contradicts itself.
|
||||
|
||||
The already-listed "Known issues" (collision check querying `Drive` instead of `DriveFile`; `resolve_file` dereferencing a `None` drive) are accurate and match the code.
|
||||
|
||||
## Suggested ordering
|
||||
- Fix `backend.js` upload (path arg + `/files/<drive>/<path>` POST); update the doc's frontend section to the 3-arg signature.
|
||||
- Correct the doc's `accessed_at`, DELETE-return, and `resolve_file`-return claims.
|
||||
- Decide whether `delete_file`/`get_metadata` need ownership checks (they currently have none) and document id-vs-name addressing accordingly.
|
||||
|
||||
---
|
||||
|
||||
# Task
|
||||
|
||||
Apply the fixes documented in the prompt above. This task is bounded on the files: docs/drives.md, frontend/src/views/Drives.vue, backend/src/drives.py. You can only access those files, any read or exploration in another location is forbidden.
|
||||
@@ -1,5 +0,0 @@
|
||||
In the recently installed telegram integration across backend/src/channel.py and backend/src/server/telegram.py, documented in docs/telegram.md.
|
||||
It collapses all known channel messages into a single inference message with the entire conversation.
|
||||
Instead, we will link a telegram chat to an actual chat.
|
||||
|
||||
This prompt is a work in progress.
|
||||
@@ -1,6 +1,7 @@
|
||||
Quart==0.20.0
|
||||
quart-auth
|
||||
quart-cors
|
||||
httpx
|
||||
sqlalchemy
|
||||
nest_asyncio
|
||||
replicate
|
||||
|
||||
+11
-2
@@ -22,6 +22,7 @@ from drives import init_drives
|
||||
from remote import RemoteConnection, RemoteToolSpec, parse_function_specs
|
||||
|
||||
from app_mixins import ChatMixin, AgentMixin
|
||||
from channels import Channels
|
||||
|
||||
class App(ChatMixin, AgentMixin):
|
||||
def __init__(self, data_directory, replicate_token=None, openrouter_token=None):
|
||||
@@ -34,6 +35,9 @@ class App(ChatMixin, AgentMixin):
|
||||
# active websocket tasks, for manual cancelling upon server shutdown
|
||||
self.ws_tasks = []
|
||||
|
||||
# Channel bindings, in-memory transcripts, pollers and HTTP.
|
||||
self.channels = Channels(self)
|
||||
|
||||
self.tmp = tmp.State()
|
||||
|
||||
# Clients
|
||||
@@ -236,13 +240,16 @@ async def main():
|
||||
openrouter_token=os.getenv("OPENROUTER_API_KEY")
|
||||
)
|
||||
|
||||
# 3. Start Server
|
||||
# 3. Set up channel listeners (webhook reconciliation or long polling)
|
||||
await horde_app.channels.setup()
|
||||
|
||||
# 4. Start Server
|
||||
# manually trigger server shutdown, avoids hypercorn installing signal handlers
|
||||
shutdown_event = asyncio.Event()
|
||||
# TODO: get port from env or don't pass it (actually server should handle none)
|
||||
server_task = asyncio.create_task(start_server(horde_app, port, shutdown_event))
|
||||
|
||||
# 4. Graceful Termination
|
||||
# 5. Graceful Termination
|
||||
loop = asyncio.get_running_loop()
|
||||
def interrupt():
|
||||
print("Interrupt Signal")
|
||||
@@ -250,6 +257,8 @@ async def main():
|
||||
# Unblock websocket handlers parked in websocket.receive()
|
||||
for task in list(horde_app.ws_tasks):
|
||||
task.cancel()
|
||||
# Stop channel pollers and any in-flight channel runs
|
||||
horde_app.channels.shutdown()
|
||||
|
||||
for sig in (signal.SIGINT, signal.SIGTERM):
|
||||
loop.add_signal_handler(sig, interrupt)
|
||||
|
||||
@@ -1,2 +1,2 @@
|
||||
from app_mixins.chat import ChatMixin
|
||||
from app_mixins.agent import AgentMixin
|
||||
from app_mixins.agent import AgentMixin
|
||||
@@ -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
|
||||
@@ -27,7 +27,7 @@ async def upload_file(user_id, drive_name, filename, stream, mime_type=None):
|
||||
drive = await db.Drive.create(user_id=user_id, name=drive_name)
|
||||
|
||||
# Check collision
|
||||
if await db.DriveFile.get_or_none(drive_id=drive.id, filename=filename):
|
||||
if await db.Drive.get_or_none(drive_id=drive.id, filename=filename):
|
||||
raise Exception("Duplicated file")
|
||||
|
||||
# 2. Prepare disk path
|
||||
@@ -82,7 +82,7 @@ async def delete_file(file_id):
|
||||
return abs_path
|
||||
|
||||
async def resolve_file(user_id, drive_name, filename):
|
||||
"""Resolve file to absolute disk path after permission check. Returns (abs_path, file.id)."""
|
||||
"""Resolve file to absolute disk path after permission check. Returns (path, mime)."""
|
||||
# Eager load drive and verify ownership
|
||||
drive = await db.Drive.get_or_none(user_id=user_id, name=drive_name)
|
||||
if drive.user_id != user_id:
|
||||
|
||||
@@ -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}
|
||||
@@ -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)
|
||||
+12
-14
@@ -1,29 +1,27 @@
|
||||
# Drives
|
||||
|
||||
Per-user file storage: a **drive** is a named container owned by a user, and each **file** is a database record plus a blob on disk. Client-facing routes are addressed by *drive name + filename* — the random file id never appears in URLs — though several `drives.py` helpers below are id-addressed internally. Drives bypass the WebSocket action protocol entirely: all endpoints are plain REST (routes in `backend/src/server/web.py`, logic in `backend/src/drives.py`), so downloads can be plain `<a>` links.
|
||||
Per-user file storage: a **drive** is a named container owned by a user, and each **file** is a database record plus a blob on disk. Everything is addressed by *drive name + filename* — the random file id never appears in URLs. Drives bypass the WebSocket action protocol entirely: all endpoints are plain REST (routes in `backend/src/server/web.py`, logic in `backend/src/drives.py`), so downloads can be plain `<a>` links.
|
||||
|
||||
## Logical model
|
||||
|
||||
- **Drive** — user-scoped, addressed by name. Created implicitly on first upload to a new name; there is no create/rename/delete-drive endpoint.
|
||||
- **File** — readable `filename`, mime type, size, `created`/`accessed` timestamps.
|
||||
- **Folders are virtual** — a filename may contain slashes (`docs/api/readme.md`); the Drives view splits names to render a tree, but storage stays flat.
|
||||
- **Identity** — `DriveFile.id` is a random hex string (`secrets.token_hex(8)`, 16 hex chars) that doubles as the on-disk filename, so user-facing names never touch the filesystem.
|
||||
- **Identity** — `DriveFile.id` is a random hex string (`secrets.token_hex(8)`) that doubles as the on-disk filename, so user-facing names never touch the filesystem.
|
||||
|
||||
## Backend interface
|
||||
|
||||
Download links can't set headers, so the GET/DELETE file routes additionally accept the token as a `?token=` query parameter.
|
||||
|
||||
Client-facing file routes are drive+filename-addressed, but the `drives.py` helpers behind them are mixed: `upload_file` accepts drive+filename directly, while `get_metadata(file_id)` and `delete_file(file_id)` are id-addressed, so their routes must first resolve filename → id (via the drive's file records) before calling them. The two listing routes are not implemented in `drives.py` at all — their logic lives elsewhere.
|
||||
|
||||
| Route | Behavior |
|
||||
| --- | --- |
|
||||
| `GET /files` | List the user's drives: `[{id, name}]`. Not implemented in `drives.py`. |
|
||||
| `GET /files/<drive>` | List files: `[{id, name, size, mime, created}]`. Not implemented in `drives.py`. |
|
||||
| `GET /files` | List the user's drives: `[{id, name}]` |
|
||||
| `GET /files/<drive>` | List files: `[{id, name, size, mime, created}]` |
|
||||
| `POST /files/<drive>/<filename>` | Multipart upload (field `file`) in a single request. The URL tail — `<path:filename>`, slashes allowed — becomes the stored filename; the part's own name is the fallback. Returns `{fileId}`. |
|
||||
| `GET /files/<drive>/<filename>` | Serve the bytes, with the mime type looked up from the file record (not from `resolve_file`), or return `{id, name, size, mime, created, accessed}` when `?metadata` is present. |
|
||||
| `DELETE /files/<drive>/<filename>` | Delete the record, then the blob. Returns the removed absolute disk path (`abs_path`). |
|
||||
| `GET /files/<drive>/<filename>` | Serve the bytes with their stored mime type, or return `{id, name, size, mime, created, accessed}` when `?metadata` is present. |
|
||||
| `DELETE /files/<drive>/<filename>` | Delete the record, then the blob. Returns `{fileId}`. |
|
||||
|
||||
Only `resolve_file` stamps `accessed_at`; `get_metadata` and `delete_file` never touch it.
|
||||
Every resolve (download, metadata, delete) stamps `accessed_at`.
|
||||
|
||||
## Frontend interface
|
||||
|
||||
@@ -31,20 +29,20 @@ In `frontend/src/backend.js`:
|
||||
|
||||
- `webFetch(path, options)` — generic REST helper: sets `Authorization: Bearer` (waiting for login if needed), parses JSON, throws on non-2xx. Downloads bypass it.
|
||||
- `getDrives()`, `getDriveFiles(driveName)`, `deleteDriveFile(driveName, filename)`
|
||||
- `uploadDriveFile(driveName, file, path)` — multipart-POSTs the whole file to `/files/<driveName>/<path>`, where `path` is the destination filename including any virtual folder prefix (`docs/api/readme.md`). The composite path belongs in the URL, since the upload route only exists at `/files/<drive>/<path:filename>`.
|
||||
- `uploadDriveFile(driveName, file)` — multipart POST of the whole file.
|
||||
- `driveFileUrl(driveName, filename)` — GET URL carrying `?token=`, used for download anchors and for opening a file in a tab.
|
||||
|
||||
`frontend/src/views/Drives.vue` lazy-loads a drive's files on first expand, rebuilds the folder tree from the flat file list on every load (open folders live on the drive object, so they survive reloads), uploads through a file picker (building the target path from the picked folder and passing it as `uploadDriveFile`'s third argument), and deletes with confirmation.
|
||||
`frontend/src/views/Drives.vue` lazy-loads a drive's files on first expand, rebuilds the folder tree from the flat file list on every load (open folders live on the drive object, so they survive reloads), uploads through a file picker, and deletes with confirmation.
|
||||
|
||||
## Implementation notes
|
||||
|
||||
- Models in `backend/src/database.py` (migration `0002_auto_20260921_1840`): `Drive(id, user_id, name, created_at)` and `DriveFile(id, drive_id, filename, mime_type, size_bytes, accessed_at, created_at)`, table `drive_files`. The `id` is a 16-char hex string, not `char(64)`. Ownership rides the `User → Drive → File` chain; it is checked in `resolve_file` before disk access, but `delete_file` and `get_metadata` take a bare `file_id` and perform no ownership check of their own — their routes must guard them.
|
||||
- Models in `backend/src/database.py` (migration `0002_auto_20260921_1840`): `Drive(id, user_id, name, created_at)` and `DriveFile(id char(64) pk, drive_id, filename, mime_type, size_bytes, accessed_at, created_at)`, table `drive_files`. Ownership rides the `User → Drive → File` chain and is checked before any disk access.
|
||||
- Disk layout: `{data_dir}/drives/{drive_id}/{file_id}` — keyed by drive id, not by user or name. `init_drives(data_directory)`, called from `main()`, creates the root.
|
||||
- Upload streams to disk in 8 KiB chunks and only then writes the DB record; delete removes the record first, then the blob.
|
||||
|
||||
## Known issues
|
||||
|
||||
- `uploadDriveFile` POSTs to `/files/<drive>`, where no POST route exists, and drops the folder-path argument `Drives.vue` passes it — UI uploads fail, and folder-targeted upload is not wired up.
|
||||
- Drives are only created implicitly, and the view only uploads into existing drives, so the first drive cannot be created from the frontend.
|
||||
- Uploads are meant to refuse a filename already used in the drive, but the check queries the `Drive` model with `DriveFile`'s fields, so it errors instead of detecting duplicates.
|
||||
- `resolve_file` dereferences the drive without a None check, so a nonexistent drive yields a 500 instead of a 404.
|
||||
- `delete_file` and `get_metadata` take a bare `file_id` with no ownership check; if their routes don't resolve and authorize the file against the owning user, any authenticated user could delete or inspect another user's file by id.
|
||||
- `resolve_file` dereferences the drive without a None check, so a nonexistent drive yields a 500 instead of a 404.
|
||||
@@ -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. |
|
||||
@@ -379,8 +379,6 @@ export function invalidateCache({ chats=false, agents=false } = {}) {
|
||||
// Drive endpoints are plain HTTP routes instead of WebSocket actions,
|
||||
// so they authenticate with the bearer token in the Authorization header.
|
||||
|
||||
// TODO: Next time these functions need updating, move them directly to views/Drives.vue
|
||||
|
||||
export function getDrives () {
|
||||
return webFetch('/files');
|
||||
}
|
||||
@@ -389,10 +387,10 @@ export function getDriveFiles (driveName) {
|
||||
return webFetch(`/files/${driveName}`);
|
||||
}
|
||||
|
||||
export function uploadDriveFile (driveName, file, path) {
|
||||
export function uploadDriveFile (driveName, file) {
|
||||
const body = new FormData();
|
||||
body.append('file', file, file.name);
|
||||
return webFetch(`/files/${driveName}/${path}`, {
|
||||
return webFetch(`/files/${driveName}`, {
|
||||
method: 'POST',
|
||||
body,
|
||||
});
|
||||
|
||||
@@ -2,7 +2,6 @@
|
||||
import { ref, computed, watch } from 'vue';
|
||||
import { marked } from "marked";
|
||||
import * as backend from '../backend';
|
||||
import * as iconoir from '@iconoir/vue';
|
||||
marked.use({
|
||||
renderer: {
|
||||
code ({lang, text}) {
|
||||
@@ -34,12 +33,6 @@ const props = defineProps({
|
||||
logExpanded: {
|
||||
type: Boolean,
|
||||
required: false,
|
||||
},
|
||||
// Toggle set of expanded message ids, shared with the parent view.
|
||||
// Used by log-groups to expand/collapse the previewed log.
|
||||
expandedLogs: {
|
||||
type: Object,
|
||||
required: false,
|
||||
}
|
||||
});
|
||||
|
||||
@@ -52,8 +45,6 @@ function chatEntryClass(msg) {
|
||||
return ['resource', 'resource-' + (msg.subkind || 'generic')];
|
||||
} else if (msg.type == 'log') {
|
||||
return ['log'];
|
||||
} else if (msg.type == 'log-group') {
|
||||
return ['log-group'];
|
||||
} else if (msg.type == 'thread-header') {
|
||||
let cls = ['thread'];
|
||||
if (!msg.active) cls.push(msg.isFork ? 'thread-fork' : 'thread-closed')
|
||||
@@ -117,106 +108,10 @@ watch(() => [props.message.format, props.message.content], ([newFormat, newConte
|
||||
}
|
||||
}, {immediate: true})
|
||||
|
||||
|
||||
// === Log groups === //
|
||||
// A log-group collects consecutive log messages. Each one is shown as a square
|
||||
// icon (which selects it as the preview); the selected log is previewed below.
|
||||
|
||||
const logs = computed(() =>
|
||||
props.message.type == 'log-group' ? (props.message.logs || []) : []
|
||||
);
|
||||
|
||||
// Icon used for a log's square button, chosen by log subkind
|
||||
function logIcon(subkind) {
|
||||
if (subkind == 'reasoning') return iconoir.Brain;
|
||||
if (subkind == 'tool-call') return iconoir.Wrench;
|
||||
return iconoir.Terminal;
|
||||
}
|
||||
|
||||
// Null means "default to the last log in the group"
|
||||
const selectedLogId = ref(null);
|
||||
|
||||
const selectedLog = computed(() => {
|
||||
const arr = logs.value;
|
||||
if (!arr.length) return null;
|
||||
|
||||
if (selectedLogId.value != null) {
|
||||
const found = arr.find(l => l.id == selectedLogId.value);
|
||||
if (found) return found;
|
||||
}
|
||||
|
||||
return arr[arr.length - 1];
|
||||
});
|
||||
|
||||
// Streamed logs resolve their content through a shared per-stream cache, so
|
||||
// switching the preview between logs still shows up-to-date content.
|
||||
const boundStreams = new Set();
|
||||
const streamContents = ref({});
|
||||
|
||||
function bindLogStreams() {
|
||||
for (const log of logs.value) {
|
||||
if (log.format != 'stream') continue;
|
||||
|
||||
const streamId = parseInt(log.content);
|
||||
if (boundStreams.has(streamId)) continue;
|
||||
boundStreams.add(streamId);
|
||||
|
||||
const stream = backend.readStream(streamId);
|
||||
const update = () => {
|
||||
streamContents.value = { ...streamContents.value, [streamId]: stream.partialContent };
|
||||
};
|
||||
update();
|
||||
stream.on('data', update);
|
||||
}
|
||||
}
|
||||
|
||||
watch(logs, bindLogStreams, { immediate: true });
|
||||
|
||||
const selectedLogText = computed(() => {
|
||||
const log = selectedLog.value;
|
||||
if (!log) return '';
|
||||
if (log.format == 'stream') return streamContents.value[parseInt(log.content)] ?? '';
|
||||
return log.content ?? '';
|
||||
});
|
||||
|
||||
const groupLogLong = computed(() => {
|
||||
if (!selectedLog.value) return false;
|
||||
const lines = String(selectedLogText.value ?? '').split('\n');
|
||||
return lines.length > LOG_MAX_LINES || lines.some(line => line.length > LOG_MAX_LENGTH);
|
||||
});
|
||||
|
||||
const groupExpanded = computed(() =>
|
||||
selectedLog.value != null && !!props.expandedLogs?.has(selectedLog.value.id)
|
||||
);
|
||||
|
||||
// Collapsed preview keeps the *last* N characters (the tail), not the head;
|
||||
// CSS then clips the top so only the bottom lines show.
|
||||
const groupLogContent = computed(() => {
|
||||
const text = String(selectedLogText.value ?? '');
|
||||
if (groupLogLong.value && !groupExpanded.value) {
|
||||
return text.slice(-(LOG_MAX_LENGTH * LOG_MAX_LINES));
|
||||
}
|
||||
return text;
|
||||
});
|
||||
|
||||
const groupLogMeta = computed(() => groupLogLong.value ? {expanded: groupExpanded.value} : null);
|
||||
|
||||
function toggleGroupLog() {
|
||||
if (selectedLog.value) props.expandedLogs?.toggle(selectedLog.value.id);
|
||||
}
|
||||
|
||||
function onContextMenu() {
|
||||
if (props.message.type == 'log-group') {
|
||||
if (selectedLog.value) emit('open-menu', selectedLog.value, groupLogMeta.value);
|
||||
} else {
|
||||
emit('open-menu', props.message, logMeta.value);
|
||||
}
|
||||
}
|
||||
|
||||
</script>
|
||||
|
||||
<template>
|
||||
<div :class="chatEntryClass(message)" @contextmenu.prevent="onContextMenu" @click="handleClick(message)">
|
||||
<div :class="chatEntryClass(message)" @contextmenu.prevent="$emit('open-menu', message, logMeta)" @click="handleClick(message)">
|
||||
<template v-if="message.type == 'message' || message.type == null">
|
||||
<div v-if="message.sender != 'user'" class="sender">{{message.sender}}</div>
|
||||
|
||||
@@ -240,28 +135,6 @@ function onContextMenu() {
|
||||
<a v-if="logLong" class="log-toggle" @click.prevent="$emit('toggle-log')">{{logExpanded ? '(collapse)' : '(expand)'}}</a>
|
||||
</template>
|
||||
|
||||
<template v-if="message.type == 'log-group'">
|
||||
<div class="log-icons">
|
||||
<button
|
||||
v-for="log in logs"
|
||||
:key="log.id"
|
||||
type="button"
|
||||
class="icon"
|
||||
:class="{ highlight: selectedLog && log.id == selectedLog.id }"
|
||||
:title="(log.sender || '') + (log.subkind ? ' [' + log.subkind + ']' : '')"
|
||||
@click="selectedLogId = log.id"
|
||||
>
|
||||
<component :is="logIcon(log.subkind)" />
|
||||
</button>
|
||||
</div>
|
||||
|
||||
<div v-if="selectedLog" class="log-preview">
|
||||
<div class="log-sender">{{selectedLog.sender}}<span v-if="selectedLog.subkind"> [{{selectedLog.subkind}}]</span></div>
|
||||
<div class="content" :class="{ clipped: !groupExpanded }">{{groupLogContent}}</div>
|
||||
<a v-if="groupLogLong" class="log-toggle" @click.prevent="toggleGroupLog()">{{groupExpanded ? '(collapse)' : '(expand)'}}</a>
|
||||
</div>
|
||||
</template>
|
||||
|
||||
<template v-if="message.type == 'thread-header'">
|
||||
<div class="thread-title">
|
||||
{{ message.threadId == null ? 'New Thread' : `Thread ${message.threadId}` }}
|
||||
@@ -346,67 +219,6 @@ function onContextMenu() {
|
||||
user-select: none;
|
||||
}
|
||||
|
||||
/* === Log groups === */
|
||||
|
||||
/* Completely transparent container; stacks icons above the preview and
|
||||
stretches to the full width of the chat */
|
||||
.log-group {
|
||||
align-self: stretch;
|
||||
width: 100%;
|
||||
padding: 0.6em 1.4em;
|
||||
background: transparent;
|
||||
display: flex;
|
||||
flex-direction: column;
|
||||
gap: 0.4em;
|
||||
}
|
||||
|
||||
/* Square buttons (sized by the global `button.icon`) that wrap to the width,
|
||||
each line centered within the container */
|
||||
.log-icons {
|
||||
display: flex;
|
||||
flex-wrap: wrap;
|
||||
justify-content: center;
|
||||
gap: 0.3em;
|
||||
}
|
||||
|
||||
.log-icons .highlight {
|
||||
color: var(--color-highlight);
|
||||
border-color: var(--color-highlight);
|
||||
}
|
||||
|
||||
.log-preview {
|
||||
font-family: monospace;
|
||||
font-size: 80%;
|
||||
max-width: 70em;
|
||||
}
|
||||
|
||||
.log-preview .log-sender {
|
||||
font-weight: bold;
|
||||
}
|
||||
|
||||
.log-preview .content {
|
||||
white-space: pre-wrap;
|
||||
word-wrap: anywhere;
|
||||
width: fit-content;
|
||||
max-width: 70em;
|
||||
line-height: 1.3;
|
||||
}
|
||||
|
||||
/* Clipping shows only the bottom lines: the content is bottom-aligned, so any
|
||||
vertical overflow is clipped off the top. ~6 lines at line-height 1.3. */
|
||||
.log-preview .content.clipped {
|
||||
display: flex;
|
||||
flex-direction: column;
|
||||
justify-content: flex-end;
|
||||
max-height: 7.8em;
|
||||
overflow: hidden;
|
||||
}
|
||||
|
||||
.log-preview .log-toggle {
|
||||
cursor: pointer;
|
||||
user-select: none;
|
||||
}
|
||||
|
||||
.resource {
|
||||
border-radius: var(--border-radius);
|
||||
border: 1px dashed var(--color-border);
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
<script setup>
|
||||
import {
|
||||
ref, computed, watch,
|
||||
ref, watch,
|
||||
onMounted, onUnmounted,
|
||||
} from "vue";
|
||||
|
||||
@@ -30,29 +30,6 @@
|
||||
creatingThread,
|
||||
} = session;
|
||||
|
||||
// Consecutive log messages collapse into a single "log-group" entry, which
|
||||
// Message.vue renders as a container of icons plus a preview.
|
||||
const displayEntries = computed(() => {
|
||||
const entries = chatEntries.value || [];
|
||||
const out = [];
|
||||
let group = null;
|
||||
|
||||
for (const entry of entries) {
|
||||
if (entry.type == 'log') {
|
||||
if (!group) {
|
||||
group = { id: 'log-group-' + entry.id, type: 'log-group', logs: [] };
|
||||
out.push(group);
|
||||
}
|
||||
group.logs.push(entry);
|
||||
} else {
|
||||
group = null;
|
||||
out.push(entry);
|
||||
}
|
||||
}
|
||||
|
||||
return out;
|
||||
});
|
||||
|
||||
// UI state coupled to the document itself
|
||||
|
||||
// Message ids of logs currently expanded
|
||||
@@ -239,7 +216,7 @@
|
||||
<div class="row" style="overflow: auto;">
|
||||
<div class="column">
|
||||
<div class="messages-container" ref="chatContainerElem">
|
||||
<div class="message-container" v-for="msg in displayEntries">
|
||||
<div class="message-container" v-for="msg in chatEntries">
|
||||
<Message
|
||||
:key="msg.id"
|
||||
:message="msg"
|
||||
@@ -248,7 +225,6 @@
|
||||
|
||||
:log-expanded="expandedLogs.has(msg.id)"
|
||||
@toggle-log="expandedLogs.toggle(msg.id)"
|
||||
:expanded-logs="expandedLogs"
|
||||
|
||||
:cancellable="messageRunMap[msg.id]"
|
||||
@cancel="session.cancelMessageRun(msg.id)"
|
||||
@@ -377,4 +353,4 @@
|
||||
height: 4em;
|
||||
cursor: pointer;
|
||||
}
|
||||
</style>
|
||||
</style>
|
||||
Reference in new issue
Block a user