X-Git-Url: http://git.giacomolli.uk/onechat/blobdiff_plain/97274864799ac2eea84bf80c9727a810b695db55..6d1d9d9a7cf90b25a2e10af1c5bc56faee7e494c:/daemon.go?ds=sidebyside

diff --git a/daemon.go b/daemon.go
index 034cac7..cedb45e 100644
--- a/daemon.go
+++ b/daemon.go
@@ -7,6 +7,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"log"
 	"net"
 	"os"
 	"path/filepath"
@@ -34,6 +35,10 @@ const (
 
 	// socketPathBytes is the longest path of a Unix socket on Linux.
 	socketPathBytes = 107
+
+	// terminalSource is the source of the user messages from the clients of
+	// the socket.
+	terminalSource = "terminal"
 )
 
 // errTokensInUse tells that another onechat process uses the token file.
@@ -47,20 +52,33 @@ type clientRequest struct {
 }
 
 // daemon runs "onechat daemon": it owns the root agent and its files and serves
-// the clients on the Unix socket in the data directory until ctx ends. It holds
-// the lock of the token file, so it is the only process that refreshes the
-// tokens.
+// the clients on the Unix socket in the data directory until ctx ends. When
+// telegram.json exists, it also runs the Telegram bot. It holds the lock of the
+// token file, so it is the only process that refreshes the tokens.
 func daemon(ctx context.Context, out io.Writer) (err error) {
 	dir, err := dataDir()
 	if err != nil {
 		return err
 	}
 
+	// Check the socket path before the daemon writes a file, so that a data
+	// directory with a long path stays empty.
+	socket := filepath.Join(dir, socketFileName)
+	err = checkSocketPath(socket)
+	if err != nil {
+		return err
+	}
+
 	config, err := loadConfig(dir)
 	if err != nil {
 		return err
 	}
 
+	botSettings, hasBot, err := loadTelegramConfig(dir)
+	if err != nil {
+		return err
+	}
+
 	tokens, err := lockTokens(dir)
 	if err != nil {
 		return err
@@ -80,7 +98,7 @@ func daemon(ctx context.Context, out io.Writer) (err error) {
 		return err
 	}
 
-	client := newClient(address, auth, agent.CacheKey(), filepath.Join(agent.dir, "usage.jsonl"))
+	client := newModelClient(address, auth, agent.CacheKey(), filepath.Join(agent.dir, "usage.jsonl"))
 	memory, err := openMemory(ctx, agent.dir, time.Now, newCompactor(client, agent.Prompt, config.Compactor))
 	if err != nil {
 		return err
@@ -90,7 +108,7 @@ func daemon(ctx context.Context, out io.Writer) (err error) {
 		err = errors.Join(err, memory.Close())
 	}()
 
-	listener, err := listenSocket(filepath.Join(dir, socketFileName))
+	listener, err := listenSocket(socket)
 	if err != nil {
 		return err
 	}
@@ -100,14 +118,24 @@ func daemon(ctx context.Context, out io.Writer) (err error) {
 		return errors.Join(err, listener.Close())
 	}
 
-	return serveAgent(ctx, listener, func(notify func(Event)) (*loop, error) {
+	var telegram *TelegramBot
+	if hasBot {
+		telegram = newTelegramBot(newTelegramAPI(telegramAPIURL, botSettings.Token), botSettings.Owner, filepath.Join(agent.dir, telegramOffsetFileName), log.New(out, "telegram: ", log.LstdFlags))
+		_, err = fmt.Fprintln(out, "onechat: the Telegram bot waits for messages of its owner")
+		if err != nil {
+			return errors.Join(err, listener.Close())
+		}
+	}
+
+	return serveAgent(ctx, listener, telegram, func(notify func(Event)) (*loop, error) {
 		return newLoop(agent, memory, client, time.Now, notify)
 	})
 }
 
 // serveAgent runs the loop that start makes and serves the clients of
-// listener until ctx ends or the loop fails. It closes listener.
-func serveAgent(ctx context.Context, listener net.Listener, start func(notify func(Event)) (*loop, error)) error {
+// listener until ctx ends or the loop fails. It closes listener. When bot is
+// not nil, it runs the bot too, and a failure of the bot stops the agent.
+func serveAgent(ctx context.Context, listener net.Listener, bot *TelegramBot, start func(notify func(Event)) (*loop, error)) error {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
@@ -115,7 +143,15 @@ func serveAgent(ctx context.Context, listener net.Listener, start func(notify fu
 		clients: make(map[*hubClient]bool),
 	}
 
-	l, err := start(hub.Broadcast)
+	notify := hub.Broadcast
+	if bot != nil {
+		notify = func(event Event) {
+			hub.Broadcast(event)
+			bot.Notify(event)
+		}
+	}
+
+	l, err := start(notify)
 	if err != nil {
 		return errors.Join(err, listener.Close())
 	}
@@ -128,17 +164,28 @@ func serveAgent(ctx context.Context, listener net.Listener, start func(notify fu
 		hub.Accept(ctx, listener, l)
 	}()
 
+	var botErr error
+	if bot != nil {
+		workers.Add(1)
+		go func() {
+			defer workers.Done()
+
+			botErr = bot.Run(ctx, l)
+			cancel()
+		}()
+	}
+
 	err = l.Run(ctx)
+	if errors.Is(err, context.Canceled) {
+		err = nil
+	}
+
 	cancel()
 	err = errors.Join(err, listener.Close())
 	hub.CloseAll()
 	workers.Wait()
 
-	if errors.Is(err, context.Canceled) {
-		return nil
-	}
-
-	return err
+	return errors.Join(err, botErr)
 }
 
 // hub holds the connected clients and sends each event to all of them.
@@ -297,7 +344,7 @@ func runRequest(l *loop, request clientRequest) error {
 			return errors.New("empty message")
 		}
 
-		_, err := l.Submit(request.Text)
+		_, err := l.Submit(terminalSource, request.Text)
 
 		return err
 	case "interrupt":