amadan
server: lchat bot announce — the Announcer in Go, Node-compatible state (#264)
Third slice of the Go bot sidecar (docs/bots.md): the first complete bot. internal/bot.Announcer ports agent/announcer.mjs — write-only (unwraps thread keys only to encrypt its own posts, reads no messages, calls no model API), polls /api/changelog, baselines silently on first run, batches fresh entries into one encrypted message per group/DM, self-declares kind "bot". lchat bot announce runs one tick (cron-friendly) or loops with -poll; it auto-provisions from LCHAT_ANNOUNCER_INVITE on first boot. Env names, the session file, and the .state.json shape all match the Node announcer, so a systemd unit swap inherits identity and memory in place — no replays. End-to-end test drives the real router: two provisioned identities, a group created via /api/threads with client-wrapped keys, baseline tick posts nothing, doctored state posts exactly one batched message that decrypts under the thread key with the entries verbatim, kind lands as bot, idle tick posts nothing more.
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
4 files changed,
+433
−3
bot_announcer_test.go+158 −0botcmd.go+56 −1docs/bots.md+9 −2internal/bot/announcer.go+210 −0
diff --git a/bot_announcer_test.go b/bot_announcer_test.go| new file mode 100644 |
| index 0000000..c1f34ff |
| --- /dev/null |
| +++ b/bot_announcer_test.go |
| @@ -0,0 +1,158 @@ |
| +package main |
| + |
| +import ( |
| + "bytes" |
| + "encoding/json" |
| + "net/http" |
| + "os" |
| + "testing" |
| + |
| + "lchat/internal/bot" |
| + "lchat/internal/e2ee" |
| +) |
| + |
| +// provisionTestBot mints an agent invite via the real admin API and claims it |
| +// with the real provisioning client. |
| +func provisionTestBot(t *testing.T, tsURL, name string) *bot.Session { |
| + t.Helper() |
| + secret := mintAgentInviteForTest(t, tsURL) |
| + sess, err := bot.Provision(tsURL, secret, name) |
| + if err != nil { |
| + t.Fatalf("provision %s: %v", name, err) |
| + } |
| + return sess |
| +} |
| + |
| +// createGroupWith makes owner create a group thread with the announcer, both |
| +// keys wrapped client-side — exactly what a member's browser does when adding |
| +// a bot to a group. Returns the thread id and the raw thread key. |
| +func createGroupWith(t *testing.T, tsURL string, s *Server, owner, announcer *bot.Session) (string, []byte) { |
| + t.Helper() |
| + threadKey, err := e2ee.GenerateKey() |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + ownerID, _ := s.db.deviceTokenUser(owner.Token) |
| + annID, _ := s.db.deviceTokenUser(announcer.Token) |
| + members := []map[string]string{} |
| + for id, sess := range map[string]*bot.Session{ownerID: owner, annID: announcer} { |
| + wrapped, err := e2ee.WrapThreadKey(e2ee.B64uEncode(threadKey), sess.UserPub) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + members = append(members, map[string]string{"user_id": id, "wrapped": wrapped}) |
| + } |
| + body, _ := json.Marshal(map[string]any{"token": owner.Token, "kind": "group", "members": members}) |
| + resp, err := http.Post(tsURL+"/api/threads", "application/json", bytes.NewReader(body)) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer resp.Body.Close() |
| + var out struct { |
| + ThreadID string `json:"thread_id"` |
| + } |
| + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil || out.ThreadID == "" { |
| + t.Fatalf("create group: status %d err %v id %q", resp.StatusCode, err, out.ThreadID) |
| + } |
| + return out.ThreadID, threadKey |
| +} |
| + |
| +func TestAnnouncerAgainstRealServer(t *testing.T) { |
| + ts, s := newTestServer(t) |
| + if len(s.changelog) < 1 { |
| + t.Fatal("embedded changelog is empty — the announcer has nothing to test with") |
| + } |
| + |
| + owner := provisionTestBot(t, ts.URL, "Owner") |
| + annSess := provisionTestBot(t, ts.URL, "Announcer") |
| + threadID, threadKey := createGroupWith(t, ts.URL, s, owner, annSess) |
| + |
| + statePath := t.TempDir() + "/announcer-session.json.state.json" |
| + a := &bot.Announcer{Base: ts.URL, Session: annSess, StatePath: statePath, Logf: t.Logf} |
| + |
| + // First tick baselines silently: state written, nothing posted. |
| + if err := a.Tick(); err != nil { |
| + t.Fatalf("baseline tick: %v", err) |
| + } |
| + msgs, err := s.db.messagesAfter(threadID, 0, 100) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(msgs) != 0 { |
| + t.Fatalf("baseline tick posted %d messages, want 0", len(msgs)) |
| + } |
| + |
| + // Forget the two newest entries (Node state shape) → next tick announces |
| + // them, batched into one message. |
| + raw, err := os.ReadFile(statePath) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + var state struct { |
| + Announced []string `json:"announced"` |
| + } |
| + if err := json.Unmarshal(raw, &state); err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(state.Announced) != len(s.changelog) { |
| + t.Fatalf("baselined %d ids, want %d", len(state.Announced), len(s.changelog)) |
| + } |
| + forget := 1 |
| + if len(state.Announced) >= 2 { |
| + forget = 2 |
| + } |
| + state.Announced = state.Announced[:len(state.Announced)-forget] |
| + raw, _ = json.Marshal(state) |
| + if err := os.WriteFile(statePath, raw, 0o600); err != nil { |
| + t.Fatal(err) |
| + } |
| + |
| + if err := a.Tick(); err != nil { |
| + t.Fatalf("announce tick: %v", err) |
| + } |
| + msgs, err = s.db.messagesAfter(threadID, 0, 100) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(msgs) != 1 { |
| + t.Fatalf("announce tick posted %d messages, want 1 (batched)", len(msgs)) |
| + } |
| + |
| + // The blob must decrypt under the thread key to the announcer's payload, |
| + // carrying the forgotten entries' verbatim text. |
| + var payload struct { |
| + Name string `json:"name"` |
| + Text string `json:"text"` |
| + TS int64 `json:"ts"` |
| + } |
| + if err := e2ee.DecryptJSON(threadKey, msgs[0].Blob, &payload); err != nil { |
| + t.Fatalf("decrypt announcement: %v", err) |
| + } |
| + if payload.Name != "Announcer" || payload.TS == 0 { |
| + t.Errorf("bad payload envelope: %+v", payload) |
| + } |
| + if !bytes.Contains([]byte(payload.Text), []byte("🚀 Just shipped:")) { |
| + t.Errorf("announcement missing header: %q", payload.Text) |
| + } |
| + for _, e := range s.changelog[len(s.changelog)-forget:] { |
| + if !bytes.Contains([]byte(payload.Text), []byte(e.Text)) { |
| + t.Errorf("announcement missing entry %s", e.ID) |
| + } |
| + } |
| + |
| + // Kind self-declared as deterministic bot. |
| + uid, _ := s.db.deviceTokenUser(annSess.Token) |
| + var kind string |
| + if err := s.db.sql.QueryRow("SELECT COALESCE(kind, '') FROM bot_configs WHERE user_id = ?", uid).Scan(&kind); err != nil || kind != "bot" { |
| + t.Errorf("announcer kind: got %q (err %v), want \"bot\"", kind, err) |
| + } |
| + |
| + // A third tick has nothing new: no further messages. |
| + if err := a.Tick(); err != nil { |
| + t.Fatalf("idle tick: %v", err) |
| + } |
| + msgs, _ = s.db.messagesAfter(threadID, 0, 100) |
| + if len(msgs) != 1 { |
| + t.Errorf("idle tick posted more messages: %d total", len(msgs)) |
| + } |
| +} |
diff --git a/botcmd.go b/botcmd.go| index c3857bf..a8982ba 100644 |
| --- a/botcmd.go |
| +++ b/botcmd.go |
| @@ -8,6 +8,7 @@ package main |
| import ( |
| "flag" |
| "fmt" |
| + "log" |
| "os" |
| "strings" |
| @@ -19,17 +20,71 @@ func runBotCmd(args []string) { |
| fmt.Fprintln(os.Stderr, `usage: lchat bot <command> |
| commands: |
| - provision claim an agent invite and write the bot's session file`) |
| + provision claim an agent invite and write the bot's session file |
| + announce post new changelog entries to the bot's threads (the Announcer)`) |
| os.Exit(2) |
| } |
| switch args[0] { |
| case "provision": |
| runBotProvision(args[1:]) |
| + case "announce": |
| + runBotAnnounce(args[1:]) |
| default: |
| die("bot: unknown command %q", args[0]) |
| } |
| } |
| +// runBotAnnounce is the Announcer (agent/announcer.mjs in Go): one tick by |
| +// default (cron-friendly), a poll loop with -poll. Env names match the Node |
| +// sidecar's so a systemd unit can swap implementations without edits; the |
| +// session, state file, and invite flow are all Node-compatible too, so the Go |
| +// announcer takes over an existing one's identity and memory in place. |
| +func runBotAnnounce(args []string) { |
| + fs := flag.NewFlagSet("bot announce", flag.ExitOnError) |
| + base := fs.String("base", envOr("LCHAT_BASE", ""), "instance URL (env: LCHAT_BASE)") |
| + session := fs.String("session", envOr("LCHAT_ANNOUNCER_SESSION", "announcer-session.json"), "session file (env: LCHAT_ANNOUNCER_SESSION)") |
| + name := fs.String("name", envOr("LCHAT_ANNOUNCER_NAME", "Announcer"), "bot name (env: LCHAT_ANNOUNCER_NAME)") |
| + poll := fs.Duration("poll", 0, "poll interval, e.g. 5m; 0 runs exactly one tick and exits") |
| + _ = fs.Parse(reorderFlagsFirst(fs, args)) |
| + |
| + if *base == "" { |
| + die("bot announce: set -base (or LCHAT_BASE) to your instance URL") |
| + } |
| + sess, err := bot.LoadSession(*session) |
| + if err != nil { |
| + die("bot announce: %v", err) |
| + } |
| + if sess == nil { |
| + // First boot: self-provision from the announcer's invite, like the Node |
| + // sidecar. Env-only for the same secrecy reason as `bot provision`. |
| + fragment := strings.TrimSpace(os.Getenv("LCHAT_ANNOUNCER_INVITE")) |
| + if fragment == "" { |
| + die("bot announce: no session at %s and no LCHAT_ANNOUNCER_INVITE set", *session) |
| + } |
| + if sess, err = bot.Provision(*base, fragment, *name); err != nil { |
| + die("bot announce: %v", err) |
| + } |
| + if err := sess.Save(*session); err != nil { |
| + die("bot announce: %v", err) |
| + } |
| + log.Printf("announcer: provisioned %q (session %s)", *name, *session) |
| + } |
| + |
| + a := &bot.Announcer{ |
| + Base: strings.TrimRight(*base, "/"), |
| + Session: sess, |
| + StatePath: bot.StatePathFor(*session), |
| + Logf: log.Printf, |
| + } |
| + if *poll <= 0 { |
| + if err := a.Tick(); err != nil { |
| + die("bot announce: %v", err) |
| + } |
| + return |
| + } |
| + a.Run(*poll) |
| +} |
| + |
| func runBotProvision(args []string) { |
| fs := flag.NewFlagSet("bot provision", flag.ExitOnError) |
| base := fs.String("base", envOr("LCHAT_BASE", ""), "instance URL, e.g. https://test.luan.chat (env: LCHAT_BASE)") |
diff --git a/docs/bots.md b/docs/bots.md| index e093a04..1e015c5 100644 |
| --- a/docs/bots.md |
| +++ b/docs/bots.md |
| @@ -110,8 +110,15 @@ them after any deliberate `crypto.js` change and re-run `go test |
| (`internal/bot` + `botcmd.go`) claims an agent invite headlessly — the invite |
| secret comes from `LCHAT_INVITE` in the env, never argv — and writes a session |
| file byte-compatible with the Node sidecar's `session.json`, so either |
| -implementation can take over the other's identity in place. Next slice: the |
| -Announcer. |
| +implementation can take over the other's identity in place. **And the |
| +Announcer**: `lchat bot announce` (one tick, cron-friendly; `-poll 5m` to |
| +loop) is the full Go port of `announcer.mjs` — auto-provisions from |
| +`LCHAT_ANNOUNCER_INVITE`, baselines silently on first run, batches new |
| +changelog entries into one encrypted post per thread, self-declares kind |
| +`bot`. Env names and the `.state.json` shape match the Node announcer, so a |
| +unit swap inherits identity *and* memory (no replayed entries). Remaining for |
| +full Node parity: the chat runtime (WS loop + Claude API) and Eleven's |
| +welcome/persona layer. |
| ## What exists today (`agent/`) |
diff --git a/internal/bot/announcer.go b/internal/bot/announcer.go| new file mode 100644 |
| index 0000000..00e4efb |
| --- /dev/null |
| +++ b/internal/bot/announcer.go |
| @@ -0,0 +1,210 @@ |
| +package bot |
| + |
| +import ( |
| + "encoding/json" |
| + "fmt" |
| + "net/http" |
| + "net/url" |
| + "os" |
| + "strings" |
| + "time" |
| + |
| + "lchat/internal/e2ee" |
| +) |
| + |
| +// Announcer is the Go port of agent/announcer.mjs: the write-only deploy bot. |
| +// It polls the instance's embedded changelog (/api/changelog) and posts entries |
| +// it hasn't announced yet into every group/DM it's been added to — so a deploy |
| +// announces itself with no operator machine in the loop. |
| +// |
| +// Deliberately write-only: it never fetches or decrypts a single message — it |
| +// unwraps thread keys only to *encrypt* what it posts, and it calls no model |
| +// API. "The Announcer is in this group" means "it can post here", not "it |
| +// reads this room". |
| +// |
| +// State (which entries have been announced) lives at <session>.state.json in |
| +// the same shape the Node announcer writes ({"announced":[ids]}), so a Go |
| +// announcer takes over a Node one's memory in place — no replayed entries. |
| +type Announcer struct { |
| + Base string |
| + Session *Session |
| + // StatePath defaults to <session path>.state.json via StatePathFor. |
| + StatePath string |
| + // Logf receives progress lines ("announced 2 entries to 3/3 threads"). |
| + Logf func(format string, args ...any) |
| + |
| + client *http.Client |
| + kindDeclared bool |
| +} |
| + |
| +// StatePathFor mirrors the Node announcer's convention. |
| +func StatePathFor(sessionPath string) string { return sessionPath + ".state.json" } |
| + |
| +type announcerState struct { |
| + Announced []string `json:"announced"` |
| +} |
| + |
| +// Tick runs one poll: declare our kind (once, best-effort), fetch the |
| +// changelog, and post anything new to every thread we're in. On the very first |
| +// run it baselines to the current changelog without posting, so a fresh |
| +// announcer announces what ships *after* it, not history. |
| +func (a *Announcer) Tick() error { |
| + if a.client == nil { |
| + a.client = &http.Client{Timeout: 30 * time.Second} |
| + } |
| + a.declareKind() |
| + |
| + var cl struct { |
| + Entries []struct { |
| + ID string `json:"id"` |
| + Text string `json:"text"` |
| + } `json:"entries"` |
| + } |
| + if err := a.get("/api/changelog", &cl); err != nil { |
| + return fmt.Errorf("changelog: %w", err) |
| + } |
| + |
| + raw, err := os.ReadFile(a.StatePath) |
| + if os.IsNotExist(err) { |
| + ids := make([]string, 0, len(cl.Entries)) |
| + for _, e := range cl.Entries { |
| + ids = append(ids, e.ID) |
| + } |
| + a.logf("baselined at %d entries", len(ids)) |
| + return a.saveState(announcerState{Announced: ids}) |
| + } |
| + if err != nil { |
| + return err |
| + } |
| + var state announcerState |
| + if err := json.Unmarshal(raw, &state); err != nil { |
| + return fmt.Errorf("bad state file %s: %w", a.StatePath, err) |
| + } |
| + seen := make(map[string]bool, len(state.Announced)) |
| + for _, id := range state.Announced { |
| + seen[id] = true |
| + } |
| + var fresh []string // entry texts, oldest first (changelog order) |
| + for _, e := range cl.Entries { |
| + if e.ID != "" && e.Text != "" && !seen[e.ID] { |
| + fresh = append(fresh, e.Text) |
| + state.Announced = append(state.Announced, e.ID) |
| + } |
| + } |
| + if len(fresh) == 0 { |
| + return nil |
| + } |
| + |
| + text := "🚀 Just shipped:\n\n" + strings.Join(fresh, "\n\n") |
| + var tl struct { |
| + Threads []struct { |
| + ID string `json:"id"` |
| + Kind string `json:"kind"` |
| + Wrapped string `json:"wrapped"` |
| + } `json:"threads"` |
| + } |
| + if err := a.get("/api/threads", &tl); err != nil { |
| + return fmt.Errorf("threads: %w", err) |
| + } |
| + posted, targets := 0, 0 |
| + for _, t := range tl.Threads { |
| + if t.Wrapped == "" || (t.Kind != "group" && t.Kind != "dm") { |
| + continue |
| + } |
| + targets++ |
| + if err := a.post(t.ID, t.Wrapped, text); err != nil { |
| + a.logf("thread %s: %v", t.ID, err) |
| + continue |
| + } |
| + posted++ |
| + } |
| + // Mark announced even with zero targets: entries shipped while the bot sat |
| + // in no groups are old news by the time someone adds it — don't replay them. |
| + if err := a.saveState(state); err != nil { |
| + return err |
| + } |
| + a.logf("announced %d entries to %d/%d threads", len(fresh), posted, targets) |
| + return nil |
| +} |
| + |
| +// Run is the poll loop. Tick errors are logged and retried next tick. |
| +func (a *Announcer) Run(interval time.Duration) { |
| + for { |
| + if err := a.Tick(); err != nil { |
| + a.logf("tick failed: %v", err) |
| + } |
| + time.Sleep(interval) |
| + } |
| +} |
| + |
| +// post encrypts one message under the thread's key and posts it — the only |
| +// thing the announcer ever does with a thread key. |
| +func (a *Announcer) post(threadID, wrapped, text string) error { |
| + keyB64, err := e2ee.UnwrapThreadKey(wrapped, a.Session.UserPriv) |
| + if err != nil { |
| + return err |
| + } |
| + key, err := e2ee.ImportKey(keyB64) |
| + if err != nil { |
| + return err |
| + } |
| + blob, err := e2ee.EncryptJSON(key, map[string]any{ |
| + "name": a.Session.Name, "text": text, "ts": time.Now().UnixMilli(), |
| + }) |
| + if err != nil { |
| + return err |
| + } |
| + body, _ := json.Marshal(map[string]string{"thread_id": threadID, "blob": blob}) |
| + resp, err := a.client.Post(a.Base+"/api/messages?token="+url.QueryEscape(a.Session.Token), |
| + "application/json", strings.NewReader(string(body))) |
| + if err != nil { |
| + return err |
| + } |
| + defer resp.Body.Close() |
| + if resp.StatusCode != http.StatusOK { |
| + return fmt.Errorf("server %d", resp.StatusCode) |
| + } |
| + return nil |
| +} |
| + |
| +// declareKind self-reports "bot" (deterministic — no model) so the UI badges |
| +// us BOT, not AI. Best-effort; retried on a later tick if it fails. |
| +func (a *Announcer) declareKind() { |
| + if a.kindDeclared { |
| + return |
| + } |
| + body, _ := json.Marshal(map[string]string{"token": a.Session.Token, "kind": "bot"}) |
| + resp, err := a.client.Post(a.Base+"/api/bot-config", "application/json", strings.NewReader(string(body))) |
| + if err != nil { |
| + a.logf("declare kind: %v", err) |
| + return |
| + } |
| + resp.Body.Close() |
| + a.kindDeclared = resp.StatusCode == http.StatusOK |
| +} |
| + |
| +func (a *Announcer) get(path string, out any) error { |
| + resp, err := a.client.Get(a.Base + path + "?token=" + url.QueryEscape(a.Session.Token)) |
| + if err != nil { |
| + return err |
| + } |
| + defer resp.Body.Close() |
| + if resp.StatusCode != http.StatusOK { |
| + return fmt.Errorf("%s: server %d", path, resp.StatusCode) |
| + } |
| + return json.NewDecoder(resp.Body).Decode(out) |
| +} |
| + |
| +func (a *Announcer) saveState(s announcerState) error { |
| + raw, err := json.Marshal(s) |
| + if err != nil { |
| + return err |
| + } |
| + return os.WriteFile(a.StatePath, raw, 0o600) |
| +} |
| + |
| +func (a *Announcer) logf(format string, args ...any) { |
| + if a.Logf != nil { |
| + a.Logf("announcer: "+format, args...) |
| + } |
| +} |