git.giacomolli.uk - onechat/blob - daemon.go

   1 package main
   2
   3 import (
   4         "bufio"
   5         "context"
   6         "encoding/json"
   7         "errors"
   8         "fmt"
   9         "io"
  10         "log"
  11         "net"
  12         "os"
  13         "path/filepath"
  14         "sync"
  15         "time"
  16 )
  17
  18 const (
  19         // agentDirName is the directory of the root agent in the data directory.
  20         agentDirName = "agent"
  21
  22         // socketFileName is the Unix socket of the daemon in the data directory.
  23         socketFileName = "onechat.sock"
  24
  25         // tokenLockFileName is the lock file of the token file in the data
  26         // directory.
  27         tokenLockFileName = "auth.lock"
  28
  29         // requestBytes is the most bytes of one request line of a client.
  30         requestBytes = 16 << 20
  31
  32         // clientEvents is the number of events that can wait for a slow client
  33         // before the daemon drops it.
  34         clientEvents = 1024
  35
  36         // socketPathBytes is the longest path of a Unix socket on Linux.
  37         socketPathBytes = 107
  38
  39         // terminalSource is the source of the user messages from the clients of
  40         // the socket.
  41         terminalSource = "terminal"
  42 )
  43
  44 // errTokensInUse tells that another onechat process uses the token file.
  45 var errTokensInUse = errors.New("another onechat process uses the token file, such as the daemon")
  46
  47 // clientRequest is one line from a client: "send" with the text of a user
  48 // message, or "interrupt".
  49 type clientRequest struct {
  50         Type string `json:"type"`
  51         Text string `json:"text,omitempty"`
  52 }
  53
  54 // daemon runs "onechat daemon": it owns the root agent and its files and serves
  55 // the clients on the Unix socket in the data directory until ctx ends. When
  56 // telegram.json exists, it also runs the Telegram bot. It holds the lock of the
  57 // token file, so it is the only process that refreshes the tokens.
  58 func daemon(ctx context.Context, out io.Writer) (err error) {
  59         dir, err := dataDir()
  60         if err != nil {
  61                 return err
  62         }
  63
  64         // Check the socket path before the daemon writes a file, so that a data
  65         // directory with a long path stays empty.
  66         socket := filepath.Join(dir, socketFileName)
  67         err = checkSocketPath(socket)
  68         if err != nil {
  69                 return err
  70         }
  71
  72         config, err := loadConfig(dir)
  73         if err != nil {
  74                 return err
  75         }
  76
  77         botSettings, hasBot, err := loadTelegramConfig(dir)
  78         if err != nil {
  79                 return err
  80         }
  81
  82         tokens, err := lockTokens(dir)
  83         if err != nil {
  84                 return err
  85         }
  86
  87         defer func() {
  88                 err = errors.Join(err, tokens.Close())
  89         }()
  90
  91         address, auth, err := modelEndpoint(config, dir)
  92         if err != nil {
  93                 return err
  94         }
  95
  96         agent, err := openAgent(filepath.Join(dir, agentDirName))
  97         if err != nil {
  98                 return err
  99         }
 100
 101         client := newClient(address, auth, agent.CacheKey(), filepath.Join(agent.dir, "usage.jsonl"))
 102         memory, err := openMemory(ctx, agent.dir, time.Now, newCompactor(client, agent.Prompt, config.Compactor))
 103         if err != nil {
 104                 return err
 105         }
 106
 107         defer func() {
 108                 err = errors.Join(err, memory.Close())
 109         }()
 110
 111         listener, err := listenSocket(socket)
 112         if err != nil {
 113                 return err
 114         }
 115
 116         _, err = fmt.Fprintf(out, "onechat: %s on %s, %s at %s effort, in %s\n", agent.config.Name, listener.Addr(), agent.config.Model, agent.config.Effort, agent.config.Workdir)
 117         if err != nil {
 118                 return errors.Join(err, listener.Close())
 119         }
 120
 121         var telegram *TelegramBot
 122         if hasBot {
 123                 telegram = newTelegramBot(newTelegramAPI(telegramAPIURL, botSettings.Token), botSettings.Owner, filepath.Join(agent.dir, telegramOffsetFileName), log.New(out, "telegram: ", log.LstdFlags))
 124                 _, err = fmt.Fprintln(out, "onechat: the Telegram bot waits for messages of its owner")
 125                 if err != nil {
 126                         return errors.Join(err, listener.Close())
 127                 }
 128         }
 129
 130         return serveAgent(ctx, listener, telegram, func(notify func(Event)) (*loop, error) {
 131                 return newLoop(agent, memory, client, time.Now, notify)
 132         })
 133 }
 134
 135 // serveAgent runs the loop that start makes and serves the clients of
 136 // listener until ctx ends or the loop fails. It closes listener. When bot is
 137 // not nil, it runs the bot too, and a failure of the bot stops the agent.
 138 func serveAgent(ctx context.Context, listener net.Listener, bot *TelegramBot, start func(notify func(Event)) (*loop, error)) error {
 139         ctx, cancel := context.WithCancel(ctx)
 140         defer cancel()
 141
 142         hub := &hub{
 143                 clients: make(map[*hubClient]bool),
 144         }
 145
 146         notify := hub.Broadcast
 147         if bot != nil {
 148                 notify = func(event Event) {
 149                         hub.Broadcast(event)
 150                         bot.Notify(event)
 151                 }
 152         }
 153
 154         l, err := start(notify)
 155         if err != nil {
 156                 return errors.Join(err, listener.Close())
 157         }
 158
 159         var workers sync.WaitGroup
 160         workers.Add(1)
 161         go func() {
 162                 defer workers.Done()
 163
 164                 hub.Accept(ctx, listener, l)
 165         }()
 166
 167         var botErr error
 168         if bot != nil {
 169                 workers.Add(1)
 170                 go func() {
 171                         defer workers.Done()
 172
 173                         botErr = bot.Run(ctx, l)
 174                         cancel()
 175                 }()
 176         }
 177
 178         err = l.Run(ctx)
 179         if errors.Is(err, context.Canceled) {
 180                 err = nil
 181         }
 182
 183         cancel()
 184         err = errors.Join(err, listener.Close())
 185         hub.CloseAll()
 186         workers.Wait()
 187
 188         return errors.Join(err, botErr)
 189 }
 190
 191 // hub holds the connected clients and sends each event to all of them.
 192 type hub struct {
 193         mutex   sync.Mutex
 194         clients map[*hubClient]bool
 195         workers sync.WaitGroup
 196 }
 197
 198 // hubClient is one connected client. Its events wait in a channel for the
 199 // writer, so a slow client does not stop the agent.
 200 type hubClient struct {
 201         conn   net.Conn
 202         events chan Event
 203         once   sync.Once
 204 }
 205
 206 // Accept serves each client of listener until the listener closes. A new
 207 // client gets a ready event with the name of the agent, and a busy event when a
 208 // turn runs.
 209 func (h *hub) Accept(ctx context.Context, listener net.Listener, l *loop) {
 210         for {
 211                 conn, err := listener.Accept()
 212                 if err != nil {
 213                         return
 214                 }
 215
 216                 client := &hubClient{
 217                         conn:   conn,
 218                         events: make(chan Event, clientEvents),
 219                 }
 220
 221                 h.mutex.Lock()
 222                 h.clients[client] = true
 223                 h.mutex.Unlock()
 224
 225                 h.workers.Add(2)
 226                 go h.write(client)
 227                 go h.read(ctx, client, l)
 228
 229                 h.Send(client, Event{
 230                         Kind: "ready",
 231                         Text: l.agent.config.Name,
 232                 })
 233
 234                 if l.Busy() {
 235                         h.Send(client, Event{
 236                                 Kind: "busy",
 237                         })
 238                 }
 239         }
 240 }
 241
 242 // write sends the events of client as JSON lines until its channel closes.
 243 func (h *hub) write(client *hubClient) {
 244         defer h.workers.Done()
 245
 246         encoder := json.NewEncoder(client.conn)
 247         for event := range client.events {
 248                 err := encoder.Encode(event)
 249                 if err != nil {
 250                         h.drop(client)
 251                 }
 252         }
 253 }
 254
 255 // read runs the requests of client until it closes its connection.
 256 func (h *hub) read(ctx context.Context, client *hubClient, l *loop) {
 257         defer h.workers.Done()
 258         defer h.drop(client)
 259
 260         scanner := bufio.NewScanner(client.conn)
 261         scanner.Buffer(make([]byte, 64*1024), requestBytes)
 262         for ctx.Err() == nil && scanner.Scan() {
 263                 var request clientRequest
 264                 err := json.Unmarshal(scanner.Bytes(), &request)
 265                 if err == nil {
 266                         err = runRequest(l, request)
 267                 }
 268
 269                 if err != nil {
 270                         h.Send(client, Event{
 271                                 Kind: "error",
 272                                 Text: err.Error(),
 273                         })
 274                 }
 275         }
 276 }
 277
 278 // Send gives event to one client, or drops the client when its events do not
 279 // fit.
 280 func (h *hub) Send(client *hubClient, event Event) {
 281         h.mutex.Lock()
 282         defer h.mutex.Unlock()
 283
 284         h.send(client, event)
 285 }
 286
 287 // send gives event to client. The caller holds the mutex.
 288 func (h *hub) send(client *hubClient, event Event) {
 289         if !h.clients[client] {
 290                 return
 291         }
 292
 293         select {
 294         case client.events <- event:
 295         default:
 296                 h.remove(client)
 297         }
 298 }
 299
 300 // Broadcast gives event to every client.
 301 func (h *hub) Broadcast(event Event) {
 302         h.mutex.Lock()
 303         defer h.mutex.Unlock()
 304
 305         for client := range h.clients {
 306                 h.send(client, event)
 307         }
 308 }
 309
 310 // drop removes client and closes its connection.
 311 func (h *hub) drop(client *hubClient) {
 312         h.mutex.Lock()
 313         defer h.mutex.Unlock()
 314
 315         h.remove(client)
 316 }
 317
 318 // remove removes client, closes its connection, and ends its writer. The
 319 // caller holds the mutex.
 320 func (h *hub) remove(client *hubClient) {
 321         delete(h.clients, client)
 322         client.once.Do(func() {
 323                 close(client.events)
 324                 _ = client.conn.Close()
 325         })
 326 }
 327
 328 // CloseAll drops every client and waits for their readers and writers.
 329 func (h *hub) CloseAll() {
 330         h.mutex.Lock()
 331         for client := range h.clients {
 332                 h.remove(client)
 333         }
 334         h.mutex.Unlock()
 335
 336         h.workers.Wait()
 337 }
 338
 339 // runRequest runs one request of a client.
 340 func runRequest(l *loop, request clientRequest) error {
 341         switch request.Type {
 342         case "send":
 343                 if request.Text == "" {
 344                         return errors.New("empty message")
 345                 }
 346
 347                 _, err := l.Submit(terminalSource, request.Text)
 348
 349                 return err
 350         case "interrupt":
 351                 if !l.Interrupt() {
 352                         return errors.New("no turn runs")
 353                 }
 354
 355                 return nil
 356         default:
 357                 return fmt.Errorf("unknown request %q", request.Type)
 358         }
 359 }
 360
 361 // listenSocket listens on the Unix socket at path, with mode 0600. The caller
 362 // owns the chat, so a socket file at path is left from a daemon that ended,
 363 // and listenSocket removes it.
 364 func listenSocket(path string) (net.Listener, error) {
 365         err := checkSocketPath(path)
 366         if err != nil {
 367                 return nil, err
 368         }
 369
 370         err = os.Remove(path)
 371         if err != nil && !errors.Is(err, os.ErrNotExist) {
 372                 return nil, fmt.Errorf("remove old socket: %w", err)
 373         }
 374
 375         listener, err := net.Listen("unix", path)
 376         if err != nil {
 377                 return nil, fmt.Errorf("listen on socket: %w", err)
 378         }
 379
 380         err = os.Chmod(path, 0o600)
 381         if err != nil {
 382                 return nil, errors.Join(fmt.Errorf("set socket mode: %w", err), listener.Close())
 383         }
 384
 385         return listener, nil
 386 }
 387
 388 // checkSocketPath makes sure that path fits in the address of a Unix socket.
 389 func checkSocketPath(path string) error {
 390         if len(path) > socketPathBytes {
 391                 return fmt.Errorf("socket path %s is longer than %d bytes: set ONECHAT_DIR to a shorter path", path, socketPathBytes)
 392         }
 393
 394         return nil
 395 }
 396
 397 // lockTokens takes the lock of the token file in dir. Each process that reads
 398 // or refreshes the tokens holds it, because a refresh in one process rotates
 399 // the refresh token of the other and logs it out.
 400 func lockTokens(dir string) (*os.File, error) {
 401         lock, err := lockFile(filepath.Join(dir, tokenLockFileName))
 402         if errors.Is(err, errLocked) {
 403                 return nil, errTokensInUse
 404         }
 405
 406         return lock, err
 407 }