| 1 | // The per-bot chat runtime, extracted from agent.mjs so one sidecar process can |
| 2 | // host several identities (Claude, Eleven, …) — each bot gets its own session, |
| 3 | // WebSocket, thread keys, cursors, and console config, all closed over here. |
| 4 | // Behaviour is unchanged from the single-bot runtime: decrypt incoming messages |
| 5 | // with the thread key, ask the Claude API for a reply, encrypt it, post it back. |
| 6 | // The lchat server only ever sees ciphertext. |
| 7 | import fs from "node:fs"; |
| 8 | import Anthropic from "@anthropic-ai/sdk"; |
| 9 | import WebSocket from "ws"; |
| 10 | import { importKey, encrypt, decrypt, unwrapThreadKey, openSealed, sealForPub } from "../web/static/crypto.js"; |
| 11 | import { BASE, WS_BASE, MODEL, CLASSIFIER_MODEL, APP_NAME, AI_ENABLED } from "./config.mjs"; |
| 12 | import { actionsFromToolUse } from "./actions.mjs"; |
| 13 | import { runCommand } from "./commands.mjs"; |
| 14 | import { trimContext, scrubSelfNotes } from "./context.mjs"; |
| 15 | import { listDomains, exchangeViaBroker } from "./dnsimple.mjs"; |
| 16 | |
| 17 | const log = (m) => { |
| 18 | try { |
| 19 | fs.writeSync(1, `${m}\n`); |
| 20 | } catch {} |
| 21 | }; |
| 22 | |
| 23 | let claude; // one API client shared by all bots in the process |
| 24 | const api = () => (claude ??= new Anthropic()); // reads ANTHROPIC_API_KEY from env |
| 25 | |
| 26 | // A non-overridable safety + app-awareness preamble is always applied, then the |
| 27 | // bot's persona (console config > builtin > env > default), then reference docs. |
| 28 | const SAFETY = (name) => |
| 29 | `You are ${name}, an AI assistant and a member of ${APP_NAME} — a small, private, ` + |
| 30 | `end-to-end encrypted group chat app; the server and its operator can't read messages, ` + |
| 31 | `and that privacy is the whole point of the app. You are one participant among the people here. ` + |
| 32 | `You may be in a one-to-one DM or in a group; in a group you're brought in only when someone addresses you directly (usually an @mention). ` + |
| 33 | `You can only take part in the conversation — no reminders, notifications, files, or web access — ` + |
| 34 | `so never promise to do those things; if asked, say you can't yet. ` + |
| 35 | `This chat may include children: always be kind and age-appropriate, and never say anything scary, adult, or unsafe. ` + |
| 36 | `Each message is prefixed with the sender's name — reply naturally and don't prefix your own name.`; |
| 37 | const DEFAULT_PERSONA = "Keep replies short, warm, and easy to read."; |
| 38 | |
| 39 | // runBot starts one bot identity: { name, session, builtinPersona?, builtinDocs? }. |
| 40 | // builtin* are code-shipped defaults (e.g. Eleven's app guide) that console |
| 41 | // config can still override per instance. `tools` (Claude tool defs) let the |
| 42 | // bot propose one-tap action chips; tool_use blocks in a reply become the |
| 43 | // payload's `actions` (see actions.mjs — the client whitelists again). |
| 44 | export function runBot({ name, session, builtinPersona = "", builtinDocs = "", tools = null, commands = false }) { |
| 45 | let persona = builtinPersona || process.env.LCHAT_SYSTEM || DEFAULT_PERSONA; |
| 46 | let model = MODEL; |
| 47 | let docs = builtinDocs; |
| 48 | // How the bot listens in groups: "mention" (default) replies only to an |
| 49 | // explicit @Name; "name" also considers a bare-name mention (still gated by |
| 50 | // the classifier below). Console config wins; env is the headless fallback. |
| 51 | let listen = process.env.LCHAT_LISTEN || "mention"; |
| 52 | |
| 53 | const tag = (m) => log(`[${name}] ${m}`); |
| 54 | |
| 55 | function systemPrompt() { |
| 56 | let s = `${SAFETY(name)}\n\n${persona}`; |
| 57 | // Bots know who their fellow helpers are — without this, one bot seeing |
| 58 | // another's message treats it as an impostor (Claude once publicly doubted |
| 59 | // Eleven's legitimacy). They still never reply to each other (processMsg). |
| 60 | const others = [...agentNames].filter((n) => n !== name); |
| 61 | if (others.length) |
| 62 | s += |
| 63 | `\n\nOther helper bots are members of this server alongside you: ${others.join(", ")}. ` + |
| 64 | `They are legitimate, separately-run assistants (Eleven guides people around the app; ` + |
| 65 | `Claude is a general companion). You may see their messages ` + |
| 66 | `in shared groups; you never reply to them, and there's no need to question their presence.`; |
| 67 | // Context is rebuilt from message TEXT only, so the model sees its own past |
| 68 | // chip replies as bare "tap to switch" texts and starts imitating them |
| 69 | // without calling the tool — the button silently vanishes. Make the rule |
| 70 | // unconditional: promising a button REQUIRES the tool call, every time. |
| 71 | if (tools) |
| 72 | s += |
| 73 | `\n\nYour buttons only exist when you call the matching tool: every reply that offers ` + |
| 74 | `a one-tap action ("tap to switch", "here you go") MUST include that tool call — text alone ` + |
| 75 | `attaches nothing, even if earlier replies in the conversation look like plain text. ` + |
| 76 | `Never tell someone a button is attached without calling the tool in that same reply. ` + |
| 77 | `"[note to self: …]" lines in the history are added automatically where you called tools ` + |
| 78 | `before — never write them yourself.`; |
| 79 | if (docs) s += `\n\nAlways keep this reference in mind:\n${docs}`; |
| 80 | return s; |
| 81 | } |
| 82 | |
| 83 | // The system prompt as a cache-friendly content block. Eleven ships an |
| 84 | // exhaustive app/architecture reference, so the same large system prompt rides |
| 85 | // EVERY DM and @mention; marking it cacheable lets Anthropic serve it from the |
| 86 | // prompt cache instead of re-reading it each call — a big cost + latency win. |
| 87 | // The whole prompt is stable between calls (the reference and persona don't |
| 88 | // change; only an occasional roster change to the "other helpers" line busts |
| 89 | // it), and `docs` sits last, so one trailing breakpoint caches everything. |
| 90 | // Small prompts (no docs) fall below the cache floor and simply aren't cached — |
| 91 | // harmless, so the marker is unconditional. |
| 92 | function systemBlocks() { |
| 93 | return [{ type: "text", text: systemPrompt(), cache_control: { type: "ephemeral" } }]; |
| 94 | } |
| 95 | |
| 96 | const keys = new Map(); // threadId -> { key: CryptoKey, kind: "dm"|"group"|"self" } |
| 97 | const cursors = new Map(); // threadId -> last message id we've processed |
| 98 | // Threads where a client-run scripted flow has the floor (flow.state events |
| 99 | // — the one-Eleven welcome): the brain stays quiet there until the script |
| 100 | // ends or the entry expires (the fail-safe for a client that dies mid-flow). |
| 101 | const muted = new Map(); // threadId -> expiry epoch ms |
| 102 | let agentNames = new Set(); // roster labels of ALL agents — never reply to one |
| 103 | |
| 104 | const get = (path) => |
| 105 | fetch(`${BASE}${path}${path.includes("?") ? "&" : "?"}token=${encodeURIComponent(session.token)}`).then((r) => |
| 106 | r.json() |
| 107 | ); |
| 108 | |
| 109 | // Bots never talk to bots: a bot's mention trigger matches its bare name, and |
| 110 | // bot-authored text naturally names other bots (Eleven recommends Claude) — |
| 111 | // without this, two bots in one group |
| 112 | // ping-pong forever. Name-based (payloads aren't sender-signed yet), refreshed |
| 113 | // on every (re)connect so newly added bots are picked up within a cycle. |
| 114 | async function loadAgents() { |
| 115 | try { |
| 116 | const { roster = [] } = await get("/api/roster"); |
| 117 | agentNames = new Set(roster.filter((e) => e.is_agent).map((e) => e.label)); |
| 118 | } catch (e) { |
| 119 | tag(`loadAgents: ${e.message}`); |
| 120 | } |
| 121 | } |
| 122 | |
| 123 | const maxId = (messages) => messages.reduce((m, x) => Math.max(m, x.id || 0), 0); |
| 124 | |
| 125 | // Fetch our threads and cache the (unwrapped) key for any we don't hold yet. A |
| 126 | // newly-seen thread's cursor starts at its latest message, so we never reply to |
| 127 | // history — only to messages that arrive after we become aware of it. |
| 128 | async function loadThreads() { |
| 129 | const { threads = [] } = await get("/api/threads"); |
| 130 | for (const t of threads) { |
| 131 | if (!t.wrapped || keys.has(t.id)) continue; |
| 132 | try { |
| 133 | const raw = await unwrapThreadKey(t.wrapped, session.userPriv); |
| 134 | keys.set(t.id, { key: await importKey(raw), kind: t.kind }); |
| 135 | const { messages = [] } = await get(`/api/messages?thread=${encodeURIComponent(t.id)}&limit=1`); |
| 136 | cursors.set(t.id, maxId(messages)); |
| 137 | } catch (e) { |
| 138 | tag(`load thread ${t.id}: ${e.message}`); |
| 139 | } |
| 140 | } |
| 141 | } |
| 142 | |
| 143 | // Pull console-managed config (persona / model / reference docs) for this bot |
| 144 | // and apply it over the builtins/env. Re-fetched on reconnect so admin edits |
| 145 | // take effect. |
| 146 | async function loadConfig() { |
| 147 | try { |
| 148 | const c = await get("/api/bot-config"); |
| 149 | if (c && typeof c === "object") { |
| 150 | persona = c.system_prompt || builtinPersona || process.env.LCHAT_SYSTEM || DEFAULT_PERSONA; |
| 151 | model = c.model || MODEL; |
| 152 | docs = c.docs || builtinDocs; |
| 153 | listen = c.listen || process.env.LCHAT_LISTEN || "mention"; |
| 154 | } |
| 155 | } catch (e) { |
| 156 | tag(`loadConfig: ${e.message}`); |
| 157 | } |
| 158 | } |
| 159 | |
| 160 | // DMs to the bot: always answer. Groups: an explicit @Name is a guaranteed, |
| 161 | // deterministic reply. A bare name counts only in listen mode "name" — and |
| 162 | // even then it's ambiguous (people talk *about* a bot as often as *to* it), |
| 163 | // so it must also pass the classifier below. Anything else is ignored. |
| 164 | function replyTrigger(kind, text) { |
| 165 | if (kind === "dm") return "always"; |
| 166 | if (new RegExp(`(^|[^a-z0-9])@${name}\\b`, "i").test(text || "")) return "always"; |
| 167 | if (listen === "name" && new RegExp(`(^|[^a-z0-9])${name}\\b`, "i").test(text || "")) return "classify"; |
| 168 | return null; |
| 169 | } |
| 170 | |
| 171 | // The bare-name gate: one tiny yes/no call over the last few messages. Errs |
| 172 | // toward silence (failure or ambiguity = no reply) — @Name is the guaranteed |
| 173 | // path, so a false negative here is always recoverable by the sender. |
| 174 | async function isAddressed(threadId, key) { |
| 175 | const { messages = [] } = await get(`/api/messages?thread=${encodeURIComponent(threadId)}&limit=6`); |
| 176 | const lines = []; |
| 177 | for (const m of messages) { |
| 178 | try { |
| 179 | const p = await decrypt(key, m.blob); |
| 180 | if (p.text) lines.push(`${p.name}: ${p.text}`); |
| 181 | } catch {} |
| 182 | } |
| 183 | if (!lines.length) return false; |
| 184 | try { |
| 185 | const r = await api().messages.create({ |
| 186 | model: CLASSIFIER_MODEL, |
| 187 | max_tokens: 5, |
| 188 | system: |
| 189 | `"${name}" is a chat bot in a group conversation. Given the recent messages, decide whether ` + |
| 190 | `the LAST message is addressed to ${name} and expects ${name} to reply — as opposed to merely ` + |
| 191 | `mentioning ${name} or talking about ${name}. Answer with exactly "yes" or "no".`, |
| 192 | messages: [{ role: "user", content: lines.join("\n") }], |
| 193 | }); |
| 194 | return /yes/i.test( |
| 195 | r.content |
| 196 | .filter((b) => b.type === "text") |
| 197 | .map((b) => b.text) |
| 198 | .join("") |
| 199 | ); |
| 200 | } catch (e) { |
| 201 | tag(`addressed-classifier: ${e.message}`); |
| 202 | return false; |
| 203 | } |
| 204 | } |
| 205 | |
| 206 | // Build the Claude message list from recent thread history (already includes |
| 207 | // the triggering message, since the server stores before it broadcasts). |
| 208 | async function buildContext(threadId, key) { |
| 209 | const { messages = [] } = await get(`/api/messages?thread=${encodeURIComponent(threadId)}&limit=30`); |
| 210 | const out = []; |
| 211 | for (const m of messages) { |
| 212 | let p; |
| 213 | try { |
| 214 | p = await decrypt(key, m.blob); |
| 215 | } catch { |
| 216 | continue; // e.g. a message from before we held the key |
| 217 | } |
| 218 | if (!p.text) continue; |
| 219 | // Self-identity is the configured name, not the session file's record — so |
| 220 | // renaming a bot is an env/default change + roster relabel, no re-provision. |
| 221 | const mine = p.name === name; |
| 222 | let content = mine ? p.text : `${p.name}: ${p.text}`; |
| 223 | // Keep history truthful about tool calls: without this note the model |
| 224 | // sees its own past chip replies as bare text, learns that bare text |
| 225 | // "worked", and starts skipping the tool call — the chip vanishes. |
| 226 | if (mine && Array.isArray(p.actions) && p.actions.length) |
| 227 | content += `\n[note to self: this reply attached ${p.actions.length} button(s) via tool calls]`; |
| 228 | out.push({ role: mine ? "assistant" : "user", content }); |
| 229 | } |
| 230 | // Drop assistant turns from both ends (see trimContext): a leading one breaks |
| 231 | // the API's "first turn must be user" rule, and a trailing one is a hard 400 |
| 232 | // under thinking mode. |
| 233 | return trimContext(out); |
| 234 | } |
| 235 | |
| 236 | async function reply(threadId, entry) { |
| 237 | // Best-effort "…is typing" hint, re-beaten for the whole generation: a |
| 238 | // single beat's 15s expiry lapses under a long thinking run, so members |
| 239 | // watched the indicator die well before the reply landed. |
| 240 | sendTyping(threadId, entry.key); |
| 241 | const beat = setInterval(() => sendTyping(threadId, entry.key), 10_000); |
| 242 | try { |
| 243 | await replyInner(threadId, entry); |
| 244 | } finally { |
| 245 | clearInterval(beat); |
| 246 | // An errored/empty run sends no message (the guaranteed clear signal) — |
| 247 | // stop the indicator explicitly rather than leaving it to lapse. |
| 248 | sendTyping(threadId, entry.key, 0); |
| 249 | } |
| 250 | } |
| 251 | |
| 252 | async function replyInner(threadId, entry) { |
| 253 | const context = await buildContext(threadId, entry.key); |
| 254 | let text; |
| 255 | let actions = []; |
| 256 | try { |
| 257 | const r = await api().messages.create({ |
| 258 | model, |
| 259 | // Thinking, text and any tool call all share this budget — a tight cap |
| 260 | // truncates the tool_use off the end and the chip silently vanishes. |
| 261 | max_tokens: 4096, |
| 262 | system: systemBlocks(), |
| 263 | thinking: { type: "adaptive" }, |
| 264 | ...(tools ? { tools } : {}), |
| 265 | messages: context.length ? context : [{ role: "user", content: "Say hello." }], |
| 266 | }); |
| 267 | text = scrubSelfNotes( |
| 268 | r.content |
| 269 | .filter((b) => b.type === "text") |
| 270 | .map((b) => b.text) |
| 271 | .join(""), |
| 272 | ); |
| 273 | // Tool calls become one-tap chips on the message; there's no tool_result |
| 274 | // round-trip — the "result" is the member tapping (or not tapping) it. |
| 275 | if (tools) actions = actionsFromToolUse(r.content); |
| 276 | // Self-heal the promised-button case. Threads full of pre-annotation chip |
| 277 | // texts still bait the model into saying "tap to switch" WITHOUT the tool |
| 278 | // call (system-prompt rules lose to 25 in-context counter-examples). The |
| 279 | // promise is detectable, so make it deterministic: ask once more with a |
| 280 | // forced tool call. Forced tool_choice requires thinking off. |
| 281 | if (tools && !actions.length && /\b(tap|button|chip)\b/i.test(text || "")) { |
| 282 | tag(`reply promised a button without a tool call — retrying forced`); |
| 283 | const r2 = await api().messages.create({ |
| 284 | model, |
| 285 | max_tokens: 4096, |
| 286 | system: systemBlocks(), |
| 287 | thinking: { type: "disabled" }, |
| 288 | tools, |
| 289 | tool_choice: { type: "any" }, |
| 290 | messages: context.length ? context : [{ role: "user", content: "Say hello." }], |
| 291 | }); |
| 292 | actions = actionsFromToolUse(r2.content); |
| 293 | const t2 = scrubSelfNotes( |
| 294 | r2.content |
| 295 | .filter((b) => b.type === "text") |
| 296 | .map((b) => b.text) |
| 297 | .join(""), |
| 298 | ); |
| 299 | if (actions.length && t2) text = t2; // keep the original text if the retry was tool-only |
| 300 | } |
| 301 | if (!text && !actions.length) tag(`empty reply dropped (stop_reason=${r.stop_reason})`); |
| 302 | } catch (e) { |
| 303 | console.error(`[${name}] Claude error:`, e.message); |
| 304 | return; |
| 305 | } |
| 306 | if (!text && !actions.length) return; |
| 307 | await postReply(threadId, entry, text, actions); |
| 308 | } |
| 309 | |
| 310 | // Encrypt + send a reply payload (text + optional one-tap chips). Shared by the |
| 311 | // model path (replyInner) and the deterministic command path (processMsg). |
| 312 | async function postReply(threadId, entry, text, actions = []) { |
| 313 | if (!text && actions.length) text = "Tap to apply:"; |
| 314 | if (!text) return; |
| 315 | const payload = { name, text, ts: Date.now(), ...(actions.length ? { actions } : {}) }; |
| 316 | send({ type: "msg", thread_id: threadId, blob: await encrypt(entry.key, payload) }); |
| 317 | } |
| 318 | |
| 319 | // A connected-service capability op (docs/connected-services.md). The member |
| 320 | // sealed a time-boxed grant to THIS bot's public key and asked us to act with |
| 321 | // it. We open it with our own private key; a grant sealed to a different bot |
| 322 | // won't open, so we stay silent (it isn't ours). The third-party token lives |
| 323 | // only for this one call — we never persist it, and the window is one op wide. |
| 324 | async function handleConnect(threadId, entry, p) { |
| 325 | let grant; |
| 326 | try { |
| 327 | grant = await openSealed(p.grant || "", session.userPriv); |
| 328 | } catch { |
| 329 | return; // not sealed to us |
| 330 | } |
| 331 | // Reply ONLY with a non-rendering `connect.result` (matched by rid) — the |
| 332 | // DNSimple app renders the data itself. We never postReply here, so a |
| 333 | // connected-service op is NEVER visible as a bot message in the chat (the |
| 334 | // domain list belongs to the DNSimple app, not the bot's DM). Ops only ever |
| 335 | // come from the app, which always sets rid; a rid-less op gets no reply. |
| 336 | const reply = async (result) => { |
| 337 | if (!p.rid) return; |
| 338 | send({ |
| 339 | type: "msg", |
| 340 | thread_id: threadId, |
| 341 | blob: await encrypt(entry.key, { type: "connect.result", service: "dnsimple", op: p.op, rid: p.rid, ...result, ts: Date.now() }), |
| 342 | }); |
| 343 | }; |
| 344 | if (!grant || (grant.exp && grant.exp < Date.now())) { |
| 345 | await reply({ error: "That authorization expired — try again." }); |
| 346 | return; |
| 347 | } |
| 348 | if (grant.service !== "dnsimple") { |
| 349 | await reply({ error: `Can't act on "${grant.service}" yet.` }); |
| 350 | return; |
| 351 | } |
| 352 | try { |
| 353 | if (p.op !== "list") { |
| 354 | await reply({ error: `Unknown action "${p.op}".` }); |
| 355 | return; |
| 356 | } |
| 357 | const domains = await listDomains(grant.token); |
| 358 | await reply({ domains }); |
| 359 | } catch (e) { |
| 360 | await reply({ error: e.message }); |
| 361 | } finally { |
| 362 | grant.token = null; // drop the token |
| 363 | } |
| 364 | } |
| 365 | |
| 366 | // A connected-service OAuth exchange (docs/connected-services.md). The member |
| 367 | // did the PKCE dance in their browser — no key pasted — and hands us the |
| 368 | // one-time `code` + `verifier`. We run the server-side token exchange (public |
| 369 | // client, no secret), seal the token to the MEMBER's public key so only they |
| 370 | // can open it, and hand it straight back. We keep no copy. |
| 371 | async function handleOAuthExchange(threadId, entry, p) { |
| 372 | if (p.service !== "dnsimple") return; |
| 373 | let token; |
| 374 | try { |
| 375 | // The broker holds the client_secret and does the exchange; this sidecar |
| 376 | // just forwards the one-time code + state with its per-instance broker token. |
| 377 | token = await exchangeViaBroker({ |
| 378 | code: p.code, |
| 379 | state: p.state, |
| 380 | brokerUrl: process.env.LCHAT_CONNECT_BROKER_URL || "https://connect.elevenmessenger.com", |
| 381 | brokerToken: process.env.LCHAT_CONNECT_BROKER_TOKEN, |
| 382 | }); |
| 383 | } catch (e) { |
| 384 | tag(`connect: dnsimple exchange failed: ${e.message}`); |
| 385 | await postReply(threadId, entry, `Couldn't finish connecting DNSimple: ${e.message}`); |
| 386 | return; |
| 387 | } |
| 388 | try { |
| 389 | const { roster = [] } = await get("/api/roster"); |
| 390 | const sender = roster.find((e) => e.user_id === p.uid); |
| 391 | if (!sender || !sender.public_key) { |
| 392 | await postReply(threadId, entry, "Connected, but I couldn't find your key to hand it back — try again."); |
| 393 | return; |
| 394 | } |
| 395 | const sealed = await sealForPub(sender.public_key, { secret: token }); |
| 396 | send({ |
| 397 | type: "msg", |
| 398 | thread_id: threadId, |
| 399 | blob: await encrypt(entry.key, { type: "connect.token", service: "dnsimple", sealed, name, ts: Date.now() }), |
| 400 | }); |
| 401 | } finally { |
| 402 | token = null; // drop the token; the member holds the only copy now |
| 403 | } |
| 404 | } |
| 405 | |
| 406 | // Strip a leading "@Name" so a group command ("@Eleven /help") parses; a bare |
| 407 | // command in a DM passes through unchanged. |
| 408 | function stripMention(text) { |
| 409 | return (text || "").replace(new RegExp(`^\\s*@${name}\\b[:,]?\\s*`, "i"), "").trim(); |
| 410 | } |
| 411 | |
| 412 | // The reply when there's no model brain and the message wasn't a command. |
| 413 | function degradeText() { |
| 414 | return "I can help with a few quick commands — try /help. (Conversational AI answers aren't enabled on this server.)"; |
| 415 | } |
| 416 | |
| 417 | // Tell the server what we are so it can badge us honestly: "ai" when the model |
| 418 | // brain is available, else "bot" (deterministic). Re-declared on every connect, |
| 419 | // so toggling the key (a redeploy) flips the badge with no admin action. |
| 420 | async function declareKind() { |
| 421 | try { |
| 422 | await fetch(`${BASE}/api/bot-config`, { |
| 423 | method: "POST", |
| 424 | headers: { "content-type": "application/json" }, |
| 425 | body: JSON.stringify({ token: session.token, kind: AI_ENABLED ? "ai" : "bot" }), |
| 426 | }); |
| 427 | } catch (e) { |
| 428 | tag(`declareKind: ${e.message}`); |
| 429 | } |
| 430 | } |
| 431 | |
| 432 | async function sendTyping(threadId, key, expiry = 15) { |
| 433 | try { |
| 434 | // Match the live chat engine's typing wire shape: a relay-only ephemeral |
| 435 | // `chat.typing` event (apps/chat.js's `ephemeral` hook), NOT the legacy |
| 436 | // dedicated `type:"typing"` frame — that only feeds the retired #feed |
| 437 | // renderer in shell.js, so the bot's typing indicator never showed in the |
| 438 | // app. The transport injects our name; expiry 0 = the "stopped" signal. |
| 439 | send({ |
| 440 | type: "msg", |
| 441 | thread_id: threadId, |
| 442 | ephemeral: true, |
| 443 | blob: await encrypt(key, { type: "chat.typing", name, expiry }), |
| 444 | }); |
| 445 | } catch {} |
| 446 | } |
| 447 | |
| 448 | // ---- WebSocket with reconnect + a small outbound queue ----------------- |
| 449 | let ws; |
| 450 | let backoff = 1000; |
| 451 | let ping; // keepalive timer (the server closes idle sockets ~60s; it ignores "ping") |
| 452 | const queue = []; |
| 453 | |
| 454 | function send(obj) { |
| 455 | const data = JSON.stringify(obj); |
| 456 | if (ws && ws.readyState === WebSocket.OPEN) ws.send(data); |
| 457 | else queue.push(data); |
| 458 | } |
| 459 | |
| 460 | function connect() { |
| 461 | // `ws` sends no Origin header, so it passes the server's same-origin check. |
| 462 | ws = new WebSocket(`${WS_BASE}/ws?token=${encodeURIComponent(session.token)}`); |
| 463 | ws.on("open", async () => { |
| 464 | backoff = 1000; |
| 465 | tag(`connected to ${BASE}`); |
| 466 | clearInterval(ping); |
| 467 | ping = setInterval(() => { |
| 468 | try { |
| 469 | if (ws.readyState === WebSocket.OPEN) ws.send(JSON.stringify({ type: "ping" })); |
| 470 | } catch {} |
| 471 | }, 30000); |
| 472 | await loadConfig(); |
| 473 | await loadAgents(); |
| 474 | await loadThreads(); |
| 475 | await declareKind(); |
| 476 | await catchUp(); |
| 477 | for (const d of queue.splice(0)) ws.send(d); |
| 478 | }); |
| 479 | ws.on("message", (data) => onFrame(data).catch((e) => tag(`frame: ${e.message}`))); |
| 480 | ws.on("error", (e) => tag(`WS error: ${e.message}`)); |
| 481 | ws.on("close", () => { |
| 482 | clearInterval(ping); |
| 483 | tag(`disconnected; reconnecting in ${backoff}ms`); |
| 484 | setTimeout(connect, backoff); |
| 485 | backoff = Math.min(backoff * 2, 30000); |
| 486 | }); |
| 487 | } |
| 488 | |
| 489 | // Decrypt one message, advance the thread cursor, and reply if it's for us. |
| 490 | async function processMsg(threadId, id, blob) { |
| 491 | const entry = keys.get(threadId); |
| 492 | if (!entry) return; |
| 493 | // Idempotency gate. Message ids are monotonic and we only ever act on ones |
| 494 | // strictly newer than the highest we've seen on this thread, so a message |
| 495 | // delivered more than once — a reconnect's catch-up overlapping a live frame, |
| 496 | // or a socket that briefly double-delivered — is dropped here instead of |
| 497 | // answered twice (the cause of the occasional double bot reply). The read + |
| 498 | // advance are synchronous (no await between) so two concurrent deliveries of |
| 499 | // the same id can't both slip past. |
| 500 | const seen = cursors.get(threadId) || 0; |
| 501 | if (id && id <= seen) return; |
| 502 | cursors.set(threadId, id || seen); |
| 503 | let p; |
| 504 | try { |
| 505 | p = await decrypt(entry.key, blob); |
| 506 | } catch { |
| 507 | return; |
| 508 | } |
| 509 | if (p.name === name) return; // our own message, echoed back |
| 510 | if (agentNames.has(p.name)) return; // bots never talk to bots (see loadAgents) |
| 511 | // A direct capability invocation (a sealed connected-service grant), not a |
| 512 | // chat trigger — handle it whether or not we have a model brain. |
| 513 | if (p.type === "connect.op") { |
| 514 | await handleConnect(threadId, entry, p); |
| 515 | return; |
| 516 | } |
| 517 | if (p.type === "connect.oauth") { |
| 518 | await handleOAuthExchange(threadId, entry, p); |
| 519 | return; |
| 520 | } |
| 521 | if (p.type === "flow.state") { |
| 522 | // A client-run scripted flow (the one-Eleven welcome, running in this |
| 523 | // bot's own DM) taking or ceding the floor — the member's typed answers |
| 524 | // belong to the script, not the brain. DMs only; hard expiry so a dead |
| 525 | // client can't mute the bot forever. Mirrors the Go runtime's handleTyped. |
| 526 | if (entry.kind === "dm") { |
| 527 | if (p.active) muted.set(threadId, Date.now() + 15 * 60 * 1000); |
| 528 | else muted.delete(threadId); |
| 529 | } |
| 530 | return; |
| 531 | } |
| 532 | if (p.type) return; // any other typed payload (a mini-app event), not chat text |
| 533 | const exp = muted.get(threadId); |
| 534 | if (exp) { |
| 535 | if (Date.now() < exp) return; // the script has the floor |
| 536 | muted.delete(threadId); |
| 537 | } |
| 538 | const trigger = replyTrigger(entry.kind, p.text); |
| 539 | if (!trigger) return; |
| 540 | // The bare-name gate needs the model; with no AI there's nothing to classify. |
| 541 | if (trigger === "classify" && (!AI_ENABLED || !(await isAddressed(threadId, entry.key)))) return; |
| 542 | // Deterministic command floor first — works with or without a model brain. |
| 543 | if (commands) { |
| 544 | const res = runCommand(stripMention(p.text), { aiEnabled: AI_ENABLED, name }); |
| 545 | if (res) { |
| 546 | await postReply(threadId, entry, res.text, res.actions || []); |
| 547 | return; |
| 548 | } |
| 549 | } |
| 550 | // Conversational upgrade: only when the model brain is available. |
| 551 | if (AI_ENABLED) { |
| 552 | await reply(threadId, entry); |
| 553 | return; |
| 554 | } |
| 555 | // No AI and not a command → a deterministic hint (command-capable bots only). |
| 556 | if (commands) await postReply(threadId, entry, degradeText(), []); |
| 557 | } |
| 558 | |
| 559 | async function onFrame(data) { |
| 560 | let f; |
| 561 | try { |
| 562 | f = JSON.parse(data); |
| 563 | } catch { |
| 564 | return; |
| 565 | } |
| 566 | // Ignore reactions/edits/deletes/typing, and — crucially — ephemeral frames: |
| 567 | // relay-only presence traffic (the chat app's "…is typing" beat, live cursors) |
| 568 | // arrives as `type:"msg"` with no id (see hub.go), so the type check alone lets |
| 569 | // it through. Without this a DM bot (replyTrigger "always") answers every beat. |
| 570 | if (f.type !== "msg" || !f.thread_id || f.ephemeral) return; |
| 571 | if (!keys.has(f.thread_id)) await loadThreads(); // a thread we were just added to |
| 572 | await processMsg(f.thread_id, f.id, f.blob); |
| 573 | } |
| 574 | |
| 575 | // On (re)connect, catch up on anything that arrived while disconnected, so a |
| 576 | // message sent during a reconnect blip still gets a reply. Cursors start at |
| 577 | // each thread's latest on first sight, so the initial connect backfills nothing. |
| 578 | async function catchUp() { |
| 579 | for (const threadId of keys.keys()) { |
| 580 | const after = cursors.get(threadId) || 0; |
| 581 | try { |
| 582 | const { messages = [] } = await get(`/api/messages?thread=${encodeURIComponent(threadId)}&after=${after}`); |
| 583 | for (const m of messages) await processMsg(threadId, m.id, m.blob); |
| 584 | } catch (e) { |
| 585 | tag(`catchUp ${threadId}: ${e.message}`); |
| 586 | } |
| 587 | } |
| 588 | } |
| 589 | |
| 590 | connect(); |
| 591 | } |
| 592 | |
amadan