Run the tools of an agent and log each call

[onechat] / memory.go

diff --git a/memory.go b/memory.go
index 733673bf8607b50019e8417a80d4d0d9c6352151..792fe21f42b2e69dbb85b1e114320124d5f10fa0 100644 (file)
--- a/memory.go
+++ b/memory.go
@@ -39,6 +39,10 @@ type Task struct {
 // start. It never builds the views again from the log, because a new view
 // differs from the live one and so loses the prompt cache.
 //
+// Each write goes to the operating system at once, and a syncer makes the
+// writes durable by group commit in the background. Flush waits until the
+// writes so far are durable.
+//
 // The summarizer builds the nodes in the queue in the background. The start
 // reads the indexes once to fill the queue again. After that, only a new
 // message or a new node adds work. Memory keeps the nodes of the views and the
@@ -48,6 +52,7 @@ type Memory struct {
        mutex      sync.Mutex
        root       string
        log        *Log
+       syncer     *syncer
        tree       *Tree
        chat       *View
        cview      *View
@@ -84,6 +89,11 @@ func openMemory(ctx context.Context, root string, clock func() time.Time, summar
 
        m.ctx, m.cancel = context.WithCancel(ctx)
 
+       m.syncer, err = openSyncer(root)
+       if err != nil {
+               return nil, errors.Join(err, m.Close())
+       }
+
        m.tree, err = openTree(ctx, root, log.Len(), clock)
        if err != nil {
                return nil, errors.Join(err, m.Close())
@@ -105,6 +115,7 @@ func openMemory(ctx context.Context, root string, clock func() time.Time, summar
        }
 
        m.mutex.Lock()
+       m.syncer.Written()
        m.dispatch()
        m.mutex.Unlock()
 
@@ -227,8 +238,8 @@ func (m *Memory) catchUp() error {
        return m.save()
 }
 
-// Close stops the builds, waits for them, closes the files of the chat, and
-// releases its lock.
+// Close stops the builds, waits for them, makes the writes durable, closes the
+// files of the chat, and releases its lock.
 func (m *Memory) Close() error {
        m.cancel()
        m.workers.Wait()
@@ -237,8 +248,12 @@ func (m *Memory) Close() error {
        defer m.mutex.Unlock()
 
        var err error
+       if m.syncer != nil {
+               err = m.syncer.Close()
+       }
+
        if m.tree != nil {
-               err = m.tree.Close()
+               err = errors.Join(err, m.tree.Close())
        }
 
        return errors.Join(err, m.log.Close())
@@ -255,6 +270,8 @@ func (m *Memory) Append(kind, text string) (Message, error) {
                return Message{}, err
        }
 
+       defer m.syncer.Written()
+
        err = m.addMessage(message)
        if err != nil {
                return Message{}, err
@@ -272,15 +289,24 @@ func (m *Memory) Append(kind, text string) (Message, error) {
 }
 
 // addMessage builds the leaf of message when it needs no model call, or else
-// adds the leaf to the queue. Then it appends the leaf to the views.
+// adds the leaf to the queue. Then it appends the leaf to the views. A crash can
+// stop an append after the leaf build and before the view save, so a leaf that
+// is built already stays as it is, and restoreQueue settles its parent.
 func (m *Memory) addMessage(message Message) error {
+       built, err := m.tree.Built(0, message.ID)
+       if err != nil {
+               return err
+       }
+
        leaf, free := freeLeaf(message)
-       if free {
-               err := m.build(leaf)
+       switch {
+       case built:
+       case free:
+               err = m.build(leaf)
                if err != nil {
                        return err
                }
-       } else {
+       default:
                m.queue.AddLeaf(message.ID)
        }
 
@@ -448,6 +474,7 @@ func (m *Memory) run(key nodeKey, task Task) {
                        Text:  text,
                        Size:  len(text),
                })
+               m.syncer.Written()
        }
 
        if err != nil {
@@ -481,6 +508,11 @@ func (m *Memory) Wait(ctx context.Context) error {
        }
 }
 
+// Flush returns when every write before the call is durable.
+func (m *Memory) Flush(ctx context.Context) error {
+       return m.syncer.Flush(ctx)
+}
+
 // save writes the views to view.json.
 func (m *Memory) save() error {
        data, err := json.Marshal(viewFile{
@@ -511,6 +543,30 @@ func (m *Memory) View() string {
        return m.chat.Render()
 }
 
+// Len returns the number of messages in the log.
+func (m *Memory) Len() int {
+       m.mutex.Lock()
+       defer m.mutex.Unlock()
+
+       return m.log.Len()
+}
+
+// Message reads the message with the given ID.
+func (m *Memory) Message(id int) (Message, error) {
+       m.mutex.Lock()
+       defer m.mutex.Unlock()
+
+       return m.log.Message(id)
+}
+
+// Node reads the node (l, i) if it is built.
+func (m *Memory) Node(level, index int) (Node, bool, error) {
+       m.mutex.Lock()
+       defer m.mutex.Unlock()
+
+       return m.tree.Node(level, index)
+}
+
 // Context returns the compaction view for the build of node (l, i). It ends
 // before the message of a leaf, or after the last message of a merge, and it
 // stops at the first line that is not built.
@@ -529,25 +585,3 @@ func (m *Memory) context(level, index int) string {
 
        return m.cview.Context(end)
 }
-
-// writeFileAtomic replaces the file at path with data through a temporary file
-// and a rename. A reader or a crash of the process sees the old file or the new
-// file, never a part of one.
-func writeFileAtomic(path string, data []byte) error {
-       file, err := os.CreateTemp(filepath.Dir(path), "."+filepath.Base(path)+".*")
-       if err != nil {
-               return fmt.Errorf("create temporary file: %w", err)
-       }
-
-       _, err = file.Write(data)
-       err = errors.Join(err, file.Close())
-       if err == nil {
-               err = os.Rename(file.Name(), path)
-       }
-
-       if err != nil {
-               return errors.Join(fmt.Errorf("replace file: %w", err), os.Remove(file.Name()))
-       }
-
-       return nil
-}