From 1fc902e8f68eccca65047150d32ae0dc1dc0b35e Mon Sep 17 00:00:00 2001 From: "bot.oppi.li" Date: Wed, 14 Oct 2026 12:06:46 +0100 Subject: [PATCH] twin: tail jetstream and send mentions to tarp twin reads the sh.tangled.* records of some dids from jetstream. When a record mentions the bot, twin writes a prompt and runs a lua chunk in tarp. The chunk finds the terminal of the thread and writes the prompt to it. If there is no terminal, the chunk starts the agent in a new one. The settings come from TWIN_* variables. The cursor is kept, so a restart continues from it. Signed-off-by: oppiliappan --- internal/config/config.go | 74 +++++++++++ internal/twin/dispatch.lua | 17 +++ internal/twin/prompt.tmpl | 33 +++++ internal/twin/twin.go | 243 +++++++++++++++++++++++++++++++++++++ main.go | 115 ++++++++++++++++++ 5 files changed, 482 insertions(+) create mode 100644 internal/config/config.go create mode 100644 internal/twin/dispatch.lua create mode 100644 internal/twin/prompt.tmpl create mode 100644 internal/twin/twin.go create mode 100644 main.go diff --git a/internal/config/config.go b/internal/config/config.go new file mode 100644 index 0000000..7a154d4 --- /dev/null +++ b/internal/config/config.go @@ -0,0 +1,74 @@ +// Package config reads the TWIN_* environment variables. +package config + +import ( + "errors" + "os" + "path/filepath" + "slices" + "strings" + "time" + + "github.com/kelseyhightower/envconfig" +) + +type Config struct { + JetstreamHosts []string `envconfig:"JETSTREAM_HOSTS" default:"jetstream.us-east.bsky.network,jetstream.us-west.bsky.network"` + DIDs []string `envconfig:"DIDS" required:"true"` + Collections []string `envconfig:"COLLECTIONS" default:"sh.tangled.*"` + Operations []string `envconfig:"OPERATIONS" default:"create"` + + MentionDIDs []string `envconfig:"MENTION_DIDS"` + MentionHandles []string `envconfig:"MENTION_HANDLES"` + + TarpAddr string `envconfig:"TARP_ADDR" default:"127.0.0.1:39423"` + TarpTimeout time.Duration `envconfig:"TARP_TIMEOUT" default:"10s"` + + AgentCommand string `envconfig:"AGENT_COMMAND" default:"exec opencode --auto --prompt \"$TWIN_PROMPT\""` + Join bool `envconfig:"JOIN" default:"true"` + PromptTemplate string `envconfig:"PROMPT_TEMPLATE"` + + StateDir string `envconfig:"STATE_DIR"` + WorkDir string `envconfig:"WORK_DIR"` + + PLCURL string `envconfig:"PLC_URL" default:"https://plc.directory"` + DryRun bool `envconfig:"DRY_RUN"` + LogLevel string `envconfig:"LOG_LEVEL" default:"info"` +} + +func Load() (*Config, error) { + var c Config + if err := envconfig.Process("twin", &c); err != nil { + return nil, err + } + // An empty variable gives one empty item. Remove these items. + for _, list := range []*[]string{&c.DIDs, &c.Collections, &c.Operations, &c.MentionDIDs, &c.MentionHandles, &c.JetstreamHosts} { + *list = slices.DeleteFunc(*list, func(s string) bool { return strings.TrimSpace(s) == "" }) + } + if len(c.DIDs) == 0 { + return nil, errors.New("TWIN_DIDS is empty") + } + if len(c.MentionDIDs) == 0 && len(c.MentionHandles) == 0 { + return nil, errors.New("set TWIN_MENTION_DIDS or TWIN_MENTION_HANDLES") + } + if c.StateDir == "" { + base := os.Getenv("XDG_STATE_HOME") + if base == "" { + home, err := os.UserHomeDir() + if err != nil { + return nil, err + } + base = filepath.Join(home, ".local", "state") + } + c.StateDir = filepath.Join(base, "twin") + } + if c.WorkDir == "" { + c.WorkDir = filepath.Join(c.StateDir, "work") + } + return &c, nil +} + +func Usage() error { + var c Config + return envconfig.Usage("twin", &c) +} diff --git a/internal/twin/dispatch.lua b/internal/twin/dispatch.lua new file mode 100644 index 0000000..26f23a2 --- /dev/null +++ b/internal/twin/dispatch.lua @@ -0,0 +1,17 @@ +-- Find the terminal of the thread and write to it. +-- If there is no terminal, start one. +local marker, spawn, input, join = ... + +if join then + for _, terminal in ipairs(tarp.terminal.list()) do + for _, arg in ipairs(terminal.command or {}) do + if arg == marker then + tarp.terminal.write({ terminal = terminal.id, data = input }) + return { terminal = terminal.id, joined = true } + end + end + end +end + +local terminal = tarp.terminal.spawn(spawn) +return { terminal = terminal.id, joined = false } diff --git a/internal/twin/prompt.tmpl b/internal/twin/prompt.tmpl new file mode 100644 index 0000000..32ead1f --- /dev/null +++ b/internal/twin/prompt.tmpl @@ -0,0 +1,33 @@ +{{- if .Joined -}} +{{.AuthorHandle}} wrote again on {{.Kind}} {{.Post.Thread}}: +{{- else -}} +You are {{.Bot}}, the twin of {{.AuthorHandle}} on tangled (https://tangled.org). +{{.AuthorHandle}} mentioned you on a {{.Kind}} and you must act on it. + +Thread: {{.Post.Thread}} ({{.Kind}}) +{{- with .Thread}}{{if .Title}} +Title: {{.Title}}{{end}}{{end}} +{{- with .Repo}} +Repository: {{.}}{{end}} +Mention: {{.Post.URI}} (cid {{.Post.CID}}) +{{- with .Post.ReplyTo}} +In reply to: {{.}}{{end}} +Working directory: {{.WorkDir}} +{{- with .Thread}}{{if and .Text (ne .URI $.Post.URI)}} + +The {{$.Kind}} says: + +{{.Text}}{{end}}{{end}} + +{{.AuthorHandle}} wrote: +{{- end}} + +{{if .Post.Title}}{{.Post.Title}} + +{{end}}{{.Post.Text}} +{{- if not .Joined}} + +Do what {{.AuthorHandle}} asks. Read the thread on tangled for more context. +When you are done, answer on the thread as {{.Bot}}: comment on it, or push a +branch and open a pull, with the bot credentials in this environment. +{{- end}} diff --git a/internal/twin/twin.go b/internal/twin/twin.go new file mode 100644 index 0000000..da4bf04 --- /dev/null +++ b/internal/twin/twin.go @@ -0,0 +1,243 @@ +// Package twin sends the records for the bot to an agent in tarp. +package twin + +import ( + "bytes" + "context" + _ "embed" + "encoding/json" + "fmt" + "log/slog" + "os" + "path/filepath" + "regexp" + "slices" + "strconv" + "strings" + "text/template" + + "github.com/bluesky-social/jetstream" + + "tangled.org/oppi.li/twin/internal/config" + "tangled.org/oppi.li/twin/internal/tangled" + "tangled.org/oppi.li/twin/internal/tarp" +) + +//go:embed dispatch.lua +var dispatchLua string + +//go:embed prompt.tmpl +var defaultPrompt string + +type Twin struct { + cfg *config.Config + log *slog.Logger + resolver *tangled.Resolver + prompt *template.Template + cursor string +} + +func New(cfg *config.Config, log *slog.Logger, resolver *tangled.Resolver) (*Twin, error) { + text := defaultPrompt + if cfg.PromptTemplate != "" { + raw, err := os.ReadFile(cfg.PromptTemplate) + if err != nil { + return nil, err + } + text = string(raw) + } + prompt, err := template.New("prompt").Parse(text) + if err != nil { + return nil, fmt.Errorf("parsing the prompt template: %w", err) + } + for _, dir := range []string{cfg.StateDir, cfg.WorkDir, filepath.Join(cfg.StateDir, "prompts")} { + if err := os.MkdirAll(dir, 0o700); err != nil { + return nil, err + } + } + return &Twin{ + cfg: cfg, + log: log, + resolver: resolver, + prompt: prompt, + cursor: filepath.Join(cfg.StateDir, "cursor"), + }, nil +} + +func (t *Twin) Cursor() uint64 { + raw, err := os.ReadFile(t.cursor) + if err != nil { + return 0 + } + cursor, err := strconv.ParseUint(strings.TrimSpace(string(raw)), 10, 64) + if err != nil { + return 0 + } + return cursor +} + +func (t *Twin) SaveCursor(cursor uint64) { + if cursor == 0 { + return + } + tmp := t.cursor + ".tmp" + if err := os.WriteFile(tmp, []byte(strconv.FormatUint(cursor, 10)), 0o600); err != nil { + t.log.Warn("saving the cursor", "error", err) + return + } + if err := os.Rename(tmp, t.cursor); err != nil { + t.log.Warn("saving the cursor", "error", err) + } +} + +func (t *Twin) HandleEvent(ctx context.Context, event *jetstream.Event) { + commit := event.Commit + if event.Kind != jetstream.KindCommit || commit == nil { + return + } + t.log.Debug("commit", "did", event.DID, "collection", commit.Collection, "rkey", commit.Rkey, "operation", commit.Operation) + if !slices.Contains(t.cfg.DIDs, event.DID) || !slices.Contains(t.cfg.Operations, string(commit.Operation)) { + return + } + raw, err := json.Marshal(commit.Record) + if err != nil { + return + } + post, err := tangled.Parse(event.DID, commit.Collection, commit.Rkey, string(commit.Operation), commit.CID, raw) + if err != nil { + t.log.Warn("skipping a record", "error", err) + return + } + if post == nil || !post.MentionsAny(t.cfg.MentionDIDs, t.cfg.MentionHandles) { + return + } + t.log.Info("mentioned", "uri", post.URI, "thread", post.Thread) + if err := t.Dispatch(ctx, post); err != nil { + t.log.Error("dispatching the agent", "uri", post.URI, "error", err) + } +} + +type Task struct { + Post *tangled.Post + Thread *tangled.Post + Kind string + Repo string + AuthorHandle string + Bot string + WorkDir string + Joined bool +} + +type dispatched struct { + Terminal string + Joined bool +} + +func (t *Twin) Dispatch(ctx context.Context, post *tangled.Post) error { + log := t.log.With("uri", post.URI, "thread", post.Thread) + task := Task{ + Post: post, + Kind: tangled.Kind(post.Thread), + Repo: post.Repo, + AuthorHandle: t.resolver.Handle(ctx, post.Author), + Bot: t.bot(), + WorkDir: filepath.Join(t.cfg.WorkDir, threadDir(post.Thread)), + } + if post.Thread == post.URI { + task.Thread = post + } else if thread, err := t.resolver.Record(ctx, post.Thread); err != nil { + log.Warn("fetching the thread", "error", err) + } else if thread != nil { + task.Thread = thread + if task.Repo == "" { + task.Repo = thread.Repo + } + } + if err := os.MkdirAll(task.WorkDir, 0o755); err != nil { + return err + } + + fresh, err := t.render(task) + if err != nil { + return err + } + task.Joined = true + followUp, err := t.render(task) + if err != nil { + return err + } + + rkey := post.URI[strings.LastIndex(post.URI, "/")+1:] + promptFile := filepath.Join(t.cfg.StateDir, "prompts", threadDir(post.Thread)+"-"+rkey+".md") + if err := os.WriteFile(promptFile, []byte(fresh), 0o600); err != nil { + return err + } + + // The marker in the command identifies the terminal of the thread. + marker := "twin:" + post.Thread + spawn := map[string]any{ + "command": []string{"sh", "-c", t.cfg.AgentCommand, marker}, + "cwd": task.WorkDir, + "env": map[string]string{ + "TWIN_PROMPT": fresh, + "TWIN_PROMPT_FILE": promptFile, + "TWIN_THREAD": post.Thread, + "TWIN_URI": post.URI, + "TWIN_CID": post.CID, + "TWIN_AUTHOR": post.Author, + "TWIN_REPO": task.Repo, + "TWIN_KIND": task.Kind, + }, + } + // Send the text as one paste, then push enter. + input := "\x1b[200~" + followUp + "\x1b[201~\r" + + if t.cfg.DryRun { + log.Info("dry run: would dispatch", "prompt", fresh) + return nil + } + var result dispatched + err = tarp.WithTimeout(ctx, t.cfg.TarpAddr, t.cfg.TarpTimeout, func(c *tarp.Client) error { + value, err := c.Eval("twin-dispatch", dispatchLua, marker, spawn, input, t.cfg.Join) + if err != nil { + return err + } + fields, ok := value.(map[string]any) + if !ok { + return fmt.Errorf("dispatch returned %T", value) + } + result.Terminal, _ = fields["terminal"].(string) + result.Joined, _ = fields["joined"].(bool) + return nil + }) + if err != nil { + return err + } + log.Info("dispatched", "terminal", result.Terminal, "joined", result.Joined) + return nil +} + +func (t *Twin) render(task Task) (string, error) { + var out bytes.Buffer + if err := t.prompt.Execute(&out, task); err != nil { + return "", fmt.Errorf("rendering the prompt: %w", err) + } + return strings.TrimSpace(out.String()) + "\n", nil +} + +func (t *Twin) bot() string { + if len(t.cfg.MentionHandles) > 0 { + return strings.TrimPrefix(t.cfg.MentionHandles[0], "@") + } + return t.cfg.MentionDIDs[0] +} + +var unsafePath = regexp.MustCompile(`[^A-Za-z0-9._-]+`) + +func threadDir(thread string) string { + parts := strings.Split(strings.TrimPrefix(thread, "at://"), "/") + if len(parts) == 3 { + parts = []string{parts[0], parts[2]} + } + return unsafePath.ReplaceAllString(strings.Join(parts, "-"), "-") +} diff --git a/main.go b/main.go new file mode 100644 index 0000000..20ddb36 --- /dev/null +++ b/main.go @@ -0,0 +1,115 @@ +// twin starts an agent in tarp when a tangled record talks to the bot. +package main + +import ( + "context" + "errors" + "fmt" + "log/slog" + "os" + "os/signal" + "syscall" + + "github.com/bluesky-social/indigo/atproto/identity" + "github.com/bluesky-social/jetstream" + + "tangled.org/oppi.li/twin/internal/config" + "tangled.org/oppi.li/twin/internal/tangled" + "tangled.org/oppi.li/twin/internal/twin" +) + +const usage = `usage: + twin tail jetstream and dispatch mentions + twin dispatch URI dispatch one record by its at-uri, mention or not + twin config list the TWIN_* settings +` + +func main() { + if err := run(); err != nil { + fmt.Fprintln(os.Stderr, "twin:", err) + os.Exit(1) + } +} + +func run() error { + args := os.Args[1:] + if len(args) > 0 && args[0] == "config" { + return config.Usage() + } + if len(args) > 0 && args[0] != "dispatch" || len(args) > 0 && len(args) != 2 { + fmt.Fprint(os.Stderr, usage) + os.Exit(2) + } + + cfg, err := config.Load() + if err != nil { + return err + } + var level slog.Level + if err := level.UnmarshalText([]byte(cfg.LogLevel)); err != nil { + return fmt.Errorf("TWIN_LOG_LEVEL: %w", err) + } + log := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: level})) + + identity.DefaultPLCURL = cfg.PLCURL + resolver := tangled.NewResolver(identity.DefaultDirectory()) + t, err := twin.New(cfg, log, resolver) + if err != nil { + return err + } + + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + + if len(args) == 2 { + post, err := resolver.Record(ctx, args[1]) + if err != nil { + return err + } + if post == nil { + return fmt.Errorf("%s has no text", args[1]) + } + return t.Dispatch(ctx, post) + } + return tail(ctx, cfg, log, t) +} + +func tail(ctx context.Context, cfg *config.Config, log *slog.Logger, t *twin.Twin) error { + if len(cfg.JetstreamHosts) == 0 { + return errors.New("TWIN_JETSTREAM_HOSTS is empty") + } + opts := []jetstream.Option{ + jetstream.WithDIDs(cfg.DIDs), + jetstream.WithCollections(cfg.Collections), + jetstream.WithKinds([]jetstream.Kind{jetstream.KindCommit}), + // Time cursors work on all hosts. + jetstream.WithCursorMode(jetstream.CursorTime), + jetstream.WithFailoverHosts(cfg.JetstreamHosts[1:]...), + jetstream.WithLogger(log), + } + cursor := t.Cursor() + if cursor != 0 { + opts = append(opts, jetstream.WithLiveCursor(cursor)) + } + stream, err := jetstream.Subscribe(cfg.JetstreamHosts[0], opts...) + if err != nil { + return err + } + defer stream.Close() + + log.Info("tailing jetstream", "host", cfg.JetstreamHosts[0], "dids", cfg.DIDs, "cursor", cursor) + for batch, err := range stream.Events(ctx) { + if err != nil { + if errors.Is(err, jetstream.ErrFatal) { + return err + } + log.Warn("jetstream", "error", err) + continue + } + for _, event := range batch.Events() { + t.HandleEvent(ctx, &event) + } + t.SaveCursor(batch.LastCursor()) + } + return nil +} -- knotmirror 0.0.1