X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/09cde69fdffe2721434392036f98dc241177f5a1..4c199e7d93b6a4529ed5956807911b5f6c5489a7:/daemon.go

diff --git a/daemon.go b/daemon.go
index 5486843..5ec220e 100644
--- a/daemon.go
+++ b/daemon.go
@@ -16,45 +16,21 @@ import (
 )
 
 const (
-	// agentDirName is the directory of the root agent in the data directory.
-	agentDirName = "agent"
-
-	// socketFileName is the Unix socket of the daemon in the data directory.
-	socketFileName = "onechat.sock"
-
-	// tokenLockFileName is the lock file of the token file in the data
+	// daemonAgentDirName is the directory of the root agent in the data
 	// directory.
-	tokenLockFileName = "auth.lock"
-
-	// requestBytes is the most bytes of one request line of a client.
-	requestBytes = 16 << 20
-
-	// clientEvents is the number of events that can wait for a slow client
-	// before the daemon drops it.
-	clientEvents = 1024
+	daemonAgentDirName = "agent"
 
-	// 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"
+	// daemonClientEvents is the number of events that can wait for a slow
+	// client before the daemon drops it.
+	daemonClientEvents = 1024
 )
 
-// errTokensInUse tells that another onechat process uses the token file.
-var errTokensInUse = errors.New("another onechat process uses the token file, such as the daemon")
-
-// clientRequest is one line from a client: "send" with the text of a user
-// message, or "interrupt".
-type clientRequest struct {
-	Type string `json:"type"`
-	Text string `json:"text,omitempty"`
-}
-
 // 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. 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.
+// telegram.json exists, it also runs the Telegram bot. It also keeps a linked
+// WhatsApp device connected. It holds the lock of the token file, so it is the
+// only process that refreshes the tokens and opens the WhatsApp session. Its
+// log lines go to out, through the loggers of daemonLogger.
 func daemon(ctx context.Context, out io.Writer) (err error) {
 	dir, err := dataDir()
 	if err != nil {
@@ -88,18 +64,30 @@ func daemon(ctx context.Context, out io.Writer) (err error) {
 		err = errors.Join(err, tokens.Close())
 	}()
 
+	whatsappSetDeviceProps()
+
+	logger := daemonLogger(out, "onechat: ")
+	whatsapp, err := openWhatsApp(ctx, dir, time.Now, daemonLogger(out, "whatsapp: "), whatsappNewClient)
+	if err != nil {
+		return err
+	}
+
+	defer func() {
+		err = errors.Join(err, whatsapp.Close())
+	}()
+
 	address, auth, err := modelEndpoint(config, dir)
 	if err != nil {
 		return err
 	}
 
-	agent, err := openAgent(filepath.Join(dir, agentDirName))
+	agent, err := openAgent(filepath.Join(dir, daemonAgentDirName))
 	if err != nil {
 		return err
 	}
 
-	client := newClient(address, auth, agent.CacheKey(), filepath.Join(agent.dir, "usage.jsonl"))
-	memory, err := openMemory(ctx, agent.dir, time.Now, newCompactor(client, agent.Prompt, config.Compactor))
+	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), logger)
 	if err != nil {
 		return err
 	}
@@ -108,121 +96,73 @@ func daemon(ctx context.Context, out io.Writer) (err error) {
 		err = errors.Join(err, memory.Close())
 	}()
 
-	listener, err := listenSocket(socket)
+	schedules, err := openSchedules(filepath.Join(agent.dir, scheduleFileName), time.Now, logger)
 	if err != nil {
 		return err
 	}
 
-	_, 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)
-	if err != nil {
-		return errors.Join(err, listener.Close())
-	}
-
-	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. 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()
-
-	hub := &hub{
-		clients: make(map[*hubClient]bool),
-	}
-
-	notify := hub.Broadcast
-	if bot != nil {
-		notify = func(event Event) {
-			hub.Broadcast(event)
-			bot.Notify(event)
-		}
-	}
+	subagents := newSubagents(filepath.Join(agent.dir, subagentDirName), time.Now, subagentNameRandom)
+	defer func() {
+		err = errors.Join(err, subagents.Close())
+	}()
 
-	l, err := start(notify)
+	listener, err := daemonListenSocket(socket)
 	if err != nil {
-		return errors.Join(err, listener.Close())
+		return err
 	}
 
-	var workers sync.WaitGroup
-	workers.Add(1)
-	go func() {
-		defer workers.Done()
+	linked := "WhatsApp not linked"
+	if whatsapp.isLinked() {
+		linked = "WhatsApp linked"
+	}
 
-		hub.Accept(ctx, listener, l)
-	}()
+	logger.Printf("ready: socket %s, model %s (%s), workdir %s, %s, %d schedules", listener.Addr(), agent.config.Model, agent.config.Effort, agent.config.Workdir, linked, schedules.Len())
 
-	var botErr error
-	if bot != nil {
-		workers.Add(1)
-		go func() {
-			defer workers.Done()
+	var telegram *TelegramBot
 
-			botErr = bot.Run(ctx, l)
-			cancel()
-		}()
-	}
+	if hasBot {
+		telegram = newTelegramBot(newTelegramAPI(telegramAPIURL, botSettings.Token), botSettings.Owner, filepath.Join(agent.dir, telegramOffsetFileName), daemonLogger(out, "telegram: "))
 
-	err = l.Run(ctx)
-	if errors.Is(err, context.Canceled) {
-		err = nil
+		logger.Print("Telegram bot started")
 	}
 
-	cancel()
-	err = errors.Join(err, listener.Close())
-	hub.CloseAll()
-	workers.Wait()
-
-	return errors.Join(err, botErr)
+	return daemonServe(ctx, listener, telegram, whatsapp, schedules, func(notify func(Event)) (*TurnLoop, error) {
+		return newTurnLoop(agent, memory, subagents, whatsapp, schedules, client, time.Now, logger, notify)
+	})
 }
 
-// hub holds the connected clients and sends each event to all of them.
-type hub struct {
+// DaemonHub holds the connected clients and sends each event to all of them. It
+// runs the WhatsApp commands of the clients on whatsapp.
+type DaemonHub struct {
+	whatsapp *WhatsApp
+
 	mutex   sync.Mutex
-	clients map[*hubClient]bool
+	clients map[*DaemonClient]bool
 	workers sync.WaitGroup
 }
 
-// hubClient is one connected client. Its events wait in a channel for the
-// writer, so a slow client does not stop the agent.
-type hubClient struct {
-	conn   net.Conn
-	events chan Event
-	once   sync.Once
-}
-
 // Accept serves each client of listener until the listener closes. A new
 // client gets a ready event with the name of the agent, and a busy event when a
 // turn runs.
-func (h *hub) Accept(ctx context.Context, listener net.Listener, l *loop) {
+func (h *DaemonHub) Accept(ctx context.Context, listener net.Listener, l *TurnLoop) {
 	for {
 		conn, err := listener.Accept()
 		if err != nil {
 			return
 		}
 
-		client := &hubClient{
+		client := &DaemonClient{
 			conn:   conn,
-			events: make(chan Event, clientEvents),
+			events: make(chan Event, daemonClientEvents),
 		}
 
 		h.mutex.Lock()
+
 		h.clients[client] = true
 		h.mutex.Unlock()
 
 		h.workers.Add(2)
+
 		go h.write(client)
 		go h.read(ctx, client, l)
 
@@ -240,7 +180,7 @@ func (h *hub) Accept(ctx context.Context, listener net.Listener, l *loop) {
 }
 
 // write sends the events of client as JSON lines until its channel closes.
-func (h *hub) write(client *hubClient) {
+func (h *DaemonHub) write(client *DaemonClient) {
 	defer h.workers.Done()
 
 	encoder := json.NewEncoder(client.conn)
@@ -253,17 +193,24 @@ func (h *hub) write(client *hubClient) {
 }
 
 // read runs the requests of client until it closes its connection.
-func (h *hub) read(ctx context.Context, client *hubClient, l *loop) {
+func (h *DaemonHub) read(ctx context.Context, client *DaemonClient, l *TurnLoop) {
 	defer h.workers.Done()
 	defer h.drop(client)
 
 	scanner := bufio.NewScanner(client.conn)
-	scanner.Buffer(make([]byte, 64*1024), requestBytes)
+	scanner.Buffer(make([]byte, 64*1024), socketRequestBytes)
+
 	for ctx.Err() == nil && scanner.Scan() {
-		var request clientRequest
+		var request SocketRequest
+
 		err := json.Unmarshal(scanner.Bytes(), &request)
 		if err == nil {
-			err = runRequest(l, request)
+			err = daemonRunRequest(l, h.whatsapp, request, func(text string) {
+				h.Send(client, Event{
+					Kind: "notice",
+					Text: text,
+				})
+			})
 		}
 
 		if err != nil {
@@ -277,15 +224,23 @@ func (h *hub) read(ctx context.Context, client *hubClient, l *loop) {
 
 // Send gives event to one client, or drops the client when its events do not
 // fit.
-func (h *hub) Send(client *hubClient, event Event) {
+func (h *DaemonHub) Send(client *DaemonClient, event Event) {
 	h.mutex.Lock()
 	defer h.mutex.Unlock()
 
 	h.send(client, event)
 }
 
+// drop removes client and closes its connection.
+func (h *DaemonHub) drop(client *DaemonClient) {
+	h.mutex.Lock()
+	defer h.mutex.Unlock()
+
+	h.remove(client)
+}
+
 // send gives event to client. The caller holds the mutex.
-func (h *hub) send(client *hubClient, event Event) {
+func (h *DaemonHub) send(client *DaemonClient, event Event) {
 	if !h.clients[client] {
 		return
 	}
@@ -297,71 +252,52 @@ func (h *hub) send(client *hubClient, event Event) {
 	}
 }
 
-// Broadcast gives event to every client.
-func (h *hub) Broadcast(event Event) {
-	h.mutex.Lock()
-	defer h.mutex.Unlock()
-
-	for client := range h.clients {
-		h.send(client, event)
-	}
-}
-
-// drop removes client and closes its connection.
-func (h *hub) drop(client *hubClient) {
-	h.mutex.Lock()
-	defer h.mutex.Unlock()
-
-	h.remove(client)
-}
-
 // remove removes client, closes its connection, and ends its writer. The
 // caller holds the mutex.
-func (h *hub) remove(client *hubClient) {
+func (h *DaemonHub) remove(client *DaemonClient) {
 	delete(h.clients, client)
 	client.once.Do(func() {
 		close(client.events)
+
 		_ = client.conn.Close()
 	})
 }
 
+// Broadcast gives event to every client.
+func (h *DaemonHub) Broadcast(event Event) {
+	h.mutex.Lock()
+	defer h.mutex.Unlock()
+
+	for client := range h.clients {
+		h.send(client, event)
+	}
+}
+
 // CloseAll drops every client and waits for their readers and writers.
-func (h *hub) CloseAll() {
+func (h *DaemonHub) CloseAll() {
 	h.mutex.Lock()
+
 	for client := range h.clients {
 		h.remove(client)
 	}
+
 	h.mutex.Unlock()
 
 	h.workers.Wait()
 }
 
-// runRequest runs one request of a client.
-func runRequest(l *loop, request clientRequest) error {
-	switch request.Type {
-	case "send":
-		if request.Text == "" {
-			return errors.New("empty message")
-		}
-
-		_, err := l.Submit(terminalSource, request.Text)
-
-		return err
-	case "interrupt":
-		if !l.Interrupt() {
-			return errors.New("no turn runs")
-		}
-
-		return nil
-	default:
-		return fmt.Errorf("unknown request %q", request.Type)
-	}
+// DaemonClient is one connected client. Its events wait in a channel for the
+// writer, so a slow client does not stop the agent.
+type DaemonClient struct {
+	conn   net.Conn
+	events chan Event
+	once   sync.Once
 }
 
-// listenSocket listens on the Unix socket at path, with mode 0600. The caller
-// owns the chat, so a socket file at path is left from a daemon that ended,
-// and listenSocket removes it.
-func listenSocket(path string) (net.Listener, error) {
+// daemonListenSocket listens on the Unix socket at path, with mode 0600. The
+// caller owns the chat, so a socket file at path is left from a daemon that
+// ended, and daemonListenSocket removes it.
+func daemonListenSocket(path string) (net.Listener, error) {
 	err := checkSocketPath(path)
 	if err != nil {
 		return nil, err
@@ -385,23 +321,147 @@ func listenSocket(path string) (net.Listener, error) {
 	return listener, nil
 }
 
-// checkSocketPath makes sure that path fits in the address of a Unix socket.
-func checkSocketPath(path string) error {
-	if len(path) > socketPathBytes {
-		return fmt.Errorf("socket path %s is longer than %d bytes: set ONECHAT_DIR to a shorter path", path, socketPathBytes)
+// daemonServe runs the loop that start makes and serves the clients of listener
+// until ctx ends or the loop fails. It closes listener. It runs whatsapp, which
+// logs its batches and notes in the loop and tells all clients its notices, and
+// a failure of whatsapp stops the agent. It runs schedules, which log their
+// runs in the loop, and a failure of schedules stops the agent too. When bot is
+// not nil, it runs the bot too, and a failure of the bot stops the agent.
+func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, whatsapp *WhatsApp, schedules *Schedules, start func(notify func(Event)) (*TurnLoop, error)) error {
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	hub := &DaemonHub{
+		whatsapp: whatsapp,
+		clients:  make(map[*DaemonClient]bool),
+	}
+
+	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())
+	}
+
+	var workers sync.WaitGroup
+
+	workers.Add(2)
+
+	go func() {
+		defer workers.Done()
+
+		hub.Accept(ctx, listener, l)
+	}()
+
+	var whatsappErr error
+
+	go func() {
+		defer workers.Done()
+
+		whatsappErr = whatsapp.Run(ctx, l, notify)
+
+		cancel()
+	}()
+
+	var schedulesErr error
+
+	workers.Add(1)
+
+	go func() {
+		defer workers.Done()
+
+		schedulesErr = schedules.Run(ctx, l)
+
+		cancel()
+	}()
+
+	var botErr error
+
+	if bot != nil {
+		workers.Add(1)
+
+		go func() {
+			defer workers.Done()
+
+			botErr = bot.Run(ctx, l, whatsapp)
+
+			cancel()
+		}()
+	}
+
+	err = l.Run(ctx)
+	if errors.Is(err, context.Canceled) {
+		err = nil
+	}
+
+	cancel()
+
+	err = errors.Join(err, listener.Close())
+
+	hub.CloseAll()
+	workers.Wait()
+
+	return errors.Join(err, whatsappErr, schedulesErr, botErr)
+}
+
+// daemonRunRequest runs one request of a client. The answers of a WhatsApp
+// command go to answer.
+func daemonRunRequest(l *TurnLoop, whatsapp *WhatsApp, request SocketRequest, answer func(text string)) error {
+	switch request.Type {
+	case "send":
+		if request.Text == "" {
+			return errors.New("empty message")
+		}
+
+		_, err := l.Submit(terminalSource, request.Text)
+
+		return err
+	case "interrupt":
+		if !l.Interrupt() {
+			return errors.New("no turn runs")
+		}
+
+		return nil
+	case "whatsapp":
+		whatsapp.Command(request.Text, answer)
+
+		return nil
+	case "stop":
+		found, err := l.StopSubagent(request.Text)
+		if err == nil && !found {
+			err = fmt.Errorf("no subagent %s runs", request.Text)
+		}
+
+		return err
+	default:
+		return fmt.Errorf("unknown request %q", request.Type)
+	}
+}
+
+// daemonLogger returns the logger of the part of the daemon with prefix, such
+// as "telegram: ", on out. The journal adds the time to each line and sets
+// JOURNAL_STREAM for a service, so then the lines have no time stamp. Otherwise
+// each line has the standard time stamp.
+func daemonLogger(out io.Writer, prefix string) *log.Logger {
+	if os.Getenv("JOURNAL_STREAM") != "" {
+		return log.New(out, prefix, 0)
 	}
 
-	return nil
+	return log.New(out, prefix, log.LstdFlags)
 }
 
-// lockTokens takes the lock of the token file in dir. Each process that reads
-// or refreshes the tokens holds it, because a refresh in one process rotates
-// the refresh token of the other and logs it out.
-func lockTokens(dir string) (*os.File, error) {
-	lock, err := lockFile(filepath.Join(dir, tokenLockFileName))
-	if errors.Is(err, errLocked) {
-		return nil, errTokensInUse
+// durationText returns d for a log line: rounded to the second, or to the
+// millisecond when it is shorter than a second.
+func durationText(d time.Duration) string {
+	if d < time.Second {
+		return d.Round(time.Millisecond).String()
 	}
 
-	return lock, err
+	return d.Round(time.Second).String()
 }