Serve the agent to a terminal client

onechat daemon owns the root agent in agent/ in the data directory: its
config, chat, compactor, and loop. It listens on the Unix socket
onechat.sock in the data directory, with mode 0600, and sends each event
to every client as a JSON line: ready with the agent's name on connect,
busy and idle around each turn, each logged message, and errors. A
client sends JSON lines too: send with a message, or interrupt. A slow
client is dropped, so it never stops the agent.

onechat chat is a line-based client without a terminal library. Each
input line is a message, the line /stop interrupts the turn, and the end
of the input ends the client. It shows the replies, the tool calls, and
the first 10 lines of each result as they happen.

The daemon, login, and probe each hold a lock on auth.lock in the data
directory while they run, so only one process reads and refreshes the
tokens. Two processes that refresh one token file rotate the refresh
token of each other and log each other out. The client never reads the
token file.

Issue: 5daf23e

diff --git a/README.rst b/README.rst
index 04dcafb923096b07be3965e6b847bc706e8f39d0..c0c6c5d2b16a6457a56a6716229ceb37b8ecd989 100644 (file)
--- a/README.rst
+++ b/README.rst
@@ -2,8 +2,9 @@
  OneChat
 #########
 
  OneChat
 #########
 

-OneChat is a Go program. It prints its name. It is in early development.
-Linux is supported.
+OneChat is one AI agent in a single chat that never ends. A daemon owns
+the agent and its files, and a terminal client talks to it. It is in
+early development. Linux is supported.

 
 ****************
  Data directory
 
 ****************
  Data directory


@@ -63,6 +64,48 @@ before they expire. The file belongs to OneChat only. Do not share it
 with the Codex CLI, because each program rotates the refresh token and
 logs the other out.
 
 with the Codex CLI, because each program rotates the refresh token and
 logs the other out.
 

+*******
+ Agent
+*******
+
+Start the daemon, and then the client in another terminal:
+
+.. code:: console
+
+   onechat daemon
+   onechat chat
+
+The daemon owns the root agent in ``agent/`` in the data directory and
+listens on the Unix socket ``onechat.sock`` there. In the client, each
+line is a message to the agent, the line ``/stop`` interrupts the
+running turn, and Ctrl-D ends the client. The client shows the replies,
+the tool calls, and the first lines of each tool result as they happen.
+The daemon is the only process that uses the token file while it runs,
+so ``onechat login`` and ``onechat probe`` refuse to start until it
+stops.
+
+The agent runs commands with bash as the user in its working directory,
+with no sandbox. It has the rights of the user, also on the token file.
+
+``agent/config.json`` holds the settings of the agent. The first start
+writes it with the defaults and a random ID. Do not change the ID: the
+prompt cache key comes from it.
+
+.. code:: json
+
+   {
+     "name": "OneChat",
+     "id": "0123456789abcdef0123456789abcdef",
+     "model": "gpt-6.1-sol",
+     "effort": "medium",
+     "workdir": "/home/user"
+   }
+
+``workdir`` is an absolute path, and its default is the home directory.
+``agent/instructions.md`` holds the user's instructions: who they are,
+how their files are organized, and how they want work done. The prompt
+ends with them, and a change applies at the next call.
+

 *******
  Probe
 *******
 *******
  Probe
 *******

diff --git a/auth.go b/auth.go
index 736624e37e0850df3f21f116c60862815282f571..fda78d8ac709473f20a0a559506e0652b220e7ba 100644 (file)
--- a/auth.go
+++ b/auth.go
@@ -97,13 +97,23 @@ type deviceGrant struct {
 }
 
 // login runs "onechat login": the device code login into the token file of the
 }
 
 // login runs "onechat login": the device code login into the token file of the

-// data directory.
-func login(ctx context.Context, out io.Writer) error {
+// data directory. It holds the lock of the token file, so it does not run while
+// the daemon runs.
+func login(ctx context.Context, out io.Writer) (err error) {

        dir, err := dataDir()
        if err != nil {
                return err
        }
 
        dir, err := dataDir()
        if err != nil {
                return err
        }
 

+       tokens, err := lockTokens(dir)
+       if err != nil {
+               return err
+       }
+
+       defer func() {
+               err = errors.Join(err, tokens.Close())
+       }()
+

        return newChatGPTAuth(filepath.Join(dir, tokenFileName), authIssuer).Login(ctx, out)
 }
 
        return newChatGPTAuth(filepath.Join(dir, tokenFileName), authIssuer).Login(ctx, out)
 }
 

diff --git a/chat.go b/chat.go
new file mode 100644 (file)
index 0000000..8bc5285
--- /dev/null
+++ b/chat.go
@@ -0,0 +1,179 @@
+package main
+
+import (
+       "bufio"
+       "context"
+       "encoding/json"
+       "errors"
+       "fmt"
+       "io"
+       "net"
+       "path/filepath"
+       "strings"
+)
+
+const (
+       // stopCommand is the input line that interrupts the running turn.
+       stopCommand = "/stop"
+
+       // resultLines is the most lines of a tool result that the client shows.
+       resultLines = 10
+)
+
+// chat runs "onechat chat": the terminal client of the daemon. Each line of
+// input is a user message, and the line /stop interrupts the running turn. The
+// client shows the replies, the tool calls, and their results as they happen.
+// It ends at the end of the input or when the daemon closes the connection.
+func chat(ctx context.Context, in io.Reader, out io.Writer) error {
+       dir, err := dataDir()
+       if err != nil {
+               return err
+       }
+
+       path := filepath.Join(dir, socketFileName)
+       err = checkSocketPath(path)
+       if err != nil {
+               return err
+       }
+
+       var dialer net.Dialer
+       conn, err := dialer.DialContext(ctx, "unix", path)
+       if err != nil {
+               return fmt.Errorf(`connect to the daemon (start it with "onechat daemon"): %w`, err)
+       }
+
+       return chatSession(ctx, conn, in, out)
+}
+
+// errDaemonClosed tells that the daemon closed the connection.
+var errDaemonClosed = errors.New("the daemon closed the connection")
+
+// chatSession runs the client on conn until the input ends, the daemon closes
+// the connection, or ctx ends, and closes conn. A read of the input that waits
+// for a line can stay behind when the daemon closes the connection.
+func chatSession(ctx context.Context, conn net.Conn, in io.Reader, out io.Writer) error {
+       sent := make(chan error, 1)
+       go func() {
+               sent <- sendLines(ctx, conn, in)
+       }()
+
+       shown := make(chan error, 1)
+       go func() {
+               shown <- showEvents(conn, out)
+       }()
+
+       var err error
+       select {
+       case err = <-sent:
+               err = errors.Join(err, conn.Close())
+               <-shown
+       case err = <-shown:
+               err = errors.Join(cmpError(err, errDaemonClosed), conn.Close())
+       case <-ctx.Done():
+               err = conn.Close()
+               <-shown
+       }
+
+       return err
+}
+
+// cmpError returns err, or fallback when err is nil.
+func cmpError(err, fallback error) error {
+       if err == nil {
+               return fallback
+       }
+
+       return err
+}
+
+// sendLines sends each line of in as a request until the input ends.
+func sendLines(ctx context.Context, conn net.Conn, in io.Reader) error {
+       encoder := json.NewEncoder(conn)
+       scanner := bufio.NewScanner(in)
+       scanner.Buffer(make([]byte, 64*1024), requestBytes)
+       for ctx.Err() == nil && scanner.Scan() {
+               line := scanner.Text()
+               request := clientRequest{
+                       Type: "send",
+                       Text: line,
+               }
+
+               switch strings.TrimSpace(line) {
+               case "":
+                       continue
+               case stopCommand:
+                       request = clientRequest{
+                               Type: "interrupt",
+                       }
+               }
+
+               err := encoder.Encode(request)
+               if err != nil {
+                       return fmt.Errorf("send to the daemon: %w", err)
+               }
+       }
+
+       return scanner.Err()
+}
+
+// showEvents writes each event from the daemon to out until the connection
+// closes.
+func showEvents(conn net.Conn, out io.Writer) error {
+       scanner := bufio.NewScanner(conn)
+       scanner.Buffer(make([]byte, 64*1024), requestBytes)
+       for scanner.Scan() {
+               var event Event
+               err := json.Unmarshal(scanner.Bytes(), &event)
+               if err != nil {
+                       return fmt.Errorf("decode event: %w", err)
+               }
+
+               text := eventText(event)
+               if text == "" {
+                       continue
+               }
+
+               _, err = io.WriteString(out, text+"\n")
+               if err != nil {
+                       return err
+               }
+       }
+
+       return scanner.Err()
+}
+
+// eventText returns the text that the client shows for event, or "" for none.
+// The client shows no user message, because the user typed it. A tool result
+// shows its first resultLines lines.
+func eventText(event Event) string {
+       switch event.Kind {
+       case "ready":
+               return "[connected to " + event.Text + "]"
+       case "reply":
+               return event.Text
+       case "tool":
+               return "[tool] " + event.Text
+       case "echo":
+               return "[result] " + firstLines(event.Text, resultLines)
+       case "busy":
+               return "[working]"
+       case "idle":
+               return "[done]"
+       case "error":
+               return "[error] " + event.Text
+       default:
+               return ""
+       }
+}
+
+// firstLines returns the first count lines of text, with the number of the
+// lines left out.
+func firstLines(text string, count int) string {
+       text = strings.TrimRight(text, "\n")
+       lines := strings.Split(text, "\n")
+       if len(lines) <= count {
+               return text
+       }
+
+       return strings.Join(lines[:count], "\n") + fmt.Sprintf("\n[%d more lines]", len(lines)-count)
+}

diff --git a/daemon.go b/daemon.go
new file mode 100644 (file)
index 0000000..034cac7
--- /dev/null
+++ b/daemon.go
@@ -0,0 +1,360 @@
+package main
+
+import (
+       "bufio"
+       "context"
+       "encoding/json"
+       "errors"
+       "fmt"
+       "io"
+       "net"
+       "os"
+       "path/filepath"
+       "sync"
+       "time"
+)
+
+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
+       // 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
+
+       // socketPathBytes is the longest path of a Unix socket on Linux.
+       socketPathBytes = 107
+)
+
+// 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. 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
+       }
+
+       config, err := loadConfig(dir)
+       if err != nil {
+               return err
+       }
+
+       tokens, err := lockTokens(dir)
+       if err != nil {
+               return err
+       }
+
+       defer func() {
+               err = errors.Join(err, tokens.Close())
+       }()
+
+       address, auth, err := modelEndpoint(config, dir)
+       if err != nil {
+               return err
+       }
+
+       agent, err := openAgent(filepath.Join(dir, agentDirName))
+       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))
+       if err != nil {
+               return err
+       }
+
+       defer func() {
+               err = errors.Join(err, memory.Close())
+       }()
+
+       listener, err := listenSocket(filepath.Join(dir, socketFileName))
+       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())
+       }
+
+       return serveAgent(ctx, listener, 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 {
+       ctx, cancel := context.WithCancel(ctx)
+       defer cancel()
+
+       hub := &hub{
+               clients: make(map[*hubClient]bool),
+       }
+
+       l, err := start(hub.Broadcast)
+       if err != nil {
+               return errors.Join(err, listener.Close())
+       }
+
+       var workers sync.WaitGroup
+       workers.Add(1)
+       go func() {
+               defer workers.Done()
+
+               hub.Accept(ctx, listener, l)
+       }()
+
+       err = l.Run(ctx)
+       cancel()
+       err = errors.Join(err, listener.Close())
+       hub.CloseAll()
+       workers.Wait()
+
+       if errors.Is(err, context.Canceled) {
+               return nil
+       }
+
+       return err
+}
+
+// hub holds the connected clients and sends each event to all of them.
+type hub struct {
+       mutex   sync.Mutex
+       clients map[*hubClient]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) {
+       for {
+               conn, err := listener.Accept()
+               if err != nil {
+                       return
+               }
+
+               client := &hubClient{
+                       conn:   conn,
+                       events: make(chan Event, clientEvents),
+               }
+
+               h.mutex.Lock()
+               h.clients[client] = true
+               h.mutex.Unlock()
+
+               h.workers.Add(2)
+               go h.write(client)
+               go h.read(ctx, client, l)
+
+               h.Send(client, Event{
+                       Kind: "ready",
+                       Text: l.agent.config.Name,
+               })
+
+               if l.Busy() {
+                       h.Send(client, Event{
+                               Kind: "busy",
+                       })
+               }
+       }
+}
+
+// write sends the events of client as JSON lines until its channel closes.
+func (h *hub) write(client *hubClient) {
+       defer h.workers.Done()
+
+       encoder := json.NewEncoder(client.conn)
+       for event := range client.events {
+               err := encoder.Encode(event)
+               if err != nil {
+                       h.drop(client)
+               }
+       }
+}
+
+// read runs the requests of client until it closes its connection.
+func (h *hub) read(ctx context.Context, client *hubClient, l *loop) {
+       defer h.workers.Done()
+       defer h.drop(client)
+
+       scanner := bufio.NewScanner(client.conn)
+       scanner.Buffer(make([]byte, 64*1024), requestBytes)
+       for ctx.Err() == nil && scanner.Scan() {
+               var request clientRequest
+               err := json.Unmarshal(scanner.Bytes(), &request)
+               if err == nil {
+                       err = runRequest(l, request)
+               }
+
+               if err != nil {
+                       h.Send(client, Event{
+                               Kind: "error",
+                               Text: err.Error(),
+                       })
+               }
+       }
+}
+
+// Send gives event to one client, or drops the client when its events do not
+// fit.
+func (h *hub) Send(client *hubClient, event Event) {
+       h.mutex.Lock()
+       defer h.mutex.Unlock()
+
+       h.send(client, event)
+}
+
+// send gives event to client. The caller holds the mutex.
+func (h *hub) send(client *hubClient, event Event) {
+       if !h.clients[client] {
+               return
+       }
+
+       select {
+       case client.events <- event:
+       default:
+               h.remove(client)
+       }
+}
+
+// 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) {
+       delete(h.clients, client)
+       client.once.Do(func() {
+               close(client.events)
+               _ = client.conn.Close()
+       })
+}
+
+// CloseAll drops every client and waits for their readers and writers.
+func (h *hub) 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(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)
+       }
+}
+
+// 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) {
+       err := checkSocketPath(path)
+       if err != nil {
+               return nil, err
+       }
+
+       err = os.Remove(path)
+       if err != nil && !errors.Is(err, os.ErrNotExist) {
+               return nil, fmt.Errorf("remove old socket: %w", err)
+       }
+
+       listener, err := net.Listen("unix", path)
+       if err != nil {
+               return nil, fmt.Errorf("listen on socket: %w", err)
+       }
+
+       err = os.Chmod(path, 0o600)
+       if err != nil {
+               return nil, errors.Join(fmt.Errorf("set socket mode: %w", err), listener.Close())
+       }
+
+       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)
+       }
+
+       return nil
+}
+
+// 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
+       }
+
+       return lock, err
+}

diff --git a/daemon_test.go b/daemon_test.go
new file mode 100644 (file)
index 0000000..91d6ecc
--- /dev/null
+++ b/daemon_test.go
@@ -0,0 +1,282 @@
+package main
+
+import (
+       "bytes"
+       "context"
+       "io"
+       "net"
+       "net/http"
+       "os"
+       "path/filepath"
+       "strings"
+       "sync"
+       "testing"
+       "time"
+
+       "github.com/stretchr/testify/require"
+)
+
+func TestClientsSeeTheTurnAsItHappens(t *testing.T) {
+       server := newFakeResponses(t)
+       server.reply = func(w http.ResponseWriter, call fakeCall) {
+               if call.number == 0 {
+                       writeEvents(w, completedEvent([]any{functionCallItem("fc_1", "call_1", "shell", `{"command":"echo hi"}`)}, 100, 0, 5))
+                       return
+               }
+
+               writeEvents(w, completedEvent([]any{messageItem("It said hi.")}, 100, 0, 5))
+       }
+
+       socket := startTestDaemon(t, server)
+       first := startTestClient(t, socket)
+       second := startTestClient(t, socket)
+       first.WaitFor(t, "[connected to OneChat]")
+       second.WaitFor(t, "[connected to OneChat]")
+
+       first.Type("Run echo hi.")
+       for _, client := range []*testClient{first, second} {
+               client.WaitFor(t, "[done]")
+               require.Equal(t, "[connected to OneChat]\n[working]\n[tool] shell {\"command\":\"echo hi\"}\n[result] hi\nIt said hi.\n[done]\n", client.Output())
+       }
+
+       first.Type(stopCommand)
+       first.WaitFor(t, "[error] no turn runs")
+       require.NotContains(t, second.Output(), "no turn runs")
+       require.NoError(t, first.Close())
+}
+
+func TestStopInterruptsTheTurn(t *testing.T) {
+       server := newFakeResponses(t)
+       arrived := make(chan struct{})
+       release := make(chan struct{})
+       t.Cleanup(func() {
+               close(release)
+       })
+
+       server.reply = func(w http.ResponseWriter, call fakeCall) {
+               close(arrived)
+               <-release
+       }
+
+       socket := startTestDaemon(t, server)
+       client := startTestClient(t, socket)
+       client.Type("a long task")
+       <-arrived
+
+       late := startTestClient(t, socket)
+       late.WaitFor(t, "[connected to OneChat]\n[working]")
+
+       client.Type(stopCommand)
+       client.WaitFor(t, "[error] interrupted\n[done]")
+}
+
+func TestClientEndsWhenTheDaemonEnds(t *testing.T) {
+       dir := t.TempDir()
+       listener, err := listenSocket(filepath.Join(dir, socketFileName))
+       require.NoError(t, err)
+
+       accepted := make(chan net.Conn)
+       go func() {
+               conn, err := listener.Accept()
+               if err == nil {
+                       accepted <- conn
+               }
+       }()
+
+       conn, err := net.Dial("unix", filepath.Join(dir, socketFileName))
+       require.NoError(t, err)
+
+       in, _ := io.Pipe()
+       done := make(chan error)
+       go func() {
+               done <- chatSession(t.Context(), conn, in, io.Discard)
+       }()
+
+       require.NoError(t, (<-accepted).Close())
+       require.ErrorIs(t, <-done, errDaemonClosed)
+       require.NoError(t, listener.Close())
+}
+
+func TestListenSocketReplacesAnOldSocketFile(t *testing.T) {
+       path := filepath.Join(t.TempDir(), socketFileName)
+       require.NoError(t, os.WriteFile(path, nil, 0o600))
+
+       listener, err := listenSocket(path)
+       require.NoError(t, err)
+
+       info, err := os.Stat(path)
+       require.NoError(t, err)
+       require.Equal(t, os.ModeSocket, info.Mode().Type())
+       require.Equal(t, os.FileMode(0o600), info.Mode().Perm())
+       require.NoError(t, listener.Close())
+}
+
+func TestListenSocketRefusesALongPath(t *testing.T) {
+       _, err := listenSocket(filepath.Join(t.TempDir(), strings.Repeat("d", 100), socketFileName))
+       require.ErrorContains(t, err, "set ONECHAT_DIR to a shorter path")
+}
+
+func TestOneProcessHoldsTheTokens(t *testing.T) {
+       dir := t.TempDir()
+       lock, err := lockTokens(dir)
+       require.NoError(t, err)
+
+       _, err = lockTokens(dir)
+       require.ErrorIs(t, err, errTokensInUse)
+       require.NoError(t, lock.Close())
+
+       lock, err = lockTokens(dir)
+       require.NoError(t, err)
+       require.NoError(t, lock.Close())
+}
+
+func TestEventText(t *testing.T) {
+       testCases := []struct {
+               name  string
+               event Event
+               text  string
+       }{
+               {
+                       name:  "reply as it is",
+                       event: Event{Kind: "reply", Text: "Done."},
+                       text:  "Done.",
+               },
+               {
+                       name:  "user message not shown",
+                       event: Event{Kind: "user", Text: "typed"},
+                       text:  "",
+               },
+               {
+                       name:  "long result cut",
+                       event: Event{Kind: "echo", Text: strings.Repeat("line\n", 12)},
+                       text:  "[result] " + strings.Repeat("line\n", 9) + "line\n[2 more lines]",
+               },
+       }
+
+       for _, tc := range testCases {
+               t.Run(tc.name, func(t *testing.T) {
+                       require.Equal(t, tc.text, eventText(tc.event))
+               })
+       }
+}
+
+// testClient is a terminal client of a test daemon, with a pipe as its input.
+type testClient struct {
+       input *io.PipeWriter
+       done  chan error
+       out   *syncBuffer
+}
+
+// Type writes one line to the input of the client.
+func (c *testClient) Type(line string) {
+       _, err := io.WriteString(c.input, line+"\n")
+       if err != nil {
+               panic(err)
+       }
+}
+
+// Output returns what the client showed.
+func (c *testClient) Output() string {
+       return c.out.String()
+}
+
+// WaitFor waits until the client showed text.
+func (c *testClient) WaitFor(t *testing.T, text string) {
+       t.Helper()
+
+       require.Eventually(t, func() bool {
+               return strings.Contains(c.Output(), text)
+       }, 10*time.Second, time.Millisecond, "output so far: %q", c.Output())
+}
+
+// Close ends the input and waits for the client.
+func (c *testClient) Close() error {
+       err := c.input.Close()
+       if err != nil {
+               return err
+       }
+
+       return <-c.done
+}
+
+// syncBuffer is a buffer that the test and the client use at once.
+type syncBuffer struct {
+       mutex  sync.Mutex
+       buffer bytes.Buffer
+}
+
+func (b *syncBuffer) Write(p []byte) (int, error) {
+       b.mutex.Lock()
+       defer b.mutex.Unlock()
+
+       return b.buffer.Write(p)
+}
+
+func (b *syncBuffer) String() string {
+       b.mutex.Lock()
+       defer b.mutex.Unlock()
+
+       return b.buffer.String()
+}
+
+// startTestDaemon serves a new agent on a socket in a temporary directory and
+// returns the socket path. The commands of the agent run in a temporary
+// directory.
+func startTestDaemon(t *testing.T, server *fakeResponses) string {
+       t.Helper()
+
+       dir := t.TempDir()
+       agent, err := openAgent(filepath.Join(dir, agentDirName))
+       require.NoError(t, err)
+
+       agent.config.Workdir = t.TempDir()
+       clock := newTestClock()
+       memory, err := openMemory(t.Context(), agent.dir, clock.Now, newFakeSummarizer())
+       require.NoError(t, err)
+
+       socket := filepath.Join(dir, socketFileName)
+       listener, err := listenSocket(socket)
+       require.NoError(t, err)
+
+       client := newClient(server.URL, apiKey(testKey), agent.CacheKey(), filepath.Join(agent.dir, "usage.jsonl"))
+       ctx, cancel := context.WithCancel(t.Context())
+       done := make(chan error)
+       go func() {
+               done <- serveAgent(ctx, listener, func(notify func(Event)) (*loop, error) {
+                       return newLoop(agent, memory, client, clock.Now, notify)
+               })
+       }()
+
+       t.Cleanup(func() {
+               cancel()
+               require.NoError(t, <-done)
+               require.NoError(t, memory.Close())
+       })
+
+       return socket
+}
+
+// startTestClient connects a client to the daemon at socket.
+func startTestClient(t *testing.T, socket string) *testClient {
+       t.Helper()
+
+       conn, err := net.Dial("unix", socket)
+       require.NoError(t, err)
+
+       reader, writer := io.Pipe()
+       client := &testClient{
+               input: writer,
+               done:  make(chan error, 1),
+               out:   &syncBuffer{},
+       }
+
+       go func() {
+               client.done <- chatSession(t.Context(), conn, reader, client.out)
+       }()
+
+       t.Cleanup(func() {
+               _ = writer.Close()
+       })
+
+       return client
+}

diff --git a/main.go b/main.go
index a561c3755c8b19860a4665762adaf2b2e9f38fd2..680c58d9a1c3884b5f8bab7bc8ef04d3874ee12c 100644 (file)
--- a/main.go
+++ b/main.go
@@ -33,6 +33,16 @@ func run() error {
                if err != nil {
                        return err
                }
                if err != nil {
                        return err
                }

+       case len(args) == 1 && args[0] == "daemon":
+               err := daemon(ctx, os.Stdout)
+               if err != nil {
+                       return err
+               }
+       case len(args) == 1 && args[0] == "chat":
+               err := chat(ctx, os.Stdin, os.Stdout)
+               if err != nil {
+                       return err
+               }

        case len(args) > 0 && args[0] == "probe":
                err := probe(ctx, args[1:], os.Stdout)
                if err != nil {
        case len(args) > 0 && args[0] == "probe":
                err := probe(ctx, args[1:], os.Stdout)
                if err != nil {

diff --git a/probe.go b/probe.go
index 6e0b81eaa48188e87af08d973864f92cb8ec3034..9d272ec69286596881b34a727fea09fe502683cf 100644 (file)
--- a/probe.go
+++ b/probe.go
@@ -42,8 +42,9 @@ const (
 // reports the calls, the time per call, the calls at once, the cached share,
 // the quota, the retries, and the line sizes. BUILDS, from 1 to maxBuilds,
 // limits the builds that call the model at once in this run only. The core
 // reports the calls, the time per call, the calls at once, the cached share,
 // the quota, the retries, and the line sizes. BUILDS, from 1 to maxBuilds,
 // limits the builds that call the model at once in this run only. The core

-// still queues up to maxBuilds builds.
-func probe(ctx context.Context, args []string, out io.Writer) error {
+// still queues up to maxBuilds builds. The run holds the lock of the token
+// file, so it does not run while the daemon runs.
+func probe(ctx context.Context, args []string, out io.Writer) (err error) {

        if len(args) != 2 && len(args) != 3 {
                return errors.New("usage: onechat probe LOG COUNT [BUILDS]")
        }
        if len(args) != 2 && len(args) != 3 {
                return errors.New("usage: onechat probe LOG COUNT [BUILDS]")
        }


@@ -76,6 +77,15 @@ func probe(ctx context.Context, args []string, out io.Writer) error {
                return err
        }
 
                return err
        }
 

+       tokens, err := lockTokens(dir)
+       if err != nil {
+               return err
+       }
+
+       defer func() {
+               err = errors.Join(err, tokens.Close())
+       }()
+

        address, auth, err := modelEndpoint(config, dir)
        if err != nil {
                return err
        address, auth, err := modelEndpoint(config, dir)
        if err != nil {
                return err

diff --git a/turn.go b/turn.go
index 75cbfb6478a4934e6e5172009fcbad718f22ef5b..f8faad83d0a8be375b4e9fd3b2f9e6766284d2cf 100644 (file)
--- a/turn.go
+++ b/turn.go
@@ -17,8 +17,8 @@ type modelClient interface {
 }
 
 // Event tells the clients what the agent does as it happens. Kind is the kind
 }
 
 // Event tells the clients what the agent does as it happens. Kind is the kind

-// of a logged message (user, reply, tool, echo), or one of busy, idle, and
-// error. ID is the ID of a logged message.
+// of a logged message (user, reply, tool, echo), or one of ready, busy, idle,
+// and error. ID is the ID of a logged message.

 type Event struct {
        Kind string `json:"kind"`
        ID   int    `json:"id,omitempty"`
 type Event struct {
        Kind string `json:"kind"`
        ID   int    `json:"id,omitempty"`


@@ -133,6 +133,14 @@ func (l *loop) Interrupt() bool {
        return true
 }
 
        return true
 }
 

+// Busy tells if a turn runs.
+func (l *loop) Busy() bool {
+       l.mutex.Lock()
+       defer l.mutex.Unlock()
+
+       return l.cancel != nil
+}
+

 // Append logs a message and tells the clients. The toolbox logs its calls and
 // results through it.
 func (l *loop) Append(kind, text string) (Message, error) {
 // Append logs a message and tells the clients. The toolbox logs its calls and
 // results through it.
 func (l *loop) Append(kind, text string) (Message, error) {