Keep each schedule with its owning agent

[onechat] / memory.go

diff --git a/memory.go b/memory.go
index 9846c9817458f77cc0c92febb0a037aa4ff86dfa..a240b6c16d1d898e3051f3e0a0c7a19434061b3e 100644 (file)
--- a/memory.go
+++ b/memory.go
@@ -5,6 +5,7 @@ import (
        "encoding/json"
        "errors"
        "fmt"
        "encoding/json"
        "errors"
        "fmt"

+       "log/slog"

        "os"
        "path/filepath"
        "strings"
        "os"
        "path/filepath"
        "strings"


@@ -12,13 +13,19 @@ import (
        "time"
 )
 
        "time"
 )
 

-// errMemoryNoView tells that a chat has messages but no saved view.
-var errMemoryNoView = errors.New("the chat has messages but no saved view")
+var (
+       // errMemoryNoView tells that a chat has messages but no saved view.
+       errMemoryNoView = errors.New("the chat has messages but no saved view")
+
+       // errMemoryUnsettled tells that a line of the view stays pending,
+       // because its build failed.
+       errMemoryUnsettled = errors.New("a summary of the view failed, and it is tried again at the next message")
+)

 
 // openMemory opens the chat in root and starts the builds that are ready. The
 
 // openMemory opens the chat in root and starts the builds that are ready. The

-// clock gives the dates of new messages and the days of new nodes. The builds
-// stop when ctx ends or at Close.
-func openMemory(ctx context.Context, root string, clock func() time.Time, summarizer Summarizer) (*Memory, error) {
+// clock gives the dates of new messages and the days of new nodes. logger gets
+// one line for each failed build. The builds stop when ctx ends or at Close.
+func openMemory(ctx context.Context, root string, clock func() time.Time, summarizer Summarizer, logger *slog.Logger) (*Memory, error) {

        log, err := openLog(ctx, root, clock)
        if err != nil {
                return nil, err
        log, err := openLog(ctx, root, clock)
        if err != nil {
                return nil, err


@@ -29,6 +36,7 @@ func openMemory(ctx context.Context, root string, clock func() time.Time, summar
                log:        log,
                queue:      newBuildQueue(),
                summarizer: summarizer,
                log:        log,
                queue:      newBuildQueue(),
                summarizer: summarizer,

+               logger:     logger,

                changed:    make(chan struct{}),
        }
 
                changed:    make(chan struct{}),
        }
 


@@ -100,6 +108,7 @@ type Memory struct {
        cview      *View
        queue      *BuildQueue
        summarizer Summarizer
        cview      *View
        queue      *BuildQueue
        summarizer Summarizer

+       logger     *slog.Logger

        ctx        context.Context
        cancel     context.CancelFunc
        workers    sync.WaitGroup
        ctx        context.Context
        cancel     context.CancelFunc
        workers    sync.WaitGroup


@@ -116,6 +125,7 @@ func (m *Memory) Close() error {
        defer m.mutex.Unlock()
 
        var err error
        defer m.mutex.Unlock()
 
        var err error

+

        if m.syncer != nil {
                err = m.syncer.Close()
        }
        if m.syncer != nil {
                err = m.syncer.Close()
        }


@@ -130,24 +140,35 @@ func (m *Memory) Close() error {
 // Append logs a new message, adds it to the tree and the views, and starts the
 // builds that are ready, also the failed ones.
 func (m *Memory) Append(kind, text string) (Message, error) {
 // Append logs a new message, adds it to the tree and the views, and starts the
 // builds that are ready, also the failed ones.
 func (m *Memory) Append(kind, text string) (Message, error) {

+       return m.AppendDated(kind, text, "")
+}
+
+// AppendDated appends a message with its RFC 3339 date and builds its nodes. An
+// empty date uses the clock, as Append does.
+func (m *Memory) AppendDated(kind, text, date string) (Message, error) {

        m.mutex.Lock()
        defer m.mutex.Unlock()
 
        m.mutex.Lock()
        defer m.mutex.Unlock()
 

-       message, err := m.log.Append(kind, text)
+       return m.appendDated(kind, text, date)
+}
+
+// appendDated appends under the memory lock and updates the views and builds.
+func (m *Memory) appendDated(kind, text, date string) (Message, error) {
+       message, err := m.log.AppendDated(kind, text, date)

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

-               return Message{}, err
+               return Message{}, fmt.Errorf("append dated log message: %w", err)

        }
 
        defer m.syncer.Written()
 
        err = m.addMessage(message)
        if err != nil {
        }
 
        defer m.syncer.Written()
 
        err = m.addMessage(message)
        if err != nil {

-               return Message{}, err
+               return Message{}, fmt.Errorf("add dated message to memory: %w", err)

        }
 
        err = m.save()
        if err != nil {
        }
 
        err = m.save()
        if err != nil {

-               return Message{}, err
+               return Message{}, fmt.Errorf("save dated message view: %w", err)

        }
 
        m.queue.Retry()
        }
 
        m.queue.Retry()


@@ -156,9 +177,48 @@ func (m *Memory) Append(kind, text string) (Message, error) {
        return message, nil
 }
 
        return message, nil
 }
 

+// AppendOnce appends a queued message. prepare saves its log ID before the
+// append, under the memory lock. A saved ID permits recovery after a stop
+// between that save and the append, without a second log entry. prepare must
+// not call Memory. The caller stops delivery after a failed append.
+func (m *Memory) AppendOnce(kind, text string, id *int, prepare func(int) error) (Message, error) {
+       m.mutex.Lock()
+       defer m.mutex.Unlock()
+
+       if id != nil && *id < m.log.Len() {
+               message, err := m.log.Message(*id)
+               if err != nil {
+                       return Message{}, fmt.Errorf("read queued message: %w", err)
+               }
+
+               if message.Kind != kind || message.Text != text {
+                       return Message{}, errors.New("queued message does not match its log entry")
+               }
+
+               return message, nil
+       }
+
+       if id != nil && *id != m.log.Len() {
+               return Message{}, errors.New("queued message is beyond the log")
+       }
+
+       err := prepare(m.log.Len())
+       if err != nil {
+               return Message{}, fmt.Errorf("prepare queued message: %w", err)
+       }
+
+       message, err := m.appendDated(kind, text, "")
+       if err != nil {
+               return Message{}, fmt.Errorf("append queued message: %w", err)
+       }
+
+       return message, nil
+}
+

 // children reads the two children of the node key.
 func (m *Memory) children(key NodeKey) (Node, Node, error) {
        var children [2]Node
 // children reads the two children of the node key.
 func (m *Memory) children(key NodeKey) (Node, Node, error) {
        var children [2]Node

+

        for n := range children {
                child, built, err := m.tree.Node(key.level-1, 2*key.index+n)
                if err != nil {
        for n := range children {
                child, built, err := m.tree.Node(key.level-1, 2*key.index+n)
                if err != nil {


@@ -179,6 +239,7 @@ func (m *Memory) children(key NodeKey) (Node, Node, error) {
 func (m *Memory) Wait(ctx context.Context) error {
        for {
                m.mutex.Lock()
 func (m *Memory) Wait(ctx context.Context) error {
        for {
                m.mutex.Lock()

+

                running := m.queue.Running()
                changed := m.changed
                m.mutex.Unlock()
                running := m.queue.Running()
                changed := m.changed
                m.mutex.Unlock()


@@ -195,11 +256,60 @@ func (m *Memory) Wait(ctx context.Context) error {
        }
 }
 
        }
 }
 

+// Settle returns when the chat view has no pending line, so that every message
+// so far is summarized. A pending line with no build that runs belongs to a
+// failed build, which waits for the next message, so Settle returns
+// errMemoryUnsettled in place of a wait that never ends.
+func (m *Memory) Settle(ctx context.Context) error {
+       for {
+               m.mutex.Lock()
+
+               pending := m.chat.Pending()
+               running := m.queue.Running()
+               changed := m.changed
+               m.mutex.Unlock()
+
+               if !pending {
+                       return nil
+               }
+
+               if running == 0 {
+                       return errMemoryUnsettled
+               }
+
+               select {
+               case <-changed:
+               case <-ctx.Done():
+                       return ctx.Err()
+               }
+       }
+}
+

 // Flush returns when every write before the call is durable.
 func (m *Memory) Flush(ctx context.Context) error {
        return m.syncer.Flush(ctx)
 }
 
 // Flush returns when every write before the call is durable.
 func (m *Memory) Flush(ctx context.Context) error {
        return m.syncer.Flush(ctx)
 }
 

+// Snapshot flushes all memory writes and runs snapshot while no append or
+// summary write can change the log, indexes, tree, or views. snapshot must not
+// call Memory.
+func (m *Memory) Snapshot(ctx context.Context, snapshot func() error) error {
+       m.mutex.Lock()
+       defer m.mutex.Unlock()
+
+       err := m.Flush(ctx)
+       if err != nil {
+               return fmt.Errorf("flush memory snapshot: %w", err)
+       }
+
+       err = snapshot()
+       if err != nil {
+               return fmt.Errorf("copy memory snapshot: %w", err)
+       }
+
+       return nil
+}
+

 // ViewLines returns the lines of the chat view, each with its newline, without
 // the leaf lines of the messages in skip.
 func (m *Memory) ViewLines(skip map[int]bool) []string {
 // ViewLines returns the lines of the chat view, each with its newline, without
 // the leaf lines of the messages in skip.
 func (m *Memory) ViewLines(skip map[int]bool) []string {


@@ -252,6 +362,7 @@ func (m *Memory) load() error {
        }
 
        var file MemoryViewFile
        }
 
        var file MemoryViewFile

+

        err = json.Unmarshal(data, &file)
        if err != nil {
                return fmt.Errorf("decode view: %w", err)
        err = json.Unmarshal(data, &file)
        if err != nil {
                return fmt.Errorf("decode view: %w", err)


@@ -299,6 +410,7 @@ func (m *Memory) restoreQueue() error {
                leaf, free := freeLeaf(message)
                if !free {
                        m.queue.AddLeaf(id)
                leaf, free := freeLeaf(message)
                if !free {
                        m.queue.AddLeaf(id)

+

                        continue
                }
 
                        continue
                }
 


@@ -360,11 +472,14 @@ func (m *Memory) dispatch() {
 
                task, err := m.task(key)
                if err != nil {
 
                task, err := m.task(key)
                if err != nil {

+                       m.failed(key, fmt.Errorf("read its input: %w", err))

                        m.queue.Fail(key)
                        m.queue.Fail(key)

+

                        continue
                }
 
                m.workers.Add(1)
                        continue
                }
 
                m.workers.Add(1)

+

                go m.run(key, task)
        }
 }
                go m.run(key, task)
        }
 }


@@ -463,6 +578,7 @@ func (m *Memory) settle(key NodeKey) error {
                node, free := freeMerge(left, right)
                if !free {
                        m.queue.AddMerge(parent)
                node, free := freeMerge(left, right)
                if !free {
                        m.queue.AddMerge(parent)

+

                        return nil
                }
 
                        return nil
                }
 


@@ -527,6 +643,10 @@ func (m *Memory) run(key NodeKey, task Task) {
                m.syncer.Written()
        }
 
                m.syncer.Written()
        }
 

+       if err != nil && m.ctx.Err() == nil {
+               m.failed(key, err)
+       }
+

        if err != nil {
                m.queue.Fail(key)
        } else {
        if err != nil {
                m.queue.Fail(key)
        } else {


@@ -534,10 +654,16 @@ func (m *Memory) run(key NodeKey, task Task) {
        }
 
        close(m.changed)
        }
 
        close(m.changed)

+

        m.changed = make(chan struct{})
        m.dispatch()
 }
 
        m.changed = make(chan struct{})
        m.dispatch()
 }
 

+// failed logs the failed build of the node key with its cause err.
+func (m *Memory) failed(key NodeKey, err error) {
+       m.logger.Warn("summary failed", "level", key.level, "index", key.index, "err", err)
+}
+

 // ready tells if both children of the node key are built and the node is not.
 func (m *Memory) ready(key NodeKey) (bool, error) {
        for _, index := range []int{2 * key.index, 2*key.index + 1} {
 // ready tells if both children of the node key are built and the node is not.
 func (m *Memory) ready(key NodeKey) (bool, error) {
        for _, index := range []int{2 * key.index, 2*key.index + 1} {


@@ -579,6 +705,80 @@ func (m *Memory) Context(level, index int) string {
        return m.context(level, index)
 }
 
        return m.context(level, index)
 }
 

+// rebuildComplete checks the whole tree, not only the visible view. A compactor
+// can fail without an HTTP error, for example with an empty line. Such a failed
+// merge can be absent from the view and must prevent a swap.
+func (m *Memory) rebuildComplete() error {
+       m.mutex.Lock()
+       defer m.mutex.Unlock()
+
+       if len(m.queue.leaves) != 0 || len(m.queue.queued) != 0 {
+               return errors.New("rebuild failed: summary tree has unfinished nodes")
+       }
+
+       return nil
+}
+
+// buildModelLines returns the sizes of the nodes that the model wrote, in tree
+// order. A node that fits its source with no model call is not one of them.
+func (m *Memory) buildModelLines() ([]int, error) {
+       m.mutex.Lock()
+       defer m.mutex.Unlock()
+
+       keys, err := m.tree.Keys()
+       if err != nil {
+               return nil, fmt.Errorf("list built nodes: %w", err)
+       }
+
+       var sizes []int
+
+       for _, key := range keys {
+               var node Node
+
+               node, _, err = m.tree.Node(key.level, key.index)
+               if err != nil {
+                       return nil, fmt.Errorf("read built node: %w", err)
+               }
+
+               var model bool
+
+               model, err = m.buildBuiltByModel(key)
+               if err != nil {
+                       return nil, fmt.Errorf("check model source: %w", err)
+               }
+
+               if model {
+                       sizes = append(sizes, node.Size)
+               }
+       }
+
+       return sizes, nil
+}
+
+// buildBuiltByModel tells if the node key needed a model call: a leaf whose
+// source is over nodeBytes, or a merge whose children do not fit in one node.
+func (m *Memory) buildBuiltByModel(key NodeKey) (bool, error) {
+       if key.level == 0 {
+               message, err := m.log.Message(key.index)
+               if err != nil {
+                       return false, fmt.Errorf("read leaf source: %w", err)
+               }
+
+               _, free := freeLeaf(message)
+
+               return !free, nil
+       }
+
+       left, right, err := m.children(key)
+       if err != nil {
+               return false, fmt.Errorf("read merge sources: %w", err)
+       }
+
+       _, free := freeMerge(left, right)
+
+       return !free, nil
+}
+

 // Task is the build of the node (Level, Index). Context is the compaction view
 // for the build. Input holds the source "kind: text" of a leaf, or the two
 // child lines of a merge with their newlines as spaces.
 // Task is the build of the node (Level, Index). Context is the compaction view
 // for the build. Input holds the source "kind: text" of a leaf, or the two
 // child lines of a merge with their newlines as spaces.