| 1 | package bot |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "fmt" |
| 7 | "io" |
| 8 | "net/http" |
| 9 | "net/url" |
| 10 | "regexp" |
| 11 | "strings" |
| 12 | "sync" |
| 13 | "sync/atomic" |
| 14 | "time" |
| 15 | |
| 16 | "github.com/coder/websocket" |
| 17 | |
| 18 | "lchat/internal/e2ee" |
| 19 | ) |
| 20 | |
| 21 | // Runtime is the Go port of agent/runtime.mjs's persona-agnostic core: one bot |
| 22 | // identity's live connection to an instance. It dials the WebSocket, unwraps |
| 23 | // thread keys, decrypts what arrives, and hands anything that triggers the bot |
| 24 | // (a DM, or an @mention in a group) to Handler — the seam where a brain plugs |
| 25 | // in. Today that's the deterministic command floor (CommandFloor); the recipe |
| 26 | // engine and the model layer slot into the same seam next (docs/bots.md). |
| 27 | // |
| 28 | // One process hosts many bots: each Runtime is self-contained (own session, |
| 29 | // socket, keys, cursors), so `lchat bot run` starts one goroutine per identity. |
| 30 | // Like every bot, it is a pure outbound client — the private key in Session |
| 31 | // never leaves this process, and the server only ever sees ciphertext. |
| 32 | type Runtime struct { |
| 33 | Base string |
| 34 | Session *Session |
| 35 | // Handler decides the response to a message. The ThreadCtx gives it the |
| 36 | // thread's services (history, typing, the model brain); return ok=false to |
| 37 | // stay silent. |
| 38 | Handler func(*ThreadCtx, Incoming) (Reply, bool) |
| 39 | // AIKey is the ANTHROPIC_API_KEY. Set, the runtime self-declares kind "ai" |
| 40 | // and the model brain (brain.go) is live; empty, the bot is deterministic |
| 41 | // and badges "bot". |
| 42 | AIKey string |
| 43 | // Ready, if set, is called once each time the runtime finishes connecting — |
| 44 | // right before it starts reading frames, so a supervisor (or a test) knows |
| 45 | // the bot is live and will receive broadcasts. |
| 46 | Ready func() |
| 47 | Logf func(format string, args ...any) |
| 48 | // Charters marks this identity as the CHARTER HOST: the one runtime in a |
| 49 | // sidecar that runs member bots (docs/member-bots.md) — it arms their |
| 50 | // schedules in its self-thread, executes due tasks, and offers the |
| 51 | // create_bot tool. Exactly one identity per instance must host, or every |
| 52 | // charter would run once per host (botcmd elects it). |
| 53 | Charters bool |
| 54 | |
| 55 | client *http.Client |
| 56 | ai *anthropicClient |
| 57 | meter *usageMeter // model-spend accounting, flushed to the instance (#857, meter.go) |
| 58 | // pendingUsage is the one usage report in flight — kept verbatim across |
| 59 | // failed flushes so the ledger can dedupe a replay whose first ack was |
| 60 | // lost. Touched only on the meterLoop goroutine. |
| 61 | pendingUsage *usageReport |
| 62 | // budgetNotified arms the once-per-exhaustion notice (meter.go). |
| 63 | budgetNotified atomic.Bool |
| 64 | // budgetState is the space's AI budget as of the last usage flush ("ok", |
| 65 | // "unmetered", …). Inert today — phase 3 (docs/ai-provisioning.md) gates |
| 66 | // the model path on it. |
| 67 | budgetState atomic.Value |
| 68 | connect *connectConfig // connected-services (docs/connected-services.md) |
| 69 | keys map[string]threadEntry |
| 70 | cursors map[string]int64 // threadId -> last message id we've processed |
| 71 | agents map[string]bool // roster labels of ALL agents — never reply to one |
| 72 | mention *regexp.Regexp |
| 73 | bare *regexp.Regexp // bare-name mention, for listen mode "name" |
| 74 | |
| 75 | // Console-managed config (loadConfig, re-fetched each connect). |
| 76 | persona string |
| 77 | model string |
| 78 | docs string |
| 79 | listen string |
| 80 | // The instance's sandboxed-app authoring guide (GET /api/member-app-guide, |
| 81 | // fetched each connect). Non-empty enables the create_app tool — the bot |
| 82 | // builds against the instance's own rules, so the two can never disagree. |
| 83 | appGuide string |
| 84 | |
| 85 | wmu sync.Mutex // coder/websocket allows one concurrent writer |
| 86 | conn *websocket.Conn |
| 87 | |
| 88 | // building guards against duplicate concurrent app builds: a member who |
| 89 | // re-asks during the ~minute a build takes (thinking it hung) would otherwise |
| 90 | // get the app built twice. Keyed by thread + requester uid (appbuild.go). |
| 91 | buildMu sync.Mutex |
| 92 | building map[string]bool |
| 93 | |
| 94 | // The decrypted member-app list behind memberAppsNote (appbuild.go), |
| 95 | // cached briefly so every model reply doesn't refetch + re-decrypt every |
| 96 | // deployed blob. Guarded by appsMu; invalidated after this sidecar's own |
| 97 | // deploys, refreshed after appsNoteTTL otherwise. |
| 98 | appsMu sync.Mutex |
| 99 | apps []deployedApp |
| 100 | appsAt time.Time |
| 101 | |
| 102 | // The charter host's self-thread (id + key), snapshotted at connect so the |
| 103 | // builder goroutine and the reconcile ticker can read it without touching |
| 104 | // r.keys (which stays single-goroutine, owned by the frame loop). |
| 105 | selfMu sync.Mutex |
| 106 | selfID string |
| 107 | selfKey []byte |
| 108 | |
| 109 | // muted holds threads where a client-run scripted flow has the floor |
| 110 | // (flow.state events — the one-Eleven welcome): the brain stays quiet |
| 111 | // there until the script ends or the entry expires — the fail-safe for a |
| 112 | // client that dies mid-flow. Frame goroutine only. |
| 113 | muted map[string]time.Time |
| 114 | |
| 115 | // charterRuns records when each charter last ran — the duplicate-tick |
| 116 | // guard for the connect-time catch-up arm (see charterTick). Frame |
| 117 | // goroutine only, like muted. |
| 118 | charterRuns map[string]time.Time |
| 119 | |
| 120 | // charterFails counts consecutive failed runs per charter — the backoff |
| 121 | // dial (charterRunFailed, chartersched.go). Unlike charterRuns it has its |
| 122 | // own lock: ticks bump it on the frame goroutine, but a successful build |
| 123 | // or a removal clears it from the builder goroutine. |
| 124 | failMu sync.Mutex |
| 125 | charterFails map[string]int |
| 126 | |
| 127 | // charterHomes maps each member bot's dedicated thread to (bot id, owner) |
| 128 | // — what makes the bot's own chat conversational: the owner talking there |
| 129 | // is always talking to the host, no @mention needed, and the brain gets |
| 130 | // the charter as context (charterHomeNote, brain.go). Rebuilt whole on |
| 131 | // every listCharters, so builds, edits, and removals all refresh it. Own |
| 132 | // lock: the frame goroutine reads per message; reconcile + builder |
| 133 | // goroutines write. |
| 134 | homeMu sync.Mutex |
| 135 | charterHomes map[string]charterHome |
| 136 | } |
| 137 | |
| 138 | // charterHome is one entry of charterHomes: which charter a thread belongs to |
| 139 | // and whose bot it is. |
| 140 | type charterHome struct { |
| 141 | BotID string |
| 142 | Owner string |
| 143 | } |
| 144 | |
| 145 | // ThreadCtx hands a Handler the thread's services: decrypted history, the |
| 146 | // typing indicator, and the model brain — everything a reply path needs |
| 147 | // beyond the triggering message itself. |
| 148 | type ThreadCtx struct { |
| 149 | rt *Runtime |
| 150 | ID string |
| 151 | Kind string // "dm" or "group" |
| 152 | key []byte |
| 153 | // Who sent the triggering message — the app-builder records the requester |
| 154 | // (SenderUID gates later updates to their app; see appbuild.go). |
| 155 | Sender string |
| 156 | SenderUID string |
| 157 | } |
| 158 | |
| 159 | // HistoryMsg is one decrypted message from the thread's recent history. |
| 160 | type HistoryMsg struct { |
| 161 | ID int64 |
| 162 | Name string |
| 163 | Text string |
| 164 | Mine bool // authored by this bot (by payload name) |
| 165 | ActionCount int // chips the message carried (history truthfulness — brain.go) |
| 166 | } |
| 167 | |
| 168 | // History returns the thread's latest messages, oldest first, decrypted. |
| 169 | // Undecryptable or non-chat payloads are skipped. |
| 170 | func (tc *ThreadCtx) History(limit int) ([]HistoryMsg, error) { |
| 171 | var ml struct { |
| 172 | Messages []struct { |
| 173 | ID int64 `json:"id"` |
| 174 | Blob string `json:"blob"` |
| 175 | TombstoneOf int64 `json:"tombstone_of"` |
| 176 | } `json:"messages"` |
| 177 | } |
| 178 | // Over-fetch: delete tombstones (#581), reactions and app events ride the |
| 179 | // window too, and each would otherwise shrink the model's context — a |
| 180 | // thread with a burst of unsends could leave the brain near-blind (#765). |
| 181 | // The server caps limit at 200; trim back to `limit` real lines below. |
| 182 | fetch := limit * 2 |
| 183 | if fetch > 200 { |
| 184 | fetch = 200 |
| 185 | } |
| 186 | path := fmt.Sprintf("/api/messages?thread=%s&limit=%d", url.QueryEscape(tc.ID), fetch) |
| 187 | if err := tc.rt.get(path, &ml); err != nil { |
| 188 | return nil, err |
| 189 | } |
| 190 | var out []HistoryMsg |
| 191 | for _, m := range ml.Messages { |
| 192 | if m.TombstoneOf != 0 { |
| 193 | continue // an unsent message's marker, not a line anyone said |
| 194 | } |
| 195 | var p struct { |
| 196 | Type string `json:"type"` |
| 197 | Name string `json:"name"` |
| 198 | Text string `json:"text"` |
| 199 | Actions []map[string]any `json:"actions"` |
| 200 | } |
| 201 | if e2ee.DecryptJSON(tc.key, m.Blob, &p) != nil || p.Type != "" { |
| 202 | continue |
| 203 | } |
| 204 | out = append(out, HistoryMsg{ |
| 205 | ID: m.ID, Name: p.Name, Text: p.Text, |
| 206 | Mine: p.Name == tc.rt.Session.Name, ActionCount: len(p.Actions), |
| 207 | }) |
| 208 | } |
| 209 | if len(out) > limit { |
| 210 | out = out[len(out)-limit:] // newest `limit` — oldest-first order kept |
| 211 | } |
| 212 | return out, nil |
| 213 | } |
| 214 | |
| 215 | // Typing sends a best-effort "…is typing" beat (expiry seconds; 0 clears it |
| 216 | // now). Same relay-only ephemeral chat.typing shape the live chat engine uses. |
| 217 | func (tc *ThreadCtx) Typing(expiry int) { |
| 218 | blob, err := e2ee.EncryptJSON(tc.key, map[string]any{"type": "chat.typing", "name": tc.rt.Session.Name, "expiry": expiry}) |
| 219 | if err != nil { |
| 220 | return |
| 221 | } |
| 222 | tc.rt.send(map[string]any{"type": "msg", "thread_id": tc.ID, "ephemeral": true, "blob": blob}) |
| 223 | } |
| 224 | |
| 225 | // ModelReply asks the model brain for a conversational reply (recipes reach it |
| 226 | // via the "model" action). ok=false when there's no API key, the call fails, |
| 227 | // or the reply is empty — the caller stays silent or falls through. |
| 228 | func (tc *ThreadCtx) ModelReply() (Reply, bool) { return tc.rt.modelReply(tc) } |
| 229 | |
| 230 | // actionLine starts a /me-style status line in this thread that the caller |
| 231 | // updates in place — a handy way for any handler to report ongoing work without |
| 232 | // stacking messages (see the actionLine type in runtime.go). |
| 233 | func (tc *ThreadCtx) actionLine(text string) *actionLine { |
| 234 | return tc.rt.newActionLine(tc.ID, tc.key, text) |
| 235 | } |
| 236 | |
| 237 | // Incoming is one decrypted chat message. The Runtime hands the Handler every |
| 238 | // person-authored message it can read (that's what lets watcher recipes exist); |
| 239 | // Triggered says whether the bot was actually addressed — a DM, or a group |
| 240 | // @mention — and mention-gated handlers must check it before acting. |
| 241 | type Incoming struct { |
| 242 | ThreadID string |
| 243 | Kind string // "dm" or "group" |
| 244 | Sender string |
| 245 | Text string // raw text, @mention included (see mentionStripper) |
| 246 | Triggered bool // a DM, or an explicit @mention in a group |
| 247 | } |
| 248 | |
| 249 | // Reply is what a Handler posts back: text plus optional one-tap action chips |
| 250 | // (the same client-whitelisted `actions` payload the Node sidecar ships). |
| 251 | type Reply struct { |
| 252 | Text string |
| 253 | Actions []map[string]any |
| 254 | } |
| 255 | |
| 256 | type threadEntry struct { |
| 257 | key []byte |
| 258 | kind string |
| 259 | } |
| 260 | |
| 261 | // Run connects and serves until ctx is cancelled, reconnecting with backoff on |
| 262 | // any drop — the same loop shape as the Node runtime's ws.on("close"). |
| 263 | func (r *Runtime) Run(ctx context.Context) error { |
| 264 | // markTransport stamps every REST call with SidecarHeader (sidecarmark.go) |
| 265 | // so a hibernating instance's router can exclude bot traffic from its idle |
| 266 | // clock; a non-hibernating instance ignores it. |
| 267 | r.client = &http.Client{Timeout: 30 * time.Second, Transport: markTransport{}} |
| 268 | r.keys = map[string]threadEntry{} |
| 269 | r.cursors = map[string]int64{} |
| 270 | r.agents = map[string]bool{} |
| 271 | r.muted = map[string]time.Time{} |
| 272 | r.charterRuns = map[string]time.Time{} |
| 273 | r.charterFails = map[string]int{} |
| 274 | r.charterHomes = map[string]charterHome{} |
| 275 | r.mention = regexp.MustCompile(`(?i)(^|[^a-z0-9])@` + regexp.QuoteMeta(r.Session.Name) + `\b`) |
| 276 | r.bare = regexp.MustCompile(`(?i)(^|[^a-z0-9])` + regexp.QuoteMeta(r.Session.Name) + `\b`) |
| 277 | r.meter = newUsageMeter() |
| 278 | if r.AIKey != "" { |
| 279 | r.ai = newAnthropicClient(r.AIKey, r.recordUsage) |
| 280 | } |
| 281 | r.connect = loadConnectConfig() |
| 282 | // The usage meter flushes on its own clock, independent of connection |
| 283 | // state — a failed flush restores the counters and the next tick retries. |
| 284 | go r.meterLoop(ctx) |
| 285 | if r.Charters { |
| 286 | // The slow reconcile loop: heals lost ticks, drops alarms for deleted |
| 287 | // charters, and arms charters that appeared while we were connected. |
| 288 | // It never touches r.keys — only the self-thread snapshot + REST. |
| 289 | go func() { |
| 290 | t := time.NewTicker(5 * time.Minute) |
| 291 | defer t.Stop() |
| 292 | for { |
| 293 | select { |
| 294 | case <-t.C: |
| 295 | r.charterReconcile(false) |
| 296 | case <-ctx.Done(): |
| 297 | return |
| 298 | } |
| 299 | } |
| 300 | }() |
| 301 | } |
| 302 | backoff := time.Second |
| 303 | for { |
| 304 | start := time.Now() |
| 305 | err := r.connectOnce(ctx) |
| 306 | if ctx.Err() != nil { |
| 307 | return ctx.Err() |
| 308 | } |
| 309 | // A connection that held for a while earned a fresh backoff. |
| 310 | if time.Since(start) > time.Minute { |
| 311 | backoff = time.Second |
| 312 | } |
| 313 | r.logf("disconnected (%v); reconnecting in %s", err, backoff) |
| 314 | select { |
| 315 | case <-time.After(backoff): |
| 316 | case <-ctx.Done(): |
| 317 | return ctx.Err() |
| 318 | } |
| 319 | backoff = min(backoff*2, 30*time.Second) |
| 320 | } |
| 321 | } |
| 322 | |
| 323 | func (r *Runtime) connectOnce(ctx context.Context) error { |
| 324 | wsBase := strings.Replace(r.Base, "http", "ws", 1) |
| 325 | // The WebSocket carries the sidecar marker too — it's the long-lived |
| 326 | // connection the activity exclusion exists for (sidecarmark.go). |
| 327 | conn, _, err := websocket.Dial(ctx, wsBase+"/ws?token="+url.QueryEscape(r.Session.Token), &websocket.DialOptions{ |
| 328 | HTTPHeader: http.Header{SidecarHeader: []string{"1"}}, |
| 329 | }) |
| 330 | if err != nil { |
| 331 | return err |
| 332 | } |
| 333 | defer conn.CloseNow() |
| 334 | // The server caps blobs at 64K; the library's 32K default would kill the |
| 335 | // socket on the first big message. |
| 336 | conn.SetReadLimit(256 * 1024) |
| 337 | r.conn = conn |
| 338 | defer func() { r.conn = nil }() |
| 339 | |
| 340 | // Same setup as the Node runtime's ws.on("open"): pull console config, |
| 341 | // learn the roster's bots, unwrap thread keys, declare what we are, then |
| 342 | // backfill anything missed. Order matters only in that declareKind is a |
| 343 | // visible "connected and caught up to here" signal; the cursor gate makes |
| 344 | // catchUp vs live frames safe in any interleaving. |
| 345 | r.loadConfig() |
| 346 | r.loadAgents() |
| 347 | r.loadThreads() |
| 348 | r.declareKind() |
| 349 | r.catchUp() |
| 350 | // The charter host reconciles member-bot alarms on every connect (and |
| 351 | // keeps reconciling on a slow tick — see Run) so a tick lost to a crash |
| 352 | // between delivery and re-arm heals itself. |
| 353 | if r.Charters { |
| 354 | r.charterReconcile(true) |
| 355 | } |
| 356 | r.logf("connected to %s", r.Base) |
| 357 | |
| 358 | // Keepalive: the server closes idle sockets and ignores "ping" frames. |
| 359 | pingCtx, cancel := context.WithCancel(ctx) |
| 360 | defer cancel() |
| 361 | go func() { |
| 362 | t := time.NewTicker(30 * time.Second) |
| 363 | defer t.Stop() |
| 364 | for { |
| 365 | select { |
| 366 | case <-t.C: |
| 367 | r.send(map[string]any{"type": "ping"}) |
| 368 | case <-pingCtx.Done(): |
| 369 | return |
| 370 | } |
| 371 | } |
| 372 | }() |
| 373 | |
| 374 | if r.Ready != nil { |
| 375 | r.Ready() |
| 376 | } |
| 377 | for { |
| 378 | _, data, err := conn.Read(ctx) |
| 379 | if err != nil { |
| 380 | return err |
| 381 | } |
| 382 | r.onFrame(data) |
| 383 | } |
| 384 | } |
| 385 | |
| 386 | type wsFrame struct { |
| 387 | Type string `json:"type"` |
| 388 | ThreadID string `json:"thread_id"` |
| 389 | ID int64 `json:"id"` |
| 390 | Blob string `json:"blob"` |
| 391 | Ephemeral bool `json:"ephemeral"` |
| 392 | } |
| 393 | |
| 394 | func (r *Runtime) onFrame(data []byte) { |
| 395 | var f wsFrame |
| 396 | if json.Unmarshal(data, &f) != nil { |
| 397 | return |
| 398 | } |
| 399 | // Ignore reactions/edits/deletes/typing, and — crucially — ephemeral frames: |
| 400 | // relay-only presence traffic (typing beats, live cursors) arrives as |
| 401 | // type:"msg" with no id, so the type check alone would let it through and a |
| 402 | // DM bot would answer every beat. |
| 403 | if f.Type != "msg" || f.ThreadID == "" || f.Ephemeral { |
| 404 | return |
| 405 | } |
| 406 | if _, ok := r.keys[f.ThreadID]; !ok { |
| 407 | r.loadThreads() // a thread we were just added to |
| 408 | // loadThreads baselines a new thread's cursor at its latest message — |
| 409 | // which is the very frame that made us load it (the server stores |
| 410 | // before it broadcasts), so the gate below would swallow it. Rewind so |
| 411 | // the triggering message gets an answer; anything older stays history. |
| 412 | // (The Node runtime drops this first message — a quirk, not mirrored.) |
| 413 | if f.ID != 0 && r.cursors[f.ThreadID] >= f.ID { |
| 414 | r.cursors[f.ThreadID] = f.ID - 1 |
| 415 | } |
| 416 | } |
| 417 | r.processMsg(f.ThreadID, f.ID, f.Blob) |
| 418 | } |
| 419 | |
| 420 | // chatPayload is the decrypted message shape the runtime cares about. Typed |
| 421 | // payloads (connect.*, mini-app events) are protocol traffic, not chat lines. |
| 422 | type chatPayload struct { |
| 423 | Type string `json:"type"` |
| 424 | Name string `json:"name"` |
| 425 | Text string `json:"text"` |
| 426 | UID string `json:"uid"` |
| 427 | } |
| 428 | |
| 429 | // processMsg decrypts one message, advances the thread cursor, and replies if |
| 430 | // it's for us. Frames and catch-up pages are processed on one goroutine per |
| 431 | // bot, so the cursor read+advance can't race — a message delivered twice (a |
| 432 | // reconnect's catch-up overlapping a live frame) is dropped here instead of |
| 433 | // answered twice. |
| 434 | func (r *Runtime) processMsg(threadID string, id int64, blob string) { |
| 435 | entry, ok := r.keys[threadID] |
| 436 | if !ok { |
| 437 | return |
| 438 | } |
| 439 | if id != 0 { |
| 440 | if id <= r.cursors[threadID] { |
| 441 | return |
| 442 | } |
| 443 | r.cursors[threadID] = id |
| 444 | } |
| 445 | var p chatPayload |
| 446 | if e2ee.DecryptJSON(entry.key, blob, &p) != nil { |
| 447 | return // e.g. a message from before we held the key |
| 448 | } |
| 449 | // Typed payloads are protocol traffic, not chat lines — route them ahead of |
| 450 | // the chat-path drops. A scheduled bot.tick is authored by this bot by |
| 451 | // design (the own-echo skip below would swallow it), and capability ops |
| 452 | // must work whoever posted them. The anti-ping-pong property survives |
| 453 | // because handleTyped never produces chat: typed traffic is answered with |
| 454 | // typed traffic or not at all. |
| 455 | if p.Type != "" { |
| 456 | r.handleTyped(threadID, entry, p.Type, blob) |
| 457 | return |
| 458 | } |
| 459 | if p.Name == r.Session.Name { |
| 460 | return // our own message, echoed back |
| 461 | } |
| 462 | if r.agents[p.Name] { |
| 463 | return // bots never talk to bots (see loadAgents) |
| 464 | } |
| 465 | if r.Handler == nil { |
| 466 | return |
| 467 | } |
| 468 | // A scripted flow has the floor here (flow.state): the client is walking |
| 469 | // the member through a conversation AS this bot, and their typed answers |
| 470 | // are the script's, not the brain's. One voice at a time. |
| 471 | if exp, ok := r.muted[threadID]; ok { |
| 472 | if time.Now().Before(exp) { |
| 473 | return |
| 474 | } |
| 475 | delete(r.muted, threadID) |
| 476 | } |
| 477 | tc := &ThreadCtx{rt: r, ID: threadID, Kind: entry.kind, key: entry.key, Sender: p.Name, SenderUID: p.UID} |
| 478 | rep, ok := r.Handler(tc, Incoming{ |
| 479 | ThreadID: threadID, |
| 480 | Kind: entry.kind, |
| 481 | Sender: p.Name, |
| 482 | Text: p.Text, |
| 483 | Triggered: r.triggered(tc, p.Text, p.UID), |
| 484 | }) |
| 485 | if !ok { |
| 486 | return |
| 487 | } |
| 488 | r.postReply(threadID, entry.key, rep) |
| 489 | } |
| 490 | |
| 491 | // handleTyped routes one typed (non-chat) payload. Unknown types — mini-app |
| 492 | // events, our own echoed connect.result — are silently ignored, exactly as |
| 493 | // the old `p.Type != "" → return` chokepoint did. |
| 494 | func (r *Runtime) handleTyped(threadID string, entry threadEntry, typ, blob string) { |
| 495 | switch typ { |
| 496 | case "bot.tick": |
| 497 | // A charter alarm (chartersched.go). Only honoured from this bot's own |
| 498 | // self-thread: nobody else is a member there, so nobody else can ring |
| 499 | // the bot's alarms early or repeatedly. |
| 500 | if !r.Charters || entry.kind != "self" { |
| 501 | return |
| 502 | } |
| 503 | var tp tickPayload |
| 504 | if e2ee.DecryptJSON(entry.key, blob, &tp) != nil { |
| 505 | return |
| 506 | } |
| 507 | r.charterTick(threadID, entry.key, &tp) |
| 508 | case "connect.research": |
| 509 | // A tapped research chip: sealed read grants + one question |
| 510 | // (research.go). Validation + ack run here; the loop takes its own |
| 511 | // goroutine so a minute of research never blocks this pump. |
| 512 | var rp researchPayload |
| 513 | if e2ee.DecryptJSON(entry.key, blob, &rp) != nil { |
| 514 | return |
| 515 | } |
| 516 | r.handleResearch(threadID, entry.key, &rp) |
| 517 | case "connect.op", "connect.oauth": |
| 518 | // A direct capability invocation (a sealed connected-services grant) — |
| 519 | // handled whether or not the bot has a recipe/brain (connect.go). |
| 520 | var cp connectPayload |
| 521 | if e2ee.DecryptJSON(entry.key, blob, &cp) != nil { |
| 522 | return |
| 523 | } |
| 524 | if cp.Type == "connect.op" { |
| 525 | r.handleConnect(threadID, entry.key, &cp) |
| 526 | } else { |
| 527 | r.handleOAuthExchange(threadID, entry.key, &cp) |
| 528 | } |
| 529 | case "flow.state": |
| 530 | // A client-run scripted flow (the one-Eleven welcome, running in this |
| 531 | // bot's own DM) taking or ceding the floor. DMs only, with a hard |
| 532 | // expiry so a client that dies mid-flow can't mute the bot forever. |
| 533 | if entry.kind != "dm" { |
| 534 | return |
| 535 | } |
| 536 | var fs struct { |
| 537 | Active bool `json:"active"` |
| 538 | } |
| 539 | if e2ee.DecryptJSON(entry.key, blob, &fs) != nil { |
| 540 | return |
| 541 | } |
| 542 | if fs.Active { |
| 543 | r.muted[threadID] = time.Now().Add(15 * time.Minute) |
| 544 | } else { |
| 545 | delete(r.muted, threadID) |
| 546 | } |
| 547 | } |
| 548 | } |
| 549 | |
| 550 | // triggered decides whether the bot was addressed. DMs always; groups on an |
| 551 | // explicit @Name — except a member bot's home thread, where the OWNER talking |
| 552 | // is always talking to the host (the thread holds no one else, and "Hello?" |
| 553 | // going unanswered there reads as the bot ignoring them). The uid gate |
| 554 | // matters: the bot's own editions post under the persona name, not ours, so |
| 555 | // without it the host would answer its own bot every occurrence. Listen mode |
| 556 | // "name" (console config) also counts a bare-name mention — but a bare name |
| 557 | // is ambiguous (people talk *about* a bot as often as *to* it), so it must |
| 558 | // additionally pass the model classifier, which means the mode does nothing |
| 559 | // without an API key. |
| 560 | func (r *Runtime) triggered(tc *ThreadCtx, text, uid string) bool { |
| 561 | if tc.Kind == "dm" { |
| 562 | return true |
| 563 | } |
| 564 | if r.mention.MatchString(text) { |
| 565 | return true |
| 566 | } |
| 567 | if r.ai != nil && uid != "" { |
| 568 | if h, ok := r.charterHomeFor(tc.ID); ok && h.Owner == uid { |
| 569 | return true |
| 570 | } |
| 571 | } |
| 572 | if r.listen == "name" && r.ai != nil && r.bare.MatchString(text) { |
| 573 | return r.isAddressed(tc) |
| 574 | } |
| 575 | return false |
| 576 | } |
| 577 | |
| 578 | // charterHomeFor looks a thread up in the charter-home map (any goroutine). |
| 579 | func (r *Runtime) charterHomeFor(threadID string) (charterHome, bool) { |
| 580 | r.homeMu.Lock() |
| 581 | defer r.homeMu.Unlock() |
| 582 | h, ok := r.charterHomes[threadID] |
| 583 | return h, ok |
| 584 | } |
| 585 | |
| 586 | // setCharterHomes replaces the charter-home map with the store's current |
| 587 | // truth — called from listCharters, so every path that reads charters |
| 588 | // (connect + slow reconcile, ticks, builds, removals) keeps it fresh. |
| 589 | func (r *Runtime) setCharterHomes(entries []charterEntry) { |
| 590 | homes := make(map[string]charterHome, len(entries)) |
| 591 | for _, e := range entries { |
| 592 | if e.Charter.Thread != "" { |
| 593 | homes[e.Charter.Thread] = charterHome{BotID: e.ID, Owner: e.Charter.Owner} |
| 594 | } |
| 595 | } |
| 596 | r.homeMu.Lock() |
| 597 | r.charterHomes = homes |
| 598 | r.homeMu.Unlock() |
| 599 | } |
| 600 | |
| 601 | // postReply encrypts and sends one chat reply payload — the same wire shape the |
| 602 | // Node sidecar posts, so clients can't tell the implementations apart. |
| 603 | func (r *Runtime) postReply(threadID string, key []byte, rep Reply) { |
| 604 | if rep.Text == "" && len(rep.Actions) > 0 { |
| 605 | rep.Text = "Tap to apply:" |
| 606 | } |
| 607 | if rep.Text == "" { |
| 608 | return |
| 609 | } |
| 610 | payload := map[string]any{"name": r.Session.Name, "text": rep.Text, "ts": time.Now().UnixMilli()} |
| 611 | if len(rep.Actions) > 0 { |
| 612 | payload["actions"] = rep.Actions |
| 613 | } |
| 614 | if err := r.sendPayload(threadID, key, payload); err != nil { |
| 615 | r.logf("send reply: %v", err) |
| 616 | } |
| 617 | } |
| 618 | |
| 619 | // sendPayload encrypts an arbitrary payload under the thread key and sends it as |
| 620 | // a msg frame. Shared by chat replies (postReply) and the non-rendering |
| 621 | // connected-services replies (connect.result / connect.token). |
| 622 | func (r *Runtime) sendPayload(threadID string, key []byte, payload map[string]any) error { |
| 623 | blob, err := e2ee.EncryptJSON(key, payload) |
| 624 | if err != nil { |
| 625 | return err |
| 626 | } |
| 627 | return r.send(map[string]any{"type": "msg", "thread_id": threadID, "blob": blob}) |
| 628 | } |
| 629 | |
| 630 | // postText posts a chat message as this bot and returns the stored message id, |
| 631 | // so the caller can edit it in place afterwards — postReply is fire-and-forget |
| 632 | // over the socket and yields no id. It goes through the REST endpoint, which |
| 633 | // returns the stored message. The payload is a normal chat payload; set |
| 634 | // "action": true for a subtle /me-style line rather than a bubble. |
| 635 | func (r *Runtime) postText(threadID string, key []byte, payload map[string]any) (int64, error) { |
| 636 | blob, err := e2ee.EncryptJSON(key, payload) |
| 637 | if err != nil { |
| 638 | return 0, err |
| 639 | } |
| 640 | body, err := json.Marshal(map[string]string{"thread_id": threadID, "blob": blob}) |
| 641 | if err != nil { |
| 642 | return 0, err |
| 643 | } |
| 644 | var m struct { |
| 645 | ID int64 `json:"id"` |
| 646 | } |
| 647 | if err := r.postJSON("/api/messages?token="+url.QueryEscape(r.Session.Token), body, &m); err != nil { |
| 648 | return 0, err |
| 649 | } |
| 650 | return m.ID, nil |
| 651 | } |
| 652 | |
| 653 | // editText edits a message this bot posted, in place — the server verifies the |
| 654 | // author and the editWindow. The whole new payload rides as the blob (a reload |
| 655 | // folds the stored blob normally), so it must be complete. |
| 656 | func (r *Runtime) editText(threadID string, key []byte, id int64, payload map[string]any) error { |
| 657 | blob, err := e2ee.EncryptJSON(key, payload) |
| 658 | if err != nil { |
| 659 | return err |
| 660 | } |
| 661 | return r.send(map[string]any{"type": "edit", "thread_id": threadID, "message_id": id, "blob": blob}) |
| 662 | } |
| 663 | |
| 664 | // actionLine is a bot's handle to a single chat message it updates in place: |
| 665 | // posted once, then edited as work progresses — so any long or multi-step task |
| 666 | // (a build, a slow fetch, a background job) reports live status as ONE moving |
| 667 | // line instead of a stack of messages. It renders as a subtle /me-style action |
| 668 | // line (action:true), which the web keeps across edits (web/static/apps/chat.js), |
| 669 | // and reuses one timestamp so the line never jumps. Every method is best-effort: |
| 670 | // a dropped post or edit costs the user a little visibility, never correctness. |
| 671 | // |
| 672 | // Start one with (*Runtime).newActionLine or (*ThreadCtx).actionLine, then call |
| 673 | // set() to advance the text. Reusable by any bot flow — nothing here is |
| 674 | // build-specific. |
| 675 | type actionLine struct { |
| 676 | r *Runtime |
| 677 | thread string |
| 678 | key []byte |
| 679 | ts int64 |
| 680 | id int64 |
| 681 | posted bool |
| 682 | } |
| 683 | |
| 684 | // newActionLine posts the initial status line and returns a handle to update it. |
| 685 | func (r *Runtime) newActionLine(threadID string, key []byte, text string) *actionLine { |
| 686 | a := &actionLine{r: r, thread: threadID, key: key, ts: time.Now().UnixMilli()} |
| 687 | a.set(text) |
| 688 | return a |
| 689 | } |
| 690 | |
| 691 | func (a *actionLine) payload(text string) map[string]any { |
| 692 | // action:true → a centered /me-style line, not a chat bubble; the flag |
| 693 | // persists across edits, so the line stays subtle as its text changes. |
| 694 | return map[string]any{"name": a.r.Session.Name, "text": text, "ts": a.ts, "action": true} |
| 695 | } |
| 696 | |
| 697 | // set shows text: it posts the line the first time and edits it in place after. |
| 698 | // A failed first post is retried on the next call; a failed edit is dropped (the |
| 699 | // line simply doesn't advance). Not safe for concurrent use — drive one line |
| 700 | // from a single goroutine. |
| 701 | func (a *actionLine) set(text string) { |
| 702 | if !a.posted { |
| 703 | id, err := a.r.postText(a.thread, a.key, a.payload(text)) |
| 704 | if err != nil { |
| 705 | a.r.logf("action line post: %v", err) |
| 706 | return |
| 707 | } |
| 708 | a.id, a.posted = id, true |
| 709 | return |
| 710 | } |
| 711 | if err := a.r.editText(a.thread, a.key, a.id, a.payload(text)); err != nil { |
| 712 | a.r.logf("action line edit: %v", err) |
| 713 | } |
| 714 | } |
| 715 | |
| 716 | func (r *Runtime) send(v any) error { |
| 717 | r.wmu.Lock() |
| 718 | defer r.wmu.Unlock() |
| 719 | conn := r.conn |
| 720 | if conn == nil { |
| 721 | return fmt.Errorf("not connected") |
| 722 | } |
| 723 | data, err := json.Marshal(v) |
| 724 | if err != nil { |
| 725 | return err |
| 726 | } |
| 727 | ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) |
| 728 | defer cancel() |
| 729 | return conn.Write(ctx, websocket.MessageText, data) |
| 730 | } |
| 731 | |
| 732 | // loadAgents refreshes the never-reply-to set: a bot's mention trigger matches |
| 733 | // its bare name, and bot-authored text naturally names other bots — without |
| 734 | // this, two bots in one group ping-pong forever. |
| 735 | func (r *Runtime) loadAgents() { |
| 736 | var rl struct { |
| 737 | Roster []struct { |
| 738 | Label string `json:"label"` |
| 739 | IsAgent bool `json:"is_agent"` |
| 740 | } `json:"roster"` |
| 741 | } |
| 742 | if err := r.get("/api/roster", &rl); err != nil { |
| 743 | r.logf("loadAgents: %v", err) |
| 744 | return |
| 745 | } |
| 746 | agents := map[string]bool{} |
| 747 | for _, e := range rl.Roster { |
| 748 | if e.IsAgent { |
| 749 | agents[e.Label] = true |
| 750 | } |
| 751 | } |
| 752 | r.agents = agents |
| 753 | } |
| 754 | |
| 755 | // loadThreads caches the (unwrapped) key for any thread we don't hold yet. A |
| 756 | // newly-seen thread's cursor starts at its latest message, so we never reply |
| 757 | // to history — only to messages that arrive after we become aware of it. |
| 758 | func (r *Runtime) loadThreads() { |
| 759 | var tl struct { |
| 760 | Threads []struct { |
| 761 | ID string `json:"id"` |
| 762 | Kind string `json:"kind"` |
| 763 | Wrapped string `json:"wrapped"` |
| 764 | } `json:"threads"` |
| 765 | } |
| 766 | if err := r.get("/api/threads", &tl); err != nil { |
| 767 | r.logf("loadThreads: %v", err) |
| 768 | return |
| 769 | } |
| 770 | for _, t := range tl.Threads { |
| 771 | if t.Wrapped == "" { |
| 772 | continue |
| 773 | } |
| 774 | if _, ok := r.keys[t.ID]; ok { |
| 775 | continue |
| 776 | } |
| 777 | keyB64, err := e2ee.UnwrapThreadKey(t.Wrapped, r.Session.UserPriv) |
| 778 | if err != nil { |
| 779 | r.logf("load thread %s: %v", t.ID, err) |
| 780 | continue |
| 781 | } |
| 782 | key, err := e2ee.ImportKey(keyB64) |
| 783 | if err != nil { |
| 784 | r.logf("load thread %s: %v", t.ID, err) |
| 785 | continue |
| 786 | } |
| 787 | r.keys[t.ID] = threadEntry{key: key, kind: t.Kind} |
| 788 | var ml struct { |
| 789 | Messages []struct { |
| 790 | ID int64 `json:"id"` |
| 791 | } `json:"messages"` |
| 792 | } |
| 793 | if err := r.get("/api/messages?thread="+url.QueryEscape(t.ID)+"&limit=1", &ml); err == nil { |
| 794 | for _, m := range ml.Messages { |
| 795 | if m.ID > r.cursors[t.ID] { |
| 796 | r.cursors[t.ID] = m.ID |
| 797 | } |
| 798 | } |
| 799 | } |
| 800 | } |
| 801 | } |
| 802 | |
| 803 | // catchUp processes anything that arrived while disconnected, so a message |
| 804 | // sent during a reconnect blip still gets a reply. Cursors start at each |
| 805 | // thread's latest on first sight, so the initial connect backfills nothing. |
| 806 | func (r *Runtime) catchUp() { |
| 807 | for threadID := range r.keys { |
| 808 | var ml struct { |
| 809 | Messages []struct { |
| 810 | ID int64 `json:"id"` |
| 811 | Blob string `json:"blob"` |
| 812 | } `json:"messages"` |
| 813 | } |
| 814 | path := fmt.Sprintf("/api/messages?thread=%s&after=%d", url.QueryEscape(threadID), r.cursors[threadID]) |
| 815 | if err := r.get(path, &ml); err != nil { |
| 816 | r.logf("catchUp %s: %v", threadID, err) |
| 817 | continue |
| 818 | } |
| 819 | for _, m := range ml.Messages { |
| 820 | r.processMsg(threadID, m.ID, m.Blob) |
| 821 | } |
| 822 | } |
| 823 | } |
| 824 | |
| 825 | // declareKind self-reports what we are so the UI badges us honestly: "ai" |
| 826 | // when the model brain is available, else "bot" (deterministic). Re-declared |
| 827 | // on every connect, so toggling the key flips the badge with no admin action. |
| 828 | func (r *Runtime) declareKind() { |
| 829 | kind := "bot" |
| 830 | if r.ai != nil { |
| 831 | kind = "ai" |
| 832 | } |
| 833 | body, _ := json.Marshal(map[string]string{"token": r.Session.Token, "kind": kind}) |
| 834 | resp, err := r.client.Post(r.Base+"/api/bot-config", "application/json", strings.NewReader(string(body))) |
| 835 | if err != nil { |
| 836 | r.logf("declareKind: %v", err) |
| 837 | return |
| 838 | } |
| 839 | resp.Body.Close() |
| 840 | } |
| 841 | |
| 842 | // ---- member-endpoint helpers (schedules, threads) --------------------------- |
| 843 | // |
| 844 | // Everything below rides the same ordinary member endpoints a browser uses — |
| 845 | // the server has no bot-specific surface, and only ever sees ciphertext. |
| 846 | |
| 847 | // ScheduleEntry is one pending server-side schedule ({id, deliver_at, blob}). |
| 848 | type ScheduleEntry struct { |
| 849 | ID string `json:"id"` |
| 850 | DeliverAt int64 `json:"deliver_at"` |
| 851 | Blob string `json:"blob"` |
| 852 | } |
| 853 | |
| 854 | // schedule encrypts payload under the thread key NOW and asks the server to |
| 855 | // deliver it into the thread at deliverAt — the mini-app "future message" |
| 856 | // primitive (miniapps.go). The server learns only (thread, time). |
| 857 | func (r *Runtime) schedule(threadID string, key []byte, payload map[string]any, deliverAt time.Time) (string, error) { |
| 858 | blob, err := e2ee.EncryptJSON(key, payload) |
| 859 | if err != nil { |
| 860 | return "", err |
| 861 | } |
| 862 | var out struct { |
| 863 | ID string `json:"id"` |
| 864 | } |
| 865 | err = r.post("/api/schedule", map[string]any{ |
| 866 | "token": r.Session.Token, "thread_id": threadID, |
| 867 | "deliver_at": deliverAt.Unix(), "blob": blob, |
| 868 | }, &out) |
| 869 | return out.ID, err |
| 870 | } |
| 871 | |
| 872 | // unschedule cancels a pending schedule. Cancelling one that already fired |
| 873 | // succeeds quietly (races are by design, matching the server's semantics). |
| 874 | func (r *Runtime) unschedule(id string) error { |
| 875 | return r.del("/api/schedule/" + url.PathEscape(id) + "?token=" + url.QueryEscape(r.Session.Token)) |
| 876 | } |
| 877 | |
| 878 | // listSchedules returns a thread's pending schedules. |
| 879 | func (r *Runtime) listSchedules(threadID string) ([]ScheduleEntry, error) { |
| 880 | var out struct { |
| 881 | Schedules []ScheduleEntry `json:"schedules"` |
| 882 | } |
| 883 | err := r.get("/api/schedule?thread="+url.QueryEscape(threadID), &out) |
| 884 | return out.Schedules, err |
| 885 | } |
| 886 | |
| 887 | // ensureSelfThread finds — or creates — the bot's private workspace: a self |
| 888 | // thread only this bot is a member of. Charter alarms live here, so no human |
| 889 | // thread carries tick traffic and nobody else can post into it. Frame |
| 890 | // goroutine only (it touches r.keys); everyone else reads the selfThread |
| 891 | // snapshot it leaves behind. |
| 892 | func (r *Runtime) ensureSelfThread() (string, []byte, error) { |
| 893 | snapshot := func(id string, key []byte) (string, []byte, error) { |
| 894 | r.selfMu.Lock() |
| 895 | r.selfID, r.selfKey = id, key |
| 896 | r.selfMu.Unlock() |
| 897 | return id, key, nil |
| 898 | } |
| 899 | for id, e := range r.keys { |
| 900 | if e.kind == "self" { |
| 901 | return snapshot(id, e.key) |
| 902 | } |
| 903 | } |
| 904 | var me struct { |
| 905 | UserID string `json:"user_id"` |
| 906 | } |
| 907 | if err := r.get("/api/me", &me); err != nil { |
| 908 | return "", nil, err |
| 909 | } |
| 910 | key, err := e2ee.GenerateKey() |
| 911 | if err != nil { |
| 912 | return "", nil, err |
| 913 | } |
| 914 | wrapped, err := e2ee.WrapThreadKey(e2ee.B64uEncode(key), r.Session.UserPub) |
| 915 | if err != nil { |
| 916 | return "", nil, err |
| 917 | } |
| 918 | var out struct { |
| 919 | ThreadID string `json:"thread_id"` |
| 920 | } |
| 921 | err = r.post("/api/threads", map[string]any{ |
| 922 | "token": r.Session.Token, "kind": "self", |
| 923 | "members": []map[string]string{{"user_id": me.UserID, "wrapped": wrapped}}, |
| 924 | }, &out) |
| 925 | if err != nil { |
| 926 | return "", nil, err |
| 927 | } |
| 928 | r.keys[out.ThreadID] = threadEntry{key: key, kind: "self"} |
| 929 | return snapshot(out.ThreadID, key) |
| 930 | } |
| 931 | |
| 932 | // selfThread reads the self-thread snapshot ensureSelfThread left — safe from |
| 933 | // any goroutine; ok=false until the first charter-host connect completes. |
| 934 | func (r *Runtime) selfThread() (string, []byte, bool) { |
| 935 | r.selfMu.Lock() |
| 936 | defer r.selfMu.Unlock() |
| 937 | return r.selfID, r.selfKey, r.selfID != "" |
| 938 | } |
| 939 | |
| 940 | // rosterPubKey returns a member's public key (for wrapping a new thread key |
| 941 | // to them) and their display label. |
| 942 | func (r *Runtime) rosterPubKey(uid string) (pub, label string, err error) { |
| 943 | var rl struct { |
| 944 | Roster []struct { |
| 945 | UserID string `json:"user_id"` |
| 946 | Label string `json:"label"` |
| 947 | PublicKey string `json:"public_key"` |
| 948 | } `json:"roster"` |
| 949 | } |
| 950 | if err := r.get("/api/roster", &rl); err != nil { |
| 951 | return "", "", err |
| 952 | } |
| 953 | for _, e := range rl.Roster { |
| 954 | if e.UserID == uid { |
| 955 | if e.PublicKey == "" { |
| 956 | return "", "", fmt.Errorf("member %s has no public key", uid) |
| 957 | } |
| 958 | return e.PublicKey, e.Label, nil |
| 959 | } |
| 960 | } |
| 961 | return "", "", fmt.Errorf("member %s not in roster", uid) |
| 962 | } |
| 963 | |
| 964 | // createCharterThread creates a member bot's dedicated thread: a titled group |
| 965 | // of exactly (owner, host), its key wrapped to both, its name encrypted under |
| 966 | // the thread key (the same name_blob shape the web client writes). Returns |
| 967 | // the thread id + key. Deliberately does NOT touch r.keys — the frame |
| 968 | // goroutine learns the thread via loadThreads when its first frame arrives. |
| 969 | func (r *Runtime) createCharterThread(ownerUID, title string) (string, []byte, error) { |
| 970 | ownerPub, _, err := r.rosterPubKey(ownerUID) |
| 971 | if err != nil { |
| 972 | return "", nil, err |
| 973 | } |
| 974 | var me struct { |
| 975 | UserID string `json:"user_id"` |
| 976 | } |
| 977 | if err := r.get("/api/me", &me); err != nil { |
| 978 | return "", nil, err |
| 979 | } |
| 980 | key, err := e2ee.GenerateKey() |
| 981 | if err != nil { |
| 982 | return "", nil, err |
| 983 | } |
| 984 | keyB64 := e2ee.B64uEncode(key) |
| 985 | mine, err := e2ee.WrapThreadKey(keyB64, r.Session.UserPub) |
| 986 | if err != nil { |
| 987 | return "", nil, err |
| 988 | } |
| 989 | theirs, err := e2ee.WrapThreadKey(keyB64, ownerPub) |
| 990 | if err != nil { |
| 991 | return "", nil, err |
| 992 | } |
| 993 | nameBlob, err := e2ee.EncryptJSON(key, map[string]any{"name": title}) |
| 994 | if err != nil { |
| 995 | return "", nil, err |
| 996 | } |
| 997 | var out struct { |
| 998 | ThreadID string `json:"thread_id"` |
| 999 | } |
| 1000 | err = r.post("/api/threads", map[string]any{ |
| 1001 | "token": r.Session.Token, "kind": "group", "name_blob": nameBlob, |
| 1002 | "members": []map[string]string{ |
| 1003 | {"user_id": me.UserID, "wrapped": mine}, |
| 1004 | {"user_id": ownerUID, "wrapped": theirs}, |
| 1005 | }, |
| 1006 | }, &out) |
| 1007 | if err != nil { |
| 1008 | return "", nil, err |
| 1009 | } |
| 1010 | return out.ThreadID, key, nil |
| 1011 | } |
| 1012 | |
| 1013 | // renameCharterThread re-titles a member bot's dedicated thread — a renamed |
| 1014 | // bot renames its chat, since the thread was titled after the bot at birth |
| 1015 | // and members read the bot's identity off that name. Same name_blob shape the |
| 1016 | // web client's "Edit group" writes, encrypted under the thread key; the key |
| 1017 | // comes off the threads endpoint (wrappedKeyFor) so the builder goroutine |
| 1018 | // stays off r.keys. Best-effort: the charter (the truth) is already stored, |
| 1019 | // so a failed rename costs a stale label, never a wrong bot. |
| 1020 | func (r *Runtime) renameCharterThread(threadID, title string) { |
| 1021 | key, err := r.wrappedKeyFor(threadID) |
| 1022 | if err != nil { |
| 1023 | r.logf("bot rename %s: %v", threadID, err) |
| 1024 | return |
| 1025 | } |
| 1026 | nameBlob, err := e2ee.EncryptJSON(key, map[string]any{"name": title}) |
| 1027 | if err != nil { |
| 1028 | r.logf("bot rename %s: %v", threadID, err) |
| 1029 | return |
| 1030 | } |
| 1031 | if err := r.post("/api/threads/"+url.PathEscape(threadID)+"/name", map[string]any{ |
| 1032 | "token": r.Session.Token, "name_blob": nameBlob, |
| 1033 | }, nil); err != nil { |
| 1034 | r.logf("bot rename %s: %v", threadID, err) |
| 1035 | } |
| 1036 | } |
| 1037 | |
| 1038 | func (r *Runtime) post(path string, body any, out any) error { |
| 1039 | data, err := json.Marshal(body) |
| 1040 | if err != nil { |
| 1041 | return err |
| 1042 | } |
| 1043 | resp, err := r.client.Post(r.Base+path, "application/json", strings.NewReader(string(data))) |
| 1044 | if err != nil { |
| 1045 | return err |
| 1046 | } |
| 1047 | defer resp.Body.Close() |
| 1048 | if resp.StatusCode != http.StatusOK { |
| 1049 | return fmt.Errorf("%s: server %d", path, resp.StatusCode) |
| 1050 | } |
| 1051 | if out == nil { |
| 1052 | return nil |
| 1053 | } |
| 1054 | return json.NewDecoder(resp.Body).Decode(out) |
| 1055 | } |
| 1056 | |
| 1057 | // put sends a JSON PUT (the member-bots deploy endpoint). |
| 1058 | func (r *Runtime) put(path string, body any) error { |
| 1059 | data, err := json.Marshal(body) |
| 1060 | if err != nil { |
| 1061 | return err |
| 1062 | } |
| 1063 | req, err := http.NewRequest(http.MethodPut, r.Base+path, strings.NewReader(string(data))) |
| 1064 | if err != nil { |
| 1065 | return err |
| 1066 | } |
| 1067 | req.Header.Set("Content-Type", "application/json") |
| 1068 | resp, err := r.client.Do(req) |
| 1069 | if err != nil { |
| 1070 | return err |
| 1071 | } |
| 1072 | defer resp.Body.Close() |
| 1073 | if resp.StatusCode != http.StatusOK { |
| 1074 | return fmt.Errorf("%s: server %d", path, resp.StatusCode) |
| 1075 | } |
| 1076 | return nil |
| 1077 | } |
| 1078 | |
| 1079 | // del sends a DELETE (the path carries its own ?token=… auth). |
| 1080 | func (r *Runtime) del(path string) error { |
| 1081 | req, err := http.NewRequest(http.MethodDelete, r.Base+path, nil) |
| 1082 | if err != nil { |
| 1083 | return err |
| 1084 | } |
| 1085 | resp, err := r.client.Do(req) |
| 1086 | if err != nil { |
| 1087 | return err |
| 1088 | } |
| 1089 | resp.Body.Close() |
| 1090 | if resp.StatusCode != http.StatusOK { |
| 1091 | return fmt.Errorf("%s: server %d", path, resp.StatusCode) |
| 1092 | } |
| 1093 | return nil |
| 1094 | } |
| 1095 | |
| 1096 | func (r *Runtime) get(path string, out any) error { |
| 1097 | sep := "?" |
| 1098 | if strings.Contains(path, "?") { |
| 1099 | sep = "&" |
| 1100 | } |
| 1101 | resp, err := r.client.Get(r.Base + path + sep + "token=" + url.QueryEscape(r.Session.Token)) |
| 1102 | if err != nil { |
| 1103 | return err |
| 1104 | } |
| 1105 | defer resp.Body.Close() |
| 1106 | if resp.StatusCode != http.StatusOK { |
| 1107 | return fmt.Errorf("%s: server %d", path, resp.StatusCode) |
| 1108 | } |
| 1109 | return json.NewDecoder(resp.Body).Decode(out) |
| 1110 | } |
| 1111 | |
| 1112 | // postJSON POSTs a JSON body to an instance endpoint and decodes the response |
| 1113 | // into out (nil to ignore it). The path must carry its own auth (?token=…). |
| 1114 | func (r *Runtime) postJSON(path string, body []byte, out any) error { |
| 1115 | resp, err := r.client.Post(r.Base+path, "application/json", strings.NewReader(string(body))) |
| 1116 | if err != nil { |
| 1117 | return err |
| 1118 | } |
| 1119 | defer resp.Body.Close() |
| 1120 | if resp.StatusCode != http.StatusOK { |
| 1121 | return fmt.Errorf("%s: server %d", path, resp.StatusCode) |
| 1122 | } |
| 1123 | if out == nil { |
| 1124 | return nil |
| 1125 | } |
| 1126 | return json.NewDecoder(resp.Body).Decode(out) |
| 1127 | } |
| 1128 | |
| 1129 | // getText fetches a plain-text (ungated) endpoint — the authoring guide. |
| 1130 | func (r *Runtime) getText(path string) (string, error) { |
| 1131 | resp, err := r.client.Get(r.Base + path) |
| 1132 | if err != nil { |
| 1133 | return "", err |
| 1134 | } |
| 1135 | defer resp.Body.Close() |
| 1136 | if resp.StatusCode != http.StatusOK { |
| 1137 | return "", fmt.Errorf("%s: server %d", path, resp.StatusCode) |
| 1138 | } |
| 1139 | b, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) |
| 1140 | if err != nil { |
| 1141 | return "", err |
| 1142 | } |
| 1143 | return string(b), nil |
| 1144 | } |
| 1145 | |
| 1146 | func (r *Runtime) logf(format string, args ...any) { |
| 1147 | if r.Logf != nil { |
| 1148 | r.Logf("["+r.Session.Name+"] "+format, args...) |
| 1149 | } |
| 1150 | } |
| 1151 | |
amadan