| 1 | package bot |
| 2 | |
| 3 | // The charter scheduler: how member bots act on their own clock, with no new |
| 4 | // server surface. The server's existing scheduled-message queue is the alarm: |
| 5 | // the charter host encrypts a bot.tick under its own SELF-THREAD key and asks |
| 6 | // the server to deliver it at the due moment (miniapps.go). Delivery arrives |
| 7 | // as an ordinary msg frame, handleTyped routes it here, the charter's task |
| 8 | // runs, and the next occurrence is re-armed. Because the queue is server-side, |
| 9 | // alarms survive sidecar restarts for free — every connect (and a slow |
| 10 | // background tick) reconciles charters against pending alarms. The server |
| 11 | // learns only (thread, time), which it already learns for every reminder. |
| 12 | // |
| 13 | // Hibernation (charter-wake registration): after every schedule recompute the |
| 14 | // host also reports the EARLIEST next run to the instance (reportCharterDue → |
| 15 | // POST /api/charter-next-due), which the router's due-time index folds in — |
| 16 | // so a hibernating instance is woken for charter work even when the queue's |
| 17 | // alarm row was consumed while no host was listening. And because a due-wake's |
| 18 | // co-spawned host can lose the delivery race (the instance's scheduler fires |
| 19 | // the tick before this host's cursor baseline), the connect-time reconcile |
| 20 | // treats a live charter with no pending alarm as owing a run NOW — armed at |
| 21 | // now, delivered within a scheduler tick, run through the normal path. |
| 22 | |
| 23 | import ( |
| 24 | "time" |
| 25 | |
| 26 | "lchat/internal/e2ee" |
| 27 | ) |
| 28 | |
| 29 | // tickPayload is the decrypted shape of one alarm. Bot is the charter id it |
| 30 | // fires (the store id, not the slug). |
| 31 | type tickPayload struct { |
| 32 | Type string `json:"type"` // "bot.tick" |
| 33 | Bot string `json:"bot"` |
| 34 | TS int64 `json:"ts"` |
| 35 | } |
| 36 | |
| 37 | // charterReconcile makes reality match the store: every live (un-paused) |
| 38 | // charter has exactly one pending alarm in the self-thread; alarms for |
| 39 | // deleted, paused, or duplicated charters are cancelled. atConnect runs on |
| 40 | // the frame goroutine and may create the self-thread; the background ticker |
| 41 | // passes false and only ever reads the snapshot. |
| 42 | func (r *Runtime) charterReconcile(atConnect bool) { |
| 43 | var selfID string |
| 44 | var selfKey []byte |
| 45 | if atConnect { |
| 46 | id, key, err := r.ensureSelfThread() |
| 47 | if err != nil { |
| 48 | r.logf("charters: self thread: %v", err) |
| 49 | return |
| 50 | } |
| 51 | selfID, selfKey = id, key |
| 52 | } else { |
| 53 | id, key, ok := r.selfThread() |
| 54 | if !ok { |
| 55 | return // never connected as host yet |
| 56 | } |
| 57 | selfID, selfKey = id, key |
| 58 | } |
| 59 | instKey, err := r.instanceKey() |
| 60 | if err != nil { |
| 61 | // Normal before any member has granted the instance key — nothing to |
| 62 | // run yet anyway (charters are sealed under it). Nudge online holders |
| 63 | // (an open tab grants on the spot), so the host heals without anyone |
| 64 | // reloading. |
| 65 | r.logf("charters: instance key not held yet (%v)", err) |
| 66 | r.nudgeInstanceKey() |
| 67 | return |
| 68 | } |
| 69 | entries, err := r.listCharters(instKey) |
| 70 | if err != nil { |
| 71 | r.logf("charters: list: %v", err) |
| 72 | return |
| 73 | } |
| 74 | live := map[string]*Charter{} |
| 75 | for _, e := range entries { |
| 76 | if !e.Charter.Paused { |
| 77 | live[e.ID] = e.Charter |
| 78 | } |
| 79 | } |
| 80 | pending, err := r.listSchedules(selfID) |
| 81 | if err != nil { |
| 82 | r.logf("charters: pending: %v", err) |
| 83 | return |
| 84 | } |
| 85 | armed := map[string]bool{} |
| 86 | for _, p := range pending { |
| 87 | var tp tickPayload |
| 88 | if e2ee.DecryptJSON(selfKey, p.Blob, &tp) != nil || tp.Type != "bot.tick" { |
| 89 | continue |
| 90 | } |
| 91 | if live[tp.Bot] == nil || armed[tp.Bot] { |
| 92 | // An alarm for a bot that's gone/paused, or a duplicate (a crash |
| 93 | // re-armed twice): cancel it. |
| 94 | _ = r.unschedule(p.ID) |
| 95 | continue |
| 96 | } |
| 97 | armed[tp.Bot] = true |
| 98 | } |
| 99 | now := time.Now() |
| 100 | for id, c := range live { |
| 101 | if armed[id] { |
| 102 | continue |
| 103 | } |
| 104 | when := c.Schedule.next(now) |
| 105 | if atConnect { |
| 106 | // A live charter with NO pending alarm at connect means its alarm |
| 107 | // fired while no host was listening (delivered into the self-thread |
| 108 | // after the last host stopped — the hibernation due-wake case: the |
| 109 | // instance's scheduler can beat this host's connect by a beat and |
| 110 | // the cursor baseline swallows the tick) — or the alarm was never |
| 111 | // armed at all. Either way the schedule owes a run NOW, not at the |
| 112 | // next occurrence (a daily bot would otherwise skip a whole day |
| 113 | // every time its wake lost that race). Arm at now: the server |
| 114 | // delivers within one scheduler tick and the run flows through the |
| 115 | // ordinary charterTick path — same crash-recovery ordering, no |
| 116 | // second execution path. (charterTick's duplicate guard covers the |
| 117 | // sliver where the original tick was in flight after all.) |
| 118 | when = now |
| 119 | } |
| 120 | r.armCharter(selfID, selfKey, id, when) |
| 121 | } |
| 122 | // Report the charter due-time so the schedule survives hibernation: the |
| 123 | // instance stores the earliest next run (a bare timestamp — the charters |
| 124 | // themselves stay ciphertext to it) and the router's due-time index wakes |
| 125 | // the sleeping instance for it. Reported on every reconcile — connect and |
| 126 | // the slow tick — so a value from a dead host heals as soon as any host |
| 127 | // connects. Zero (no live charters) is reported too: it clears a stale |
| 128 | // value that would otherwise buy one pointless wake. |
| 129 | r.reportCharterDue(earliestCharterDue(live, now)) |
| 130 | } |
| 131 | |
| 132 | // earliestCharterDue is the soonest next occurrence across live charters |
| 133 | // (unix seconds; 0 = none). Pure — the clock is a parameter, like |
| 134 | // ScheduleSpec.next. It recomputes next(now) rather than reading the armed |
| 135 | // alarms: an armed alarm may be sooner (e.g. an interval schedule armed a |
| 136 | // while ago), but the alarm is ALSO a server-side scheduled message the |
| 137 | // instance already folds into its next-due answer, so the earlier truth |
| 138 | // still wins — this report is the backstop for exactly the alarms the queue |
| 139 | // has already consumed. |
| 140 | func earliestCharterDue(live map[string]*Charter, now time.Time) int64 { |
| 141 | var due int64 |
| 142 | for _, c := range live { |
| 143 | if n := c.Schedule.next(now).Unix(); due == 0 || n < due { |
| 144 | due = n |
| 145 | } |
| 146 | } |
| 147 | return due |
| 148 | } |
| 149 | |
| 150 | // reportCharterDue posts the earliest charter run to the instance |
| 151 | // (POST /api/charter-next-due, memberbots.go). Best-effort — a failed report |
| 152 | // costs promptness on the next hibernated occurrence (the instance wakes for |
| 153 | // a member request instead), never correctness — and it rides r.client's |
| 154 | // marked transport like every sidecar request, so it stamps no activity. |
| 155 | func (r *Runtime) reportCharterDue(due int64) { |
| 156 | if err := r.post("/api/charter-next-due", map[string]any{ |
| 157 | "token": r.Session.Token, "due": due, |
| 158 | }, nil); err != nil { |
| 159 | r.logf("charters: report next due: %v", err) |
| 160 | } |
| 161 | } |
| 162 | |
| 163 | // recomputeCharterDue re-derives and reports the earliest charter run — for |
| 164 | // call sites that just mutated the store (build, update, remove) rather than |
| 165 | // folded it (botbuild.go). |
| 166 | func (r *Runtime) recomputeCharterDue(instKey []byte) { |
| 167 | entries, err := r.listCharters(instKey) |
| 168 | if err != nil { |
| 169 | r.logf("charters: recompute next due: %v", err) |
| 170 | return |
| 171 | } |
| 172 | live := map[string]*Charter{} |
| 173 | for _, e := range entries { |
| 174 | if !e.Charter.Paused { |
| 175 | live[e.ID] = e.Charter |
| 176 | } |
| 177 | } |
| 178 | r.reportCharterDue(earliestCharterDue(live, time.Now())) |
| 179 | } |
| 180 | |
| 181 | // armCharter queues one alarm for a charter. |
| 182 | func (r *Runtime) armCharter(selfID string, selfKey []byte, charterID string, when time.Time) { |
| 183 | payload := map[string]any{"type": "bot.tick", "bot": charterID, "ts": time.Now().UnixMilli()} |
| 184 | if _, err := r.schedule(selfID, selfKey, payload, when); err != nil { |
| 185 | r.logf("charters: arm %s: %v", charterID, err) |
| 186 | } |
| 187 | } |
| 188 | |
| 189 | // charterTick fires one delivered alarm (frame goroutine, via handleTyped). |
| 190 | // The charter is re-read from the store first — it may have been edited, |
| 191 | // paused, or deleted since the alarm was set — and the NEXT occurrence is |
| 192 | // armed before the run, so a task that crashes the process can't silence the |
| 193 | // bot forever (a duplicate alarm from that ordering is dropped by reconcile). |
| 194 | func (r *Runtime) charterTick(selfID string, selfKey []byte, tp *tickPayload) { |
| 195 | // Duplicate-tick guard: the connect-time catch-up arm (charterReconcile) |
| 196 | // re-arms a charter whose alarm looks consumed-unheard, but a delivery can |
| 197 | // land in the sliver between the connect's cursor baseline and the |
| 198 | // reconcile's pending-alarm read — then BOTH the original tick and the |
| 199 | // catch-up alarm arrive. Two ticks for one charter closer together than |
| 200 | // any expressible schedule are that race, not a run owed twice: drop the |
| 201 | // second without re-arming (the first already armed the next occurrence). |
| 202 | if last, ok := r.charterRuns[tp.Bot]; ok && time.Since(last) < minEvery()/2 { |
| 203 | r.logf("charters: tick %s: duplicate within %s — dropped", tp.Bot, minEvery()/2) |
| 204 | return |
| 205 | } |
| 206 | instKey, err := r.instanceKey() |
| 207 | if err != nil { |
| 208 | r.logf("charters: tick %s: instance key: %v", tp.Bot, err) |
| 209 | return |
| 210 | } |
| 211 | // One store read serves both needs: the ticked charter, and the live set |
| 212 | // for the wake-time report below. |
| 213 | entries, err := r.listCharters(instKey) |
| 214 | if err != nil { |
| 215 | r.logf("charters: tick %s: %v", tp.Bot, err) |
| 216 | return |
| 217 | } |
| 218 | var c *Charter |
| 219 | live := map[string]*Charter{} |
| 220 | for _, e := range entries { |
| 221 | if !e.Charter.Paused { |
| 222 | live[e.ID] = e.Charter |
| 223 | } |
| 224 | if e.ID == tp.Bot { |
| 225 | c = e.Charter |
| 226 | } |
| 227 | } |
| 228 | if c == nil || c.Paused { |
| 229 | // Deleted or paused since the alarm was set — don't re-arm, but do |
| 230 | // refresh the report: this alarm may have been the earliest. |
| 231 | r.reportCharterDue(earliestCharterDue(live, time.Now())) |
| 232 | return |
| 233 | } |
| 234 | r.armCharter(selfID, selfKey, tp.Bot, c.Schedule.next(time.Now())) |
| 235 | // Re-report after each run's re-arm, so a hibernating instance's stored |
| 236 | // wake-time always tracks the freshest schedule. |
| 237 | r.reportCharterDue(earliestCharterDue(live, time.Now())) |
| 238 | r.charterRuns[tp.Bot] = time.Now() |
| 239 | |
| 240 | key, ok := r.threadKeyFor(c.Thread) |
| 241 | if !ok { |
| 242 | r.logf("charters: %s: no key for thread %s — skipping run", c.Name, c.Thread) |
| 243 | return |
| 244 | } |
| 245 | r.logf("charters: running %s (%s)", c.Name, c.Schedule.Summary()) |
| 246 | res, err := r.execTask(c.Title, &c.Task) |
| 247 | switch { |
| 248 | case err != nil: |
| 249 | r.logf("charters: %s: %v", c.Name, err) |
| 250 | r.charterRunFailed(selfID, selfKey, tp.Bot, c, key, taskResult{}) |
| 251 | case res.sourcesDown(): |
| 252 | // Before the empty check: an all-sources-failed run that also came |
| 253 | // back textless should log (and, on the first failure, explain) the |
| 254 | // real problem. |
| 255 | r.logf("charters: %s: reached none of its sources (%d fetches)", c.Name, res.fetches) |
| 256 | r.charterRunFailed(selfID, selfKey, tp.Bot, c, key, res) |
| 257 | case res.text == "": |
| 258 | r.logf("charters: %s: empty result — nothing posted", c.Name) |
| 259 | r.charterRunFailed(selfID, selfKey, tp.Bot, c, key, taskResult{}) |
| 260 | default: |
| 261 | r.clearCharterFails(tp.Bot) |
| 262 | r.postTaskResult(c.Thread, key, c.Title, &c.Task, res.text, res.card) |
| 263 | r.meterInc("charter_runs", 1) // completed runs only — failures cost tokens but deliver nothing |
| 264 | } |
| 265 | } |
| 266 | |
| 267 | // How a failing charter behaves (the RTE-down case: a source 403s for a day, |
| 268 | // an hourly bot): ONE notice per outage, not one per attempt, and the retries |
| 269 | // space themselves out instead of hammering a dead feed on schedule. |
| 270 | const charterBackoffCap = 24 * time.Hour |
| 271 | |
| 272 | // The one notice a failing bot posts, appended to the model's own explanation |
| 273 | // of what went wrong. Two variants: interval schedules also stretch their |
| 274 | // retries (backoffDelay), daily ones keep their day. |
| 275 | const ( |
| 276 | failNoteStretch = "I'll keep trying quietly — a little less often while my sources are unreachable — and post again as soon as I can read them." |
| 277 | failNotePlain = "I'll keep trying quietly on my schedule and post again as soon as I can read my sources." |
| 278 | ) |
| 279 | |
| 280 | // backoffDelay is how long after `fails` consecutive failed runs the next |
| 281 | // attempt should wait: interval schedules double per failure (2×, 4×, 8×… |
| 282 | // the interval) capped at a day; daily (at) schedules return 0 — their |
| 283 | // cadence is already gentle, and the one-notice rule alone stops the spam. |
| 284 | // Pure; the schedule must have been validated (compiled `every`). |
| 285 | func backoffDelay(s *ScheduleSpec, fails int) time.Duration { |
| 286 | if s.Every == "" || fails < 1 { |
| 287 | return 0 |
| 288 | } |
| 289 | d := s.every |
| 290 | for i := 0; i < fails && d < charterBackoffCap; i++ { |
| 291 | d *= 2 |
| 292 | } |
| 293 | return min(d, charterBackoffCap) |
| 294 | } |
| 295 | |
| 296 | // charterRunFailed folds one failed run into the backoff state: bump the |
| 297 | // consecutive count, push the pre-armed next alarm out by the backoff, and |
| 298 | // post the model's explanation — but only on the FIRST failure of an outage |
| 299 | // (err/empty runs have no text and post nothing, exactly as before). |
| 300 | func (r *Runtime) charterRunFailed(selfID string, selfKey []byte, botID string, c *Charter, key []byte, res taskResult) { |
| 301 | n := r.bumpCharterFails(botID) |
| 302 | if d := backoffDelay(&c.Schedule, n); d > 0 { |
| 303 | // charterTick armed the next ordinary occurrence before the run (the |
| 304 | // crash-safety ordering); replace it — a source that's down now is |
| 305 | // usually still down one interval later. |
| 306 | r.dropCharterAlarms(selfID, selfKey, botID) |
| 307 | r.armCharter(selfID, selfKey, botID, time.Now().Add(d)) |
| 308 | r.logf("charters: %s: failure #%d — next attempt in %s", c.Name, n, d) |
| 309 | } else { |
| 310 | r.logf("charters: %s: failure #%d", c.Name, n) |
| 311 | } |
| 312 | if n > 1 || res.text == "" { |
| 313 | return |
| 314 | } |
| 315 | note := failNotePlain |
| 316 | if c.Schedule.Every != "" { |
| 317 | note = failNoteStretch |
| 318 | } |
| 319 | // The explanation posts without any card the run half-composed — a failure |
| 320 | // notice should read as one, not dress up as an edition. |
| 321 | r.postTaskResult(c.Thread, key, c.Title, &c.Task, res.text+"\n\n"+note, nil) |
| 322 | } |
| 323 | |
| 324 | // bumpCharterFails / clearCharterFails track consecutive failed runs per |
| 325 | // charter — in memory only (a restart forgives; worst case one extra notice). |
| 326 | // Own lock: ticks run on the frame goroutine, but builds and removals clear |
| 327 | // from their own goroutines. |
| 328 | func (r *Runtime) bumpCharterFails(botID string) int { |
| 329 | r.failMu.Lock() |
| 330 | defer r.failMu.Unlock() |
| 331 | r.charterFails[botID]++ |
| 332 | return r.charterFails[botID] |
| 333 | } |
| 334 | |
| 335 | func (r *Runtime) clearCharterFails(botID string) { |
| 336 | r.failMu.Lock() |
| 337 | defer r.failMu.Unlock() |
| 338 | delete(r.charterFails, botID) |
| 339 | } |
| 340 | |
| 341 | // dropCharterAlarms cancels every pending alarm for one charter in the |
| 342 | // self-thread — schedule edits, retirement, and backoff re-arms all need it. |
| 343 | func (r *Runtime) dropCharterAlarms(selfID string, selfKey []byte, botID string) { |
| 344 | pending, err := r.listSchedules(selfID) |
| 345 | if err != nil { |
| 346 | return |
| 347 | } |
| 348 | for _, p := range pending { |
| 349 | var tp tickPayload |
| 350 | if e2ee.DecryptJSON(selfKey, p.Blob, &tp) == nil && tp.Type == "bot.tick" && tp.Bot == botID { |
| 351 | _ = r.unschedule(p.ID) |
| 352 | } |
| 353 | } |
| 354 | } |
| 355 | |
| 356 | // threadKeyFor returns the unwrapped key for a thread the bot holds. Frame |
| 357 | // goroutine only (r.keys). |
| 358 | func (r *Runtime) threadKeyFor(threadID string) ([]byte, bool) { |
| 359 | e, ok := r.keys[threadID] |
| 360 | if !ok { |
| 361 | // A thread created since the last loadThreads (e.g. by our own builder |
| 362 | // goroutine): one refresh is cheap and self-heals. |
| 363 | r.loadThreads() |
| 364 | e, ok = r.keys[threadID] |
| 365 | } |
| 366 | if !ok { |
| 367 | return nil, false |
| 368 | } |
| 369 | return e.key, true |
| 370 | } |
| 371 | |
amadan