1package agentrunner23import (4 "context"5 "encoding/json"6 "errors"7 "fmt"8 "slices"9 "strconv"10 "strings"11 "sync"12 "time"1314 "github.com/FelineStateMachine/tinyrelay/internal/event"15 "github.com/google/uuid"16)1718type ServiceOptions struct {19 RelayURL, Secret, StatePath string20 Rooms []string21 Process ProcessOptions22 Timeout time.Duration23 Log func(error)24}25type service struct {26 options ServiceOptions27 relay RelayClient28 queue *Runner29 ctx context.Context30 mu sync.Mutex31 decisions map[string]chan event.Event32}3334func Serve(ctx context.Context, opts ServiceOptions) error {35 if opts.Timeout <= 0 {36 opts.Timeout = 30 * time.Minute37 }38 pubkey, err := event.PublicKey(opts.Secret)39 if err != nil {40 return err41 }42 s := &service{options: opts, ctx: ctx, decisions: map[string]chan event.Event{}, relay: RelayClient{URL: opts.RelayURL, Secret: opts.Secret, PubKey: pubkey, Rooms: opts.Rooms, OnError: opts.Log}}43 queue, err := New(Options{AllowedRooms: opts.Rooms, StatePath: opts.StatePath, Handle: s.handle, OnError: opts.Log})44 if err != nil {45 return err46 }47 s.queue = queue48 s.relay.OnEvent = s.receive49 // Start queues only after a connection is established, so journal recovery50 // can publish status before it invokes the agent process.51 s.relay.OnReady = func() { queue.Start(ctx) }52 defer queue.Stop()53 err = s.relay.Run(ctx)54 if errors.Is(err, context.Canceled) {55 return nil56 }57 return err58}59func (s *service) receive(e event.Event) {60 s.mu.Lock()61 for _, id := range event.TagValues(e, "e") {62 if wait := s.decisions[id]; wait != nil {63 select {64 case wait <- e:65 default:66 }67 }68 }69 s.mu.Unlock()70 m, ok := s.relay.Mention(e)71 if !ok {72 return73 }74 if isCancelCommand(m.Prompt) {75 s.queue.CancelFrom(m.Room, m.Author)76 return77 }78 if err := s.queue.Enqueue(m); err != nil && !errors.Is(err, ErrDuplicate) && s.options.Log != nil {79 s.options.Log(err)80 }81}82func (s *service) handle(parent context.Context, batch []Mention) error {83 ctx, cancel := context.WithTimeout(parent, s.options.Timeout)84 defer cancel()85 first := batch[0]86 root, err := s.relay.ResolveRoot(ctx, first.Room, first.Root)87 if err != nil {88 return err89 }90 first.Root = root91 // Keep the canonical root on the batch so permission requests emitted92 // while this task runs stay attached to the same top-level conversation.93 batch[0].Root = root94 tags := [][]string{{"h", first.Room}, {"e", first.Root}, {"p", s.relay.PubKey}, {"output", "text/plain"}, {"subject", shortText(first.Prompt, 100)}, {"expiration", strconv.FormatInt(time.Now().Add(s.options.Timeout).Unix(), 10)}}95 var prompt strings.Builder96 prompt.WriteString("These are messages addressed to you in a Nostr chat. Reply to the users' request. Room: " + first.Room + "\n\n")97 for _, m := range batch {98 tags = append(tags, []string{"i", m.EventID, "event"})99 fmt.Fprintf(&prompt, "Author: %s\nMessage: %s\n%s\n\n", m.Author, m.EventID, m.Prompt)100 }101 request, err := s.relay.Publish(ctx, event.Event{Kind: 5000, CreatedAt: first.CreatedAt, Tags: tags, Content: shortText(first.Prompt, 1000)})102 if err != nil {103 return err104 }105 feedback := func(status, info string) error {106 _, err := s.relay.Publish(ctx, event.Event{Kind: 7000, Tags: [][]string{{"h", first.Room}, {"e", request.ID}, {"p", request.PubKey}, {"status", status, shortText(info, 500)}}, Content: shortText(info, 4000)})107 return err108 }109 if err := feedback("processing", "Working on your request"); err != nil {110 return err111 }112 process := s.options.Process113 var last time.Time114 process.Progress = func(info string) {115 if time.Since(last) < 2*time.Second {116 return117 }118 last = time.Now()119 if err := feedback("processing", info); err != nil {120 cancel()121 }122 }123 process.Permission = func(ctx context.Context, p Permission) (string, error) { return s.permission(ctx, batch, p) }124 heartbeatCtx, stopHeartbeat := context.WithCancel(ctx)125 heartbeatDone := make(chan struct{})126 go func() { defer close(heartbeatDone); s.heartbeat(heartbeatCtx, first.Room) }()127 output, runErr := RunProcess(ctx, process, prompt.String())128 stopHeartbeat()129 <-heartbeatDone130 finishCtx, finish := context.WithTimeout(s.ctx, 10*time.Second)131 defer finish()132 finalStatus := "idle"133 if runErr != nil {134 finalStatus = "offline"135 _ = s.publishPresence(finishCtx, first.Room, finalStatus)136 message := "Agent stopped: " + runErr.Error()137 if errors.Is(runErr, context.Canceled) {138 message = "Canceled"139 }140 _, publishErr := s.relay.Publish(finishCtx, event.Event{Kind: 7000, Tags: [][]string{{"h", first.Room}, {"e", request.ID}, {"p", request.PubKey}, {"status", "error"}}, Content: shortText(message, 1000)})141 if publishErr != nil {142 return errors.Join(runErr, publishErr)143 }144 return runErr145 }146 _ = s.publishPresence(finishCtx, first.Room, finalStatus)147 if output == "" {148 output = "Task completed."149 }150 raw, _ := json.Marshal(request)151 _, err = s.relay.Publish(ctx, event.Event{Kind: 6000, Tags: [][]string{{"h", first.Room}, {"e", request.ID}, {"p", request.PubKey}, {"request", string(raw)}}, Content: output})152 if err != nil {153 return err154 }155 replyTags := [][]string{{"h", first.Room}, {"e", first.Root, "", "root"}}156 for _, m := range batch {157 if !slices.ContainsFunc(replyTags, func(tag []string) bool { return len(tag) > 1 && tag[0] == "p" && tag[1] == m.Author }) {158 replyTags = append(replyTags, []string{"p", m.Author})159 }160 }161 _, err = s.relay.Publish(ctx, event.Event{Kind: 9, Tags: replyTags, Content: output})162 return err163}164165func isCancelCommand(prompt string) bool {166 fields := strings.Fields(prompt)167 return len(fields) > 0 && (fields[0] == "/cancel" || fields[0] == "/stop")168}169170func (s *service) publishPresence(ctx context.Context, room, status string) error {171 _, err := s.relay.Publish(ctx, event.Event{Kind: 20001, Tags: [][]string{{"h", room}}, Content: status})172 return err173}174func (s *service) heartbeat(ctx context.Context, room string) {175 timer := time.NewTicker(30 * time.Second)176 defer timer.Stop()177 for {178 _, err := s.relay.Publish(ctx, event.Event{Kind: 20001, Tags: [][]string{{"h", room}}, Content: "working"})179 if err != nil {180 return181 }182 select {183 case <-ctx.Done():184 return185 case <-timer.C:186 }187 }188}189func (s *service) permission(ctx context.Context, batch []Mention, p Permission) (string, error) {190 allow, deny := "", ""191 for _, option := range p.Options {192 if option.Kind == "allow_once" {193 allow = option.ID194 }195 if option.Kind == "reject_once" {196 deny = option.ID197 }198 }199 if allow == "" {200 return deny, nil201 }202 first := batch[0]203 expires := time.Now().Add(10 * time.Minute).Unix()204 if deadline, ok := ctx.Deadline(); ok && deadline.Unix() < expires {205 expires = deadline.Unix()206 }207 request := event.Event{Kind: 9, CreatedAt: time.Now().Unix(), Tags: [][]string{{"h", first.Room}, {"e", first.Root, "", "root"}, {"request", "approve"}, {"d", uuid.NewString()}, {"expiration", strconv.FormatInt(expires, 10)}, {"subject", shortText(p.Title, 150)}, {"p", first.Author}}, Content: "Allow this tool action once?\n\n" + shortText(p.Title, 4000) + "\n\n" + p.Details}208 if err := event.Sign(&request, s.options.Secret); err != nil {209 return "", err210 }211 wait := make(chan event.Event, 1)212 s.mu.Lock()213 s.decisions[request.ID] = wait214 s.mu.Unlock()215 defer func() { s.mu.Lock(); delete(s.decisions, request.ID); s.mu.Unlock() }()216 if _, err := s.relay.Publish(ctx, request); err != nil {217 return "", err218 }219 for {220 select {221 case <-ctx.Done():222 return "", ctx.Err()223 case answer := <-wait:224 if answer.PubKey != first.Author || answer.Kind != 7 {225 continue226 }227 switch strings.TrimSpace(answer.Content) {228 case "+", "":229 return allow, nil230 case "-":231 return deny, nil232 }233 }234 }235}236func shortText(value string, size int) string {237 runes := []rune(value)238 if len(runes) > size {239 return string(runes[:size]) + "…"240 }241 return value242}