X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/522d7fd117285e4e4ce3ea5ec9b5a1722521b681..c781f79f3c8cb9a7577ee1574d196988da2719b3:/memory.go?ds=sidebyside diff --git a/memory.go b/memory.go index 7cbcce6..5ea9007 100644 --- a/memory.go +++ b/memory.go @@ -5,7 +5,7 @@ import ( "encoding/json" "errors" "fmt" - "log" + "log/slog" "os" "path/filepath" "strings" @@ -22,10 +22,68 @@ var ( errMemoryUnsettled = errors.New("a summary of the view failed, and it is tried again at the next message") ) +// Summarizer builds the text of one node with a model. +type Summarizer interface { + Summarize(ctx context.Context, task Task) (string, error) +} + +// Memory is the memory of one chat: the message log, the summary tree, the chat +// view, the compaction view, and the queue of the builds. The compaction view +// is the chat view merged further, at the limits of 16 to 32 KB. It takes the +// same new lines, and it merges again in one batch when it is too large or when +// the chat view merges. So the compactions share its prefix in the prompt +// cache. It saves the views to view.json after each change and loads them at +// 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 +// tasks of the running builds. It reads all other messages and nodes from their +// day files through the indexes. +type Memory struct { + mutex sync.Mutex + root string + log *Log + syncer *Syncer + tree *Tree + chat *View + cview *View + queue *BuildQueue + summarizer Summarizer + logger *slog.Logger + ctx context.Context + cancel context.CancelFunc + workers sync.WaitGroup + changed chan struct{} +} + +// 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. Images holds the leaf +// image filenames and is empty for merges. +type Task struct { + Level int + Index int + Context string + Input []string + Images []string +} + +// MemoryViewFile is the content of view.json. +type MemoryViewFile struct { + Chat ViewState `json:"chat"` + Compaction ViewState `json:"compaction"` +} + // 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. 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 *log.Logger) (*Memory, error) { +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 @@ -75,46 +133,6 @@ func openMemory(ctx context.Context, root string, clock func() time.Time, summar return m, nil } -// Summarizer builds the text of one node with a model. -type Summarizer interface { - Summarize(ctx context.Context, task Task) (string, error) -} - -// Memory is the memory of one chat: the message log, the summary tree, the chat -// view, the compaction view, and the queue of the builds. The compaction view -// is the chat view merged further, at the limits of 16 to 32 KB. It takes the -// same new lines, and it merges again in one batch when it is too large or when -// the chat view merges. So the compactions share its prefix in the prompt -// cache. It saves the views to view.json after each change and loads them at -// 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 -// tasks of the running builds. It reads all other messages and nodes from their -// day files through the indexes. -type Memory struct { - mutex sync.Mutex - root string - log *Log - syncer *Syncer - tree *Tree - chat *View - cview *View - queue *BuildQueue - summarizer Summarizer - logger *log.Logger - ctx context.Context - cancel context.CancelFunc - workers sync.WaitGroup - changed chan struct{} -} - // 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 { @@ -149,6 +167,11 @@ func (m *Memory) AppendDated(kind, text, date string) (Message, error) { m.mutex.Lock() defer m.mutex.Unlock() + 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 { return Message{}, fmt.Errorf("append dated log message: %w", err) @@ -172,6 +195,44 @@ func (m *Memory) AppendDated(kind, text, date string) (Message, error) { 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 @@ -247,6 +308,26 @@ 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 { @@ -422,9 +503,9 @@ func (m *Memory) dispatch() { } // 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. +// 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 { @@ -544,6 +625,7 @@ func (m *Memory) task(key NodeKey) (Task, error) { } task.Input = []string{leafSource(message)} + task.Images = imageFiles(message.Text) return task, nil } @@ -577,6 +659,7 @@ func (m *Memory) run(key NodeKey, task Task) { Text: text, Size: len(text), }) + m.syncer.Written() } @@ -598,7 +681,7 @@ func (m *Memory) run(key NodeKey, task Task) { // failed logs the failed build of the node key with its cause err. func (m *Memory) failed(key NodeKey, err error) { - m.logger.Printf("summary failed, retry at next message: node (%d, %d): %v", key.level, key.index, err) + 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. @@ -715,19 +798,3 @@ func (m *Memory) buildBuiltByModel(key NodeKey) (bool, error) { 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. -type Task struct { - Level int - Index int - Context string - Input []string -} - -// MemoryViewFile is the content of view.json. -type MemoryViewFile struct { - Chat ViewState `json:"chat"` - Compaction ViewState `json:"compaction"` -}