Name and order the daemon and socket code

Every package-level name that only the daemon uses starts with daemon:
its hub of socket clients is DaemonHub, a connected client is
DaemonClient, and the loop that serves the agent is daemonServe. So a
client of the socket does not read as the model client or the Bot API
client.

What the daemon and the terminal client share is in socket.go: the
socket file, the size limits, the request type SocketRequest, and the
check of the socket path. The file locks that the log, the daemon, the
login, and the probe share are in lock.go, with their test in
lock_test.go. The declarations of these files follow the order of the
go-code skill. The behavior does not change.

Issue: 947bddb

diff --git a/daemon.go b/daemon.go
index cedb45e236dd4e856b4567ff40536392cdde5d07..9d8402ad024de807b70ddfdfabbe4531d864e5a7 100644 (file)
--- a/daemon.go
+++ b/daemon.go
@@ -16,41 +16,15 @@ import (
 )
 
 const (
 )
 
 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.
        // 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
 // 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


@@ -93,7 +67,7 @@ func daemon(ctx context.Context, out io.Writer) (err error) {
                return err
        }
 
                return err
        }
 

-       agent, err := openAgent(filepath.Join(dir, agentDirName))
+       agent, err := openAgent(filepath.Join(dir, daemonAgentDirName))

        if err != nil {
                return err
        }
        if err != nil {
                return err
        }


@@ -108,7 +82,7 @@ func daemon(ctx context.Context, out io.Writer) (err error) {
                err = errors.Join(err, memory.Close())
        }()
 
                err = errors.Join(err, memory.Close())
        }()
 

-       listener, err := listenSocket(socket)
+       listener, err := daemonListenSocket(socket)

        if err != nil {
                return err
        }
        if err != nil {
                return err
        }


@@ -127,95 +101,31 @@ func daemon(ctx context.Context, out io.Writer) (err error) {
                }
        }
 
                }
        }
 

-       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)
        })
 }
 
                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
        mutex   sync.Mutex

-       clients map[*hubClient]bool
+       clients map[*DaemonClient]bool

        workers sync.WaitGroup
 }
 
        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.
 // 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
                }
 
        for {
                conn, err := listener.Accept()
                if err != nil {
                        return
                }
 

-               client := &hubClient{
+               client := &DaemonClient{

                        conn:   conn,
                        conn:   conn,

-                       events: make(chan Event, clientEvents),
+                       events: make(chan Event, daemonClientEvents),

                }
 
                h.mutex.Lock()
                }
 
                h.mutex.Lock()


@@ -240,7 +150,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.
 }
 
 // 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)
        defer h.workers.Done()
 
        encoder := json.NewEncoder(client.conn)


@@ -253,17 +163,17 @@ func (h *hub) write(client *hubClient) {
 }
 
 // read runs the requests of client until it closes its connection.
 }
 
 // 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)
        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() {
        for ctx.Err() == nil && scanner.Scan() {

-               var request clientRequest
+               var request SocketRequest

                err := json.Unmarshal(scanner.Bytes(), &request)
                if err == nil {
                err := json.Unmarshal(scanner.Bytes(), &request)
                if err == nil {

-                       err = runRequest(l, request)
+                       err = daemonRunRequest(l, request)

                }
 
                if err != nil {
                }
 
                if err != nil {


@@ -277,15 +187,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.
 
 // 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)
 }
 
        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.
 // 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
        }
        if !h.clients[client] {
                return
        }


@@ -297,27 +215,9 @@ 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.
 // 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)
        delete(h.clients, client)
        client.once.Do(func() {
                close(client.events)


@@ -325,8 +225,18 @@ func (h *hub) remove(client *hubClient) {
        })
 }
 
        })
 }
 

+// 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.
 // 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.Lock()
        for client := range h.clients {
                h.remove(client)


@@ -336,32 +246,18 @@ func (h *hub) CloseAll() {
        h.workers.Wait()
 }
 
        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
        err := checkSocketPath(path)
        if err != nil {
                return nil, err


@@ -385,23 +281,80 @@ func listenSocket(path string) (net.Listener, error) {
        return listener, nil
 }
 
        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)
+       }

 }
 }

diff --git a/daemon_test.go b/daemon_test.go
index 3d84faaffc7e03117ae2878beafe16cb7a536d70..e27c6ae52e27260f5df00f8bee89215132762623 100644 (file)
--- a/daemon_test.go
+++ b/daemon_test.go
@@ -23,7 +23,7 @@ func TestClientsSeeTheTurnAsItHappens(t *testing.T) {
                modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("It said hi.")}, 100, 0, 5))
        }
 
                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]")
        first := terminalStartTestClient(t, socket)
        second := terminalStartTestClient(t, socket)
        first.WaitFor(t, "[connected to OneChat]")


@@ -54,7 +54,7 @@ func TestStopInterruptsTheTurn(t *testing.T) {
                <-release
        }
 
                <-release
        }
 

-       socket := startTestDaemon(t, server)
+       socket := daemonStartTest(t, server)

        client := terminalStartTestClient(t, socket)
        client.Type("a long task")
        <-arrived
        client := terminalStartTestClient(t, socket)
        client.Type("a long task")
        <-arrived


@@ -70,7 +70,7 @@ func TestListenSocketReplacesAnOldSocketFile(t *testing.T) {
        path := filepath.Join(t.TempDir(), socketFileName)
        require.NoError(t, os.WriteFile(path, nil, 0o600))
 
        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)
        require.NoError(t, err)
 
        info, err := os.Stat(path)


@@ -81,7 +81,7 @@ func TestListenSocketReplacesAnOldSocketFile(t *testing.T) {
 }
 
 func TestListenSocketRefusesALongPath(t *testing.T) {
 }
 
 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.ErrorContains(t, err, "set ONECHAT_DIR to a shorter path")
 }
 


@@ -97,35 +97,21 @@ func TestDaemonWritesNoFileForALongSocketPath(t *testing.T) {
        require.Empty(t, entries)
 }
 
        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.
 // 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()
 
        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()
        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)
 
        agent.config.Workdir = t.TempDir()


@@ -134,14 +120,14 @@ func startTestDaemonWithBot(t *testing.T, server *ModelFake, bot *TelegramBot) s
        require.NoError(t, err)
 
        socket := filepath.Join(dir, socketFileName)
        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() {
        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)
                })
        }()
                        return newLoop(agent, memory, client, clock.Now, notify)
                })
        }()

diff --git a/lock.go b/lock.go
new file mode 100644 (file)
index 0000000..56b6fc6
--- /dev/null
+++ b/lock.go
@@ -0,0 +1,50 @@
+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
+}

diff --git a/lock_test.go b/lock_test.go
new file mode 100644 (file)
index 0000000..5308dc3
--- /dev/null
+++ b/lock_test.go
@@ -0,0 +1,21 @@
+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())
+}

diff --git a/log.go b/log.go
index dfb6e7098ab1df2048da5c96604f37ef6e591345..c3d7190be9d6fa85467bc5a06cf1c1d7cbd4019b 100644 (file)
--- a/log.go
+++ b/log.go
@@ -7,14 +7,10 @@ import (
        "fmt"
        "os"
        "path/filepath"
        "fmt"
        "os"
        "path/filepath"

-       "syscall"

        "time"
        "unicode/utf8"
 )
 
        "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,
 // kinds holds the kinds that a message can have.
 var kinds = map[string]bool{
        "user":  true,


@@ -250,23 +246,3 @@ func validateMessage(message Message) error {
 
        return nil
 }
 
        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
-}

diff --git a/socket.go b/socket.go
new file mode 100644 (file)
index 0000000..a454728
--- /dev/null
+++ b/socket.go
@@ -0,0 +1,37 @@
+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"`
+}

diff --git a/telegram_test.go b/telegram_test.go
index a1a1ed24bc42d842a1327e778e48bfde5a90f3a3..44c9c8fe16de2bb285b58c976947a0be583e8a43 100644 (file)
--- a/telegram_test.go
+++ b/telegram_test.go
@@ -21,7 +21,7 @@ func TestOwnerTalksToTheAgentThroughTelegram(t *testing.T) {
 
        telegram := newTelegramFake(t)
        bot, _ := telegramStartTestBot(t, telegram)
 
        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]")
 
        terminal := terminalStartTestClient(t, socket)
        terminal.WaitFor(t, "[connected to OneChat]")
 


@@ -43,7 +43,7 @@ func TestOwnerTalksToTheAgentThroughTelegram(t *testing.T) {
 func TestBotIgnoresEveryoneButTheOwner(t *testing.T) {
        telegram := newTelegramFake(t)
        bot, logs := telegramStartTestBot(t, telegram)
 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]")
 
        terminal := terminalStartTestClient(t, socket)
        terminal.WaitFor(t, "[connected to OneChat]")
 


@@ -81,7 +81,7 @@ func TestBotIgnoresEveryoneButTheOwner(t *testing.T) {
 func TestBotAnswersThatOnlyTextIsSupported(t *testing.T) {
        telegram := newTelegramFake(t)
        bot, _ := telegramStartTestBot(t, telegram)
 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.Add(TelegramUpdate{
                UpdateID: 1,


@@ -106,7 +106,7 @@ func TestStopFromTelegramInterruptsTheTurn(t *testing.T) {
 
        telegram := newTelegramFake(t)
        bot, _ := telegramStartTestBot(t, telegram)
 
        telegram := newTelegramFake(t)
        bot, _ := telegramStartTestBot(t, telegram)

-       startTestDaemonWithBot(t, server, bot)
+       daemonStartTestWithBot(t, server, bot)

 
        telegram.Add(TelegramUpdate{
                UpdateID: 1,
 
        telegram.Add(TelegramUpdate{
                UpdateID: 1,


@@ -149,7 +149,7 @@ func TestBotStartsAtTheStoredOffset(t *testing.T) {
                Message:  telegramOwnerMessage("new after the restart"),
        })
 
                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")
        terminal := terminalStartTestClient(t, socket)
        terminal.WaitFor(t, "[connected to OneChat]")
        telegramWaitForSent(t, telegram, "ok")


@@ -189,7 +189,7 @@ func TestBotTriesAgainAfterFailures(t *testing.T) {
        }
 
        bot, logs := telegramStartTestBot(t, telegram)
        }
 
        bot, logs := telegramStartTestBot(t, telegram)

-       startTestDaemonWithBot(t, server, bot)
+       daemonStartTestWithBot(t, server, bot)

 
        telegram.Add(TelegramUpdate{
                UpdateID: 1,
 
        telegram.Add(TelegramUpdate{
                UpdateID: 1,


@@ -236,7 +236,7 @@ func TestRepliesGoAsRichMessages(t *testing.T) {
 
        telegram := newTelegramFake(t)
        bot, _ := telegramStartTestBot(t, telegram)
 
        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]")
 
        terminal := terminalStartTestClient(t, socket)
        terminal.WaitFor(t, "[connected to OneChat]")
 


@@ -423,7 +423,7 @@ func TestBotLogsNoTokenWhenTheNetworkFails(t *testing.T) {
        telegram.Close()
 
        bot, logs := telegramStartTestBot(t, telegram)
        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
 
        require.Eventually(t, func() bool {
                return strings.Count(logs.String(), "no valid answer") >= 2

diff --git a/terminal.go b/terminal.go
index 2906204477b14baa33600fd3efb743865cc85037..c785a969aef149ffd823e970d2aa1d04334d8919 100644 (file)
--- a/terminal.go
+++ b/terminal.go
@@ -76,10 +76,10 @@ func terminalSession(ctx context.Context, conn net.Conn, in io.Reader, out io.Wr
 func terminalSendLines(ctx context.Context, conn net.Conn, in io.Reader) error {
        encoder := json.NewEncoder(conn)
        scanner := bufio.NewScanner(in)
 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()
        for ctx.Err() == nil && scanner.Scan() {
                line := scanner.Text()

-               request := clientRequest{
+               request := SocketRequest{

                        Type: "send",
                        Text: line,
                }
                        Type: "send",
                        Text: line,
                }


@@ -88,7 +88,7 @@ func terminalSendLines(ctx context.Context, conn net.Conn, in io.Reader) error {
                case "":
                        continue
                case stopCommand:
                case "":
                        continue
                case stopCommand:

-                       request = clientRequest{
+                       request = SocketRequest{

                                Type: "interrupt",
                        }
                }
                                Type: "interrupt",
                        }
                }


@@ -106,7 +106,7 @@ func terminalSendLines(ctx context.Context, conn net.Conn, in io.Reader) error {
 // connection closes.
 func terminalShowEvents(conn net.Conn, out io.Writer) error {
        scanner := bufio.NewScanner(conn)
 // 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)
        for scanner.Scan() {
                var event Event
                err := json.Unmarshal(scanner.Bytes(), &event)

diff --git a/terminal_test.go b/terminal_test.go
index 9df3e0a99902382e0d59ebc24e8c2393b961b542..d86f88aaba229fea6f216a4e59623d4440cf67ef 100644 (file)
--- a/terminal_test.go
+++ b/terminal_test.go
@@ -13,7 +13,7 @@ import (
 
 func TestClientEndsWhenTheDaemonEnds(t *testing.T) {
        dir := t.TempDir()
 
 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)
        require.NoError(t, err)
 
        accepted := make(chan net.Conn)