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
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
*******
}
// 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
}
+ 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)
}
--- /dev/null
+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)
+}
--- /dev/null
+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
+}
--- /dev/null
+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
+}
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 {
// 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]")
}
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
}
// 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"`
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) {
projects / onechat / commitdiff