17 // errMemoryNoView tells that a chat has messages but no saved view.
18 errMemoryNoView = errors.New("the chat has messages but no saved view")
20 // errMemoryUnsettled tells that a line of the view stays pending,
21 // because its build failed.
22 errMemoryUnsettled = errors.New("a summary of the view failed, and it is tried again at the next message")
25 // Summarizer builds the text of one node with a model.
26 type Summarizer interface {
27 Summarize(ctx context.Context, task Task) (string, error)
30 // Memory is the memory of one chat: the message log, the summary tree, the chat
31 // view, the compaction view, and the queue of the builds. The compaction view
32 // is the chat view merged further, at the limits of 16 to 32 KB. It takes the
33 // same new lines, and it merges again in one batch when it is too large or when
34 // the chat view merges. So the compactions share its prefix in the prompt
35 // cache. It saves the views to view.json after each change and loads them at
36 // start. It never builds the views again from the log, because a new view
37 // differs from the live one and so loses the prompt cache.
39 // Each write goes to the operating system at once, and a syncer makes the
40 // writes durable by group commit in the background. Flush waits until the
41 // writes so far are durable.
43 // The summarizer builds the nodes in the queue in the background. The start
44 // reads the indexes once to fill the queue again. After that, only a new
45 // message or a new node adds work. Memory keeps the nodes of the views and the
46 // tasks of the running builds. It reads all other messages and nodes from their
47 // day files through the indexes.
60 cancel context.CancelFunc
61 workers sync.WaitGroup
65 // Task is the build of the node (Level, Index). Context is the compaction view
66 // for the build. Input holds the source "kind: text" of a leaf, or the two
67 // child lines of a merge with their newlines as spaces. Images holds the leaf
68 // image filenames and is empty for merges.
77 // MemoryViewFile is the content of view.json.
78 type MemoryViewFile struct {
79 Chat ViewState `json:"chat"`
80 Compaction ViewState `json:"compaction"`
83 // openMemory opens the chat in root and starts the builds that are ready. The
84 // clock gives the dates of new messages and the days of new nodes. logger gets
85 // one line for each failed build. The builds stop when ctx ends or at Close.
86 func openMemory(ctx context.Context, root string, clock func() time.Time, summarizer Summarizer, logger *slog.Logger) (*Memory, error) {
87 log, err := openLog(ctx, root, clock)
89 return nil, fmt.Errorf("open memory log: %w", err)
95 queue: newBuildQueue(),
96 summarizer: summarizer,
98 changed: make(chan struct{}),
101 m.ctx, m.cancel = context.WithCancel(ctx)
103 m.syncer, err = openSyncer(root)
105 return nil, errors.Join(err, m.Close())
108 m.tree, err = openTree(ctx, root, log.Len(), clock)
110 return nil, errors.Join(err, m.Close())
115 return nil, errors.Join(fmt.Errorf("load view: %w", err), m.Close())
118 err = m.restoreQueue()
120 return nil, errors.Join(fmt.Errorf("restore build queue: %w", err), m.Close())
125 return nil, errors.Join(err, m.Close())
136 // Close stops the builds, waits for them, makes the writes durable, closes the
137 // files of the chat, and releases its lock.
138 func (m *Memory) Close() error {
143 defer m.mutex.Unlock()
148 err = m.syncer.Close()
152 err = errors.Join(err, m.tree.Close())
155 return errors.Join(err, m.log.Close())
158 // load reads the saved views.
159 func (m *Memory) load() error {
160 data, err := os.ReadFile(m.viewPath())
161 if errors.Is(err, os.ErrNotExist) && m.log.Len() == 0 {
162 m.chat = newView(viewChatHigh, viewChatLow)
163 m.cview = newView(viewCompactionHigh, viewCompactionLow)
168 if errors.Is(err, os.ErrNotExist) {
169 return errMemoryNoView
173 return fmt.Errorf("read view: %w", err)
176 var file MemoryViewFile
178 err = json.Unmarshal(data, &file)
180 return fmt.Errorf("decode view: %w", err)
183 m.chat, err = restoreView(file.Chat, viewChatHigh, viewChatLow, m.tree)
185 return fmt.Errorf("chat view: %w", err)
188 m.cview, err = restoreView(file.Compaction, viewCompactionHigh, viewCompactionLow, m.tree)
190 return fmt.Errorf("compaction view: %w", err)
193 if m.cview.end != m.chat.end {
194 return fmt.Errorf("compaction view covers %d messages, but the chat view %d", m.cview.end, m.chat.end)
197 if m.chat.end > m.log.Len() {
198 return fmt.Errorf("view covers %d messages, but the log has %d", m.chat.end, m.log.Len())
204 // restoreQueue adds the work of the messages in the view to the queue. It
205 // builds each node that needs no model call and that a crash left out.
206 func (m *Memory) restoreQueue() error {
207 for id := range m.chat.end {
208 built, err := m.tree.Built(0, id)
210 return fmt.Errorf("check restored leaf: %w", err)
217 message, err := m.log.Message(id)
219 return fmt.Errorf("read restored message: %w", err)
222 leaf, free := freeLeaf(message)
231 return fmt.Errorf("build restored leaf: %w", err)
235 keys, err := m.tree.Keys()
237 return fmt.Errorf("read built nodes: %w", err)
240 for _, key := range keys {
241 if key.index%2 != 0 {
247 return fmt.Errorf("settle restored node: %w", err)
254 // catchUp adds the messages that the view does not cover yet. A crash can stop
255 // an append after the log write and before the view save.
256 func (m *Memory) catchUp() error {
257 if m.chat.end == m.log.Len() {
261 for m.chat.end < m.log.Len() {
262 message, err := m.log.Message(m.chat.end)
264 return fmt.Errorf("read pending message: %w", err)
267 err = m.addMessage(message)
269 return fmt.Errorf("add pending message: %w", err)
275 return fmt.Errorf("save caught-up memory: %w", err)
281 // dispatch starts the builds that are ready, up to the limit. A build whose
282 // input cannot be read fails and waits for the next message.
283 func (m *Memory) dispatch() {
284 for m.ctx.Err() == nil {
285 key, ready := m.queue.Next()
290 task, err := m.task(key)
292 m.failed(key, fmt.Errorf("read its input: %w", err))
304 func (m *Memory) viewPath() string {
305 return filepath.Join(m.root, "view.json")
308 // build adds node to the tree, keeps it in the views when it is a pending line,
309 // and settles its parent.
310 func (m *Memory) build(node Node) error {
311 err := m.tree.Add(node)
313 return fmt.Errorf("add built node: %w", err)
317 m.cview.Refresh(node)
319 err = m.settle(NodeKey{node.Level, node.Index})
321 return fmt.Errorf("settle built node: %w", err)
327 // settle looks at the parent of the built node key when both children are
328 // built. It builds each parent that needs no model call, up the tree, and adds
329 // the first other parent to the queue.
330 func (m *Memory) settle(key NodeKey) error {
332 parent := NodeKey{key.level + 1, key.index / 2}
333 ready, err := m.ready(parent)
334 if err != nil || !ready {
336 return fmt.Errorf("check parent readiness: %w", err)
342 left, right, err := m.children(parent)
344 return fmt.Errorf("read parent children: %w", err)
347 node, free := freeMerge(left, right)
349 m.queue.AddMerge(parent)
354 err = m.tree.Add(node)
356 return fmt.Errorf("add merged node: %w", err)
363 // addMessage builds the leaf of message when it needs no model call, or else
364 // adds the leaf to the queue. Then it appends the leaf to the views. A crash
365 // can stop an append after the leaf build and before the view save, so a leaf
366 // that is built already stays as it is, and restoreQueue settles its parent.
367 func (m *Memory) addMessage(message Message) error {
368 built, err := m.tree.Built(0, message.ID)
370 return fmt.Errorf("check message leaf: %w", err)
373 leaf, free := freeLeaf(message)
379 return fmt.Errorf("build message leaf: %w", err)
382 m.queue.AddLeaf(message.ID)
385 merged, err := m.chat.Append(m.tree)
387 return fmt.Errorf("append to chat view: %w", err)
390 _, err = m.cview.Append(m.tree)
391 if err == nil && merged {
392 err = m.cview.Coarsen(m.chat, m.tree)
396 return fmt.Errorf("append to compaction view: %w", err)
402 // save writes the views to view.json.
403 func (m *Memory) save() error {
404 data, err := json.Marshal(MemoryViewFile{
405 Chat: m.chat.State(),
406 Compaction: m.cview.State(),
409 return fmt.Errorf("encode view: %w", err)
412 err = writeFileAtomic(m.viewPath(), append(data, '\n'))
414 return fmt.Errorf("save view: %w", err)
420 // task returns the build task of the node key, with its input read from the log
422 func (m *Memory) task(key NodeKey) (Task, error) {
426 Context: m.context(key.level, key.index),
430 message, err := m.log.Message(key.index)
432 return Task{}, fmt.Errorf("read leaf message: %w", err)
435 task.Input = []string{leafSource(message)}
436 task.Images = imageFiles(message.Text)
441 left, right, err := m.children(key)
443 return Task{}, fmt.Errorf("read task children: %w", err)
446 for _, child := range []Node{left, right} {
447 task.Input = append(task.Input, strings.ReplaceAll(child.Text, "\n", " "))
453 // failed logs the failed build of the node key with its cause err.
454 func (m *Memory) failed(key NodeKey, err error) {
455 m.logger.Warn("summary failed", "level", key.level, "index", key.index, "err", err)
458 // run builds one node with the summarizer. A failed build stays in the queue
459 // until the next message.
460 func (m *Memory) run(key NodeKey, task Task) {
461 defer m.workers.Done()
463 text, err := m.summarizer.Summarize(m.ctx, task)
466 defer m.mutex.Unlock()
479 if err != nil && m.ctx.Err() == nil {
491 m.changed = make(chan struct{})
495 // ready tells if both children of the node key are built and the node is not.
496 func (m *Memory) ready(key NodeKey) (bool, error) {
497 for _, index := range []int{2 * key.index, 2*key.index + 1} {
498 built, err := m.tree.Built(key.level-1, index)
499 if err != nil || !built {
501 return false, fmt.Errorf("check child readiness: %w", err)
508 built, err := m.tree.Built(key.level, key.index)
510 return !built, fmt.Errorf("check parent readiness: %w", err)
516 // children reads the two children of the node key.
517 func (m *Memory) children(key NodeKey) (Node, Node, error) {
520 for n := range children {
521 child, built, err := m.tree.Node(key.level-1, 2*key.index+n)
523 return Node{}, Node{}, fmt.Errorf("read child node: %w", err)
527 return Node{}, Node{}, fmt.Errorf("node (%d, %d) is not built", key.level-1, 2*key.index+n)
533 return children[0], children[1], nil
536 func (m *Memory) context(level, index int) string {
539 end = nodeEnd(level, index)
542 return m.cview.Context(end)
545 // Append logs a new message, adds it to the tree and the views, and starts the
546 // builds that are ready, also the failed ones.
547 func (m *Memory) Append(kind, text string) (Message, error) {
548 return m.AppendDated(kind, text, "")
551 // AppendDated appends a message with its RFC 3339 date and builds its nodes. An
552 // empty date uses the clock, as Append does.
553 func (m *Memory) AppendDated(kind, text, date string) (Message, error) {
555 defer m.mutex.Unlock()
557 return m.appendDated(kind, text, date)
560 // appendDated appends under the memory lock and updates the views and builds.
561 func (m *Memory) appendDated(kind, text, date string) (Message, error) {
562 message, err := m.log.AppendDated(kind, text, date)
564 return Message{}, fmt.Errorf("append dated log message: %w", err)
567 defer m.syncer.Written()
569 err = m.addMessage(message)
571 return Message{}, fmt.Errorf("add dated message to memory: %w", err)
576 return Message{}, fmt.Errorf("save dated message view: %w", err)
585 // AppendOnce appends a queued message. prepare saves its log ID before the
586 // append, under the memory lock. A saved ID permits recovery after a stop
587 // between that save and the append, without a second log entry. prepare must
588 // not call Memory. The caller stops delivery after a failed append.
589 func (m *Memory) AppendOnce(kind, text string, id *int, prepare func(int) error) (Message, error) {
591 defer m.mutex.Unlock()
593 if id != nil && *id < m.log.Len() {
594 message, err := m.log.Message(*id)
596 return Message{}, fmt.Errorf("read queued message: %w", err)
599 if message.Kind != kind || message.Text != text {
600 return Message{}, errors.New("queued message does not match its log entry")
606 if id != nil && *id != m.log.Len() {
607 return Message{}, errors.New("queued message is beyond the log")
610 err := prepare(m.log.Len())
612 return Message{}, fmt.Errorf("prepare queued message: %w", err)
615 message, err := m.appendDated(kind, text, "")
617 return Message{}, fmt.Errorf("append queued message: %w", err)
623 // Wait returns when no build runs. The failed builds wait for the next message.
624 func (m *Memory) Wait(ctx context.Context) error {
628 running := m.queue.Running()
644 // Settle returns when the chat view has no pending line, so that every message
645 // so far is summarized. A pending line with no build that runs belongs to a
646 // failed build, which waits for the next message, so Settle returns
647 // errMemoryUnsettled in place of a wait that never ends.
648 func (m *Memory) Settle(ctx context.Context) error {
652 pending := m.chat.Pending()
653 running := m.queue.Running()
662 return errMemoryUnsettled
673 // Snapshot flushes all memory writes and runs snapshot while no append or
674 // summary write can change the log, indexes, tree, or views. snapshot must not
676 func (m *Memory) Snapshot(ctx context.Context, snapshot func() error) error {
678 defer m.mutex.Unlock()
682 return fmt.Errorf("flush memory snapshot: %w", err)
687 return fmt.Errorf("copy memory snapshot: %w", err)
693 // Flush returns when every write before the call is durable.
694 func (m *Memory) Flush(ctx context.Context) error {
695 return m.syncer.Flush(ctx)
698 // ViewLines returns the lines of the chat view, each with its newline, without
699 // the leaf lines of the messages in skip.
700 func (m *Memory) ViewLines(skip map[int]bool) []string {
702 defer m.mutex.Unlock()
704 return m.chat.LinesWithout(skip)
707 // Len returns the number of messages in the log.
708 func (m *Memory) Len() int {
710 defer m.mutex.Unlock()
715 // Message reads the message with the given ID.
716 func (m *Memory) Message(id int) (Message, error) {
718 defer m.mutex.Unlock()
720 return m.log.Message(id)
723 // Node reads the node (l, i) if it is built.
724 func (m *Memory) Node(level, index int) (Node, bool, error) {
726 defer m.mutex.Unlock()
728 return m.tree.Node(level, index)
731 // View returns the chat view as the model reads it.
732 func (m *Memory) View() string {
734 defer m.mutex.Unlock()
736 return m.chat.Render()
739 // Context returns the compaction view for the build of node (l, i). It ends
740 // before the message of a leaf, or after the last message of a merge, and it
741 // stops at the first line that is not built.
742 func (m *Memory) Context(level, index int) string {
744 defer m.mutex.Unlock()
746 return m.context(level, index)
749 // rebuildComplete checks the whole tree, not only the visible view. A compactor
750 // can fail without an HTTP error, for example with an empty line. Such a failed
751 // merge can be absent from the view and must prevent a swap.
752 func (m *Memory) rebuildComplete() error {
754 defer m.mutex.Unlock()
756 if len(m.queue.leaves) != 0 || len(m.queue.queued) != 0 {
757 return errors.New("rebuild failed: summary tree has unfinished nodes")
763 // buildModelLines returns the sizes of the nodes that the model wrote, in tree
764 // order. A node that fits its source with no model call is not one of them.
765 func (m *Memory) buildModelLines() ([]int, error) {
767 defer m.mutex.Unlock()
769 keys, err := m.tree.Keys()
771 return nil, fmt.Errorf("list built nodes: %w", err)
776 for _, key := range keys {
779 node, _, err = m.tree.Node(key.level, key.index)
781 return nil, fmt.Errorf("read built node: %w", err)
786 model, err = m.buildBuiltByModel(key)
788 return nil, fmt.Errorf("check model source: %w", err)
792 sizes = append(sizes, node.Size)
799 // buildBuiltByModel tells if the node key needed a model call: a leaf whose
800 // source is over nodeBytes, or a merge whose children do not fit in one node.
801 func (m *Memory) buildBuiltByModel(key NodeKey) (bool, error) {
803 message, err := m.log.Message(key.index)
805 return false, fmt.Errorf("read leaf source: %w", err)
808 _, free := freeLeaf(message)
813 left, right, err := m.children(key)
815 return false, fmt.Errorf("read merge sources: %w", err)
818 _, free := freeMerge(left, right)