+// dispatch starts the builds that are ready, up to the limit. A build whose
+// input cannot be read fails and waits for the next message.
+func (m *Memory) dispatch() {
+ for m.ctx.Err() == nil {
+ key, ready := m.queue.Next()
+ if !ready {
+ return
+ }
+
+ task, err := m.task(key)
+ if err != nil {
+ m.failed(key, fmt.Errorf("read its input: %w", err))
+ m.queue.Fail(key)
+
+ continue
+ }
+
+ m.workers.Add(1)
+
+ go m.run(key, task)
+ }
+}
+
+func (m *Memory) viewPath() string {
+ return filepath.Join(m.root, "view.json")
+}
+
+// build adds node to the tree, keeps it in the views when it is a pending line,
+// and settles its parent.
+func (m *Memory) build(node Node) error {
+ err := m.tree.Add(node)
+ if err != nil {
+ return fmt.Errorf("add built node: %w", err)
+ }
+
+ m.chat.Refresh(node)
+ m.cview.Refresh(node)
+
+ return m.settle(NodeKey{node.Level, node.Index})
+}
+
+// settle looks at the parent of the built node key when both children are
+// built. It builds each parent that needs no model call, up the tree, and adds
+// the first other parent to the queue.
+func (m *Memory) settle(key NodeKey) error {
+ for {
+ parent := NodeKey{key.level + 1, key.index / 2}
+ ready, err := m.ready(parent)
+ if err != nil || !ready {
+ if err != nil {
+ return fmt.Errorf("check parent readiness: %w", err)
+ }
+
+ return nil
+ }
+
+ left, right, err := m.children(parent)
+ if err != nil {
+ return fmt.Errorf("read parent children: %w", err)
+ }
+
+ node, free := freeMerge(left, right)
+ if !free {
+ m.queue.AddMerge(parent)
+
+ return nil
+ }
+
+ err = m.tree.Add(node)
+ if err != nil {
+ return fmt.Errorf("add merged node: %w", err)
+ }
+
+ key = parent
+ }
+}
+
+// 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. 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 fmt.Errorf("check message leaf: %w", err)
+ }
+
+ leaf, free := freeLeaf(message)
+ switch {
+ case built:
+ case free:
+ err = m.build(leaf)
+ if err != nil {
+ return fmt.Errorf("build message leaf: %w", err)
+ }
+ default:
+ m.queue.AddLeaf(message.ID)
+ }
+
+ merged, err := m.chat.Append(m.tree)
+ if err != nil {
+ return fmt.Errorf("append to chat view: %w", err)
+ }
+
+ _, err = m.cview.Append(m.tree)
+ if err == nil && merged {
+ err = m.cview.Coarsen(m.chat, m.tree)
+ }
+
+ if err != nil {
+ return fmt.Errorf("append to compaction view: %w", err)
+ }
+
+ return nil
+}
+
+// save writes the views to view.json.
+func (m *Memory) save() error {
+ data, err := json.Marshal(MemoryViewFile{
+ Chat: m.chat.State(),
+ Compaction: m.cview.State(),
+ })
+ if err != nil {
+ return fmt.Errorf("encode view: %w", err)
+ }
+
+ err = writeFileAtomic(m.viewPath(), append(data, '\n'))
+ if err != nil {
+ return fmt.Errorf("save view: %w", err)
+ }
+
+ return nil
+}
+
+// task returns the build task of the node key, with its input read from the log
+// or the tree.
+func (m *Memory) task(key NodeKey) (Task, error) {
+ task := Task{
+ Level: key.level,
+ Index: key.index,
+ Context: m.context(key.level, key.index),
+ }
+
+ if key.level == 0 {
+ message, err := m.log.Message(key.index)
+ if err != nil {
+ return Task{}, fmt.Errorf("read leaf message: %w", err)
+ }
+
+ task.Input = []string{leafSource(message)}
+ task.Images = imageFiles(message.Text)
+
+ return task, nil
+ }
+
+ left, right, err := m.children(key)
+ if err != nil {
+ return Task{}, fmt.Errorf("read task children: %w", err)
+ }
+
+ for _, child := range []Node{left, right} {
+ task.Input = append(task.Input, strings.ReplaceAll(child.Text, "\n", " "))
+ }
+
+ return task, nil
+}
+
+// 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)
+}
+
+// run builds one node with the summarizer. A failed build stays in the queue
+// until the next message.
+func (m *Memory) run(key NodeKey, task Task) {
+ defer m.workers.Done()
+
+ text, err := m.summarizer.Summarize(m.ctx, task)
+
projects / onechat / blobdiff