)
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
+ daemonAgentDirName = "agent"
- // 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
-
- // 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
return err
}
- agent, err := openAgent(filepath.Join(dir, agentDirName))
+ agent, err := openAgent(filepath.Join(dir, daemonAgentDirName))
if err != nil {
return err
}
err = errors.Join(err, memory.Close())
}()
- listener, err := listenSocket(socket)
+ listener, err := daemonListenSocket(socket)
if err != nil {
return err
}
}
}
- return serveAgent(ctx, listener, telegram, func(notify func(Event)) (*loop, error) {
+ return daemonServe(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)
- }
- }
-
- l, err := start(notify)
- 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)
- }()
-
- 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()
-
- return errors.Join(err, botErr)
-}
-
-// 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.
+type DaemonHub struct {
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 *loop) {
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()
}
// 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)
}
// 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 *loop) {
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, request)
}
if err != nil {
// 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
}
}
}
-// 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)
})
}
+// 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.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
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. 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, start func(notify func(Event)) (*loop, error)) error {
+ ctx, cancel := context.WithCancel(ctx)
+ defer cancel()
+
+ hub := &DaemonHub{
+ clients: make(map[*DaemonClient]bool),
}
- return nil
-}
+ notify := hub.Broadcast
+ if bot != nil {
+ notify = func(event Event) {
+ hub.Broadcast(event)
+ bot.Notify(event)
+ }
+ }
-// 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
+ l, err := start(notify)
+ if err != nil {
+ return errors.Join(err, listener.Close())
}
- return lock, err
+ var workers sync.WaitGroup
+ workers.Add(1)
+ go func() {
+ defer workers.Done()
+
+ 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()
+
+ return errors.Join(err, botErr)
+}
+
+// daemonRunRequest runs one request of a client.
+func daemonRunRequest(l *loop, request SocketRequest) 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)
+ }
}
modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("It said hi.")}, 100, 0, 5))
}
- socket := startTestDaemon(t, server)
+ socket := daemonStartTest(t, server)
first := terminalStartTestClient(t, socket)
second := terminalStartTestClient(t, socket)
first.WaitFor(t, "[connected to OneChat]")
<-release
}
- socket := startTestDaemon(t, server)
+ socket := daemonStartTest(t, server)
client := terminalStartTestClient(t, socket)
client.Type("a long task")
<-arrived
path := filepath.Join(t.TempDir(), socketFileName)
require.NoError(t, os.WriteFile(path, nil, 0o600))
- listener, err := listenSocket(path)
+ listener, err := daemonListenSocket(path)
require.NoError(t, err)
info, err := os.Stat(path)
}
func TestListenSocketRefusesALongPath(t *testing.T) {
- _, err := listenSocket(filepath.Join(t.TempDir(), strings.Repeat("d", 100), socketFileName))
+ _, err := daemonListenSocket(filepath.Join(t.TempDir(), strings.Repeat("d", 100), socketFileName))
require.ErrorContains(t, err, "set ONECHAT_DIR to a shorter path")
}
require.Empty(t, entries)
}
-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())
-}
-
-// startTestDaemon serves a new agent on a socket in a temporary directory and
+// daemonStartTest 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 *ModelFake) string {
+func daemonStartTest(t *testing.T, server *ModelFake) string {
t.Helper()
- return startTestDaemonWithBot(t, server, nil)
+ return daemonStartTestWithBot(t, server, nil)
}
-// startTestDaemonWithBot is startTestDaemon with bot, which can be nil.
-func startTestDaemonWithBot(t *testing.T, server *ModelFake, bot *TelegramBot) string {
+// daemonStartTestWithBot is daemonStartTest with bot, which can be nil.
+func daemonStartTestWithBot(t *testing.T, server *ModelFake, bot *TelegramBot) string {
t.Helper()
dir := t.TempDir()
- agent, err := openAgent(filepath.Join(dir, agentDirName))
+ agent, err := openAgent(filepath.Join(dir, daemonAgentDirName))
require.NoError(t, err)
agent.config.Workdir = t.TempDir()
require.NoError(t, err)
socket := filepath.Join(dir, socketFileName)
- listener, err := listenSocket(socket)
+ listener, err := daemonListenSocket(socket)
require.NoError(t, err)
client := newModelClient(server.URL, ModelAPIKey(modelTestKey), agent.CacheKey(), filepath.Join(agent.dir, "usage.jsonl"))
ctx, cancel := context.WithCancel(t.Context())
done := make(chan error)
go func() {
- done <- serveAgent(ctx, listener, bot, func(notify func(Event)) (*loop, error) {
+ done <- daemonServe(ctx, listener, bot, func(notify func(Event)) (*loop, error) {
return newLoop(agent, memory, client, clock.Now, notify)
})
}()
--- /dev/null
+package main
+
+import (
+ "errors"
+ "fmt"
+ "os"
+ "path/filepath"
+ "syscall"
+)
+
+// tokenLockFileName is the lock file of the token file in the data directory.
+const tokenLockFileName = "auth.lock"
+
+// 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")
+
+// errLocked tells that another process owns the chat.
+var errLocked = errors.New("another process owns the chat")
+
+// 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
+}
+
+// lockFile opens the file at path and takes an exclusive lock on it. The lock
+// stays until the file closes or the process ends.
+func lockFile(path string) (*os.File, error) {
+ file, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o600)
+ if err != nil {
+ return nil, fmt.Errorf("open lock file: %w", err)
+ }
+
+ err = syscall.Flock(int(file.Fd()), syscall.LOCK_EX|syscall.LOCK_NB)
+ if errors.Is(err, syscall.EWOULDBLOCK) {
+ err = errLocked
+ }
+
+ if err != nil {
+ return nil, errors.Join(fmt.Errorf("lock chat: %w", err), file.Close())
+ }
+
+ return file, nil
+}
--- /dev/null
+package main
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/require"
+)
+
+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())
+}
"fmt"
"os"
"path/filepath"
- "syscall"
"time"
"unicode/utf8"
)
-// errLocked tells that another process owns the chat.
-var errLocked = errors.New("another process owns the chat")
-
// kinds holds the kinds that a message can have.
var kinds = map[string]bool{
"user": true,
return nil
}
-
-// lockFile opens the file at path and takes an exclusive lock on it. The lock
-// stays until the file closes or the process ends.
-func lockFile(path string) (*os.File, error) {
- file, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o600)
- if err != nil {
- return nil, fmt.Errorf("open lock file: %w", err)
- }
-
- err = syscall.Flock(int(file.Fd()), syscall.LOCK_EX|syscall.LOCK_NB)
- if errors.Is(err, syscall.EWOULDBLOCK) {
- err = errLocked
- }
-
- if err != nil {
- return nil, errors.Join(fmt.Errorf("lock chat: %w", err), file.Close())
- }
-
- return file, nil
-}
--- /dev/null
+package main
+
+import (
+ "fmt"
+)
+
+const (
+ // socketFileName is the Unix socket of the daemon in the data
+ // directory.
+ socketFileName = "onechat.sock"
+
+ // socketRequestBytes is the most bytes of one request line of a client.
+ socketRequestBytes = 16 << 20
+
+ // 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"
+)
+
+// 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
+}
+
+// SocketRequest is one line from a client: "send" with the text of a user
+// message, or "interrupt".
+type SocketRequest struct {
+ Type string `json:"type"`
+ Text string `json:"text,omitempty"`
+}
telegram := newTelegramFake(t)
bot, _ := telegramStartTestBot(t, telegram)
- socket := startTestDaemonWithBot(t, server, bot)
+ socket := daemonStartTestWithBot(t, server, bot)
terminal := terminalStartTestClient(t, socket)
terminal.WaitFor(t, "[connected to OneChat]")
func TestBotIgnoresEveryoneButTheOwner(t *testing.T) {
telegram := newTelegramFake(t)
bot, logs := telegramStartTestBot(t, telegram)
- socket := startTestDaemonWithBot(t, newModelFake(t), bot)
+ socket := daemonStartTestWithBot(t, newModelFake(t), bot)
terminal := terminalStartTestClient(t, socket)
terminal.WaitFor(t, "[connected to OneChat]")
func TestBotAnswersThatOnlyTextIsSupported(t *testing.T) {
telegram := newTelegramFake(t)
bot, _ := telegramStartTestBot(t, telegram)
- startTestDaemonWithBot(t, newModelFake(t), bot)
+ daemonStartTestWithBot(t, newModelFake(t), bot)
telegram.Add(TelegramUpdate{
UpdateID: 1,
telegram := newTelegramFake(t)
bot, _ := telegramStartTestBot(t, telegram)
- startTestDaemonWithBot(t, server, bot)
+ daemonStartTestWithBot(t, server, bot)
telegram.Add(TelegramUpdate{
UpdateID: 1,
Message: telegramOwnerMessage("new after the restart"),
})
- socket := startTestDaemonWithBot(t, server, bot)
+ socket := daemonStartTestWithBot(t, server, bot)
terminal := terminalStartTestClient(t, socket)
terminal.WaitFor(t, "[connected to OneChat]")
telegramWaitForSent(t, telegram, "ok")
}
bot, logs := telegramStartTestBot(t, telegram)
- startTestDaemonWithBot(t, server, bot)
+ daemonStartTestWithBot(t, server, bot)
telegram.Add(TelegramUpdate{
UpdateID: 1,
telegram := newTelegramFake(t)
bot, _ := telegramStartTestBot(t, telegram)
- socket := startTestDaemonWithBot(t, server, bot)
+ socket := daemonStartTestWithBot(t, server, bot)
terminal := terminalStartTestClient(t, socket)
terminal.WaitFor(t, "[connected to OneChat]")
telegram.Close()
bot, logs := telegramStartTestBot(t, telegram)
- startTestDaemonWithBot(t, newModelFake(t), bot)
+ daemonStartTestWithBot(t, newModelFake(t), bot)
require.Eventually(t, func() bool {
return strings.Count(logs.String(), "no valid answer") >= 2
func terminalSendLines(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)
+ scanner.Buffer(make([]byte, 64*1024), socketRequestBytes)
for ctx.Err() == nil && scanner.Scan() {
line := scanner.Text()
- request := clientRequest{
+ request := SocketRequest{
Type: "send",
Text: line,
}
case "":
continue
case stopCommand:
- request = clientRequest{
+ request = SocketRequest{
Type: "interrupt",
}
}
// connection closes.
func terminalShowEvents(conn net.Conn, out io.Writer) error {
scanner := bufio.NewScanner(conn)
- scanner.Buffer(make([]byte, 64*1024), requestBytes)
+ scanner.Buffer(make([]byte, 64*1024), socketRequestBytes)
for scanner.Scan() {
var event Event
err := json.Unmarshal(scanner.Bytes(), &event)
func TestClientEndsWhenTheDaemonEnds(t *testing.T) {
dir := t.TempDir()
- listener, err := listenSocket(filepath.Join(dir, socketFileName))
+ listener, err := daemonListenSocket(filepath.Join(dir, socketFileName))
require.NoError(t, err)
accepted := make(chan net.Conn)
projects / onechat / commitdiff