Wait until the view has no pending summary

Memory.Settle returns when every line of the chat view is built, so a
subagent that starts gets a view with no pending summary, and the view
holds every message so far. A pending line with no running build belongs
to a failed build, which waits for the next message, so Settle returns
an error in place of a wait that never ends. It also ends with its
context, so an interrupted turn does not wait.

Issue: b1ac698

diff --git a/memory.go b/memory.go
index 9846c9817458f77cc0c92febb0a037aa4ff86dfa..0fd7d835f584bdea877f09bc063aebe969ddf1fa 100644 (file)
--- a/memory.go
+++ b/memory.go
@@ -12,8 +12,14 @@ import (
        "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
 // clock gives the dates of new messages and the days of new nodes. The builds
@@ -195,6 +201,34 @@ 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)

diff --git a/memory_test.go b/memory_test.go
index 001f852293d427e053e5f6e964f6e741147cf0be..4dee8b2951e18d2cb6d47ca60a0414e02054545d 100644 (file)
--- a/memory_test.go
+++ b/memory_test.go
@@ -1,6 +1,7 @@
 package main
 
 import (
+       "context"
        "fmt"
        "maps"
        "os"
@@ -144,6 +145,75 @@ func TestMemoryRunsAtMostEightBuilds(t *testing.T) {
        }
 }
 
+func TestMemorySettleWaitsForThePendingLines(t *testing.T) {
+       summarizer := newFakeSummarizer()
+       summarizer.gate = make(chan struct{})
+       memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, summarizer)
+       require.NoError(t, err)
+
+       defer func() {
+               require.NoError(t, memory.Close())
+       }()
+
+       _, err = memory.Append("echo", memoryLongText(0))
+       require.NoError(t, err)
+       require.Contains(t, memory.View(), viewPending)
+
+       settled := make(chan error, 1)
+       go func() {
+               settled <- memory.Settle(t.Context())
+       }()
+
+       select {
+       case err = <-settled:
+               require.Fail(t, "Settle returned before the build", "error: %v", err)
+       case <-time.After(50 * time.Millisecond):
+       }
+
+       close(summarizer.gate)
+       require.NoError(t, <-settled)
+       require.NotContains(t, memory.View(), viewPending)
+}
+
+func TestMemorySettleFailsWhileABuildFailed(t *testing.T) {
+       summarizer := newFakeSummarizer()
+       summarizer.SetFailing(true)
+       memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, summarizer)
+       require.NoError(t, err)
+
+       defer func() {
+               require.NoError(t, memory.Close())
+       }()
+
+       _, err = memory.Append("echo", memoryLongText(0))
+       require.NoError(t, err)
+       require.ErrorIs(t, memory.Settle(t.Context()), errMemoryUnsettled)
+
+       summarizer.SetFailing(false)
+       _, err = memory.Append("user", "retry")
+       require.NoError(t, err)
+       require.NoError(t, memory.Settle(t.Context()))
+}
+
+func TestMemorySettleEndsWithItsContext(t *testing.T) {
+       summarizer := newFakeSummarizer()
+       summarizer.gate = make(chan struct{})
+       memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, summarizer)
+       require.NoError(t, err)
+
+       defer func() {
+               require.NoError(t, memory.Close())
+       }()
+
+       _, err = memory.Append("echo", memoryLongText(0))
+       require.NoError(t, err)
+
+       ctx, cancel := context.WithTimeout(t.Context(), 50*time.Millisecond)
+       defer cancel()
+
+       require.ErrorIs(t, memory.Settle(ctx), context.DeadlineExceeded)
+}
+
 func TestMemoryStartsALeafAfterFewerThanEightUnbuilt(t *testing.T) {
        summarizer := newFakeSummarizer()
        summarizer.SetFailing(true)

diff --git a/view.go b/view.go
index 9fcd7d8bf91eff927daa49aa94a2b7d93168339f..233e08921f7a1ec2699244832177be784c3faae4 100644 (file)
--- a/view.go
+++ b/view.go
@@ -168,6 +168,18 @@ func (v *View) Refresh(node Node) {
        }
 }
 
+// Pending tells if a line of the view is not built yet.
+func (v *View) Pending() bool {
+       for _, line := range v.lines {
+               _, built := v.nodes[line]
+               if !built {
+                       return true
+               }
+       }
+
+       return false
+}
+
 // Render returns the view as the model reads it: one line "id+n|text" for each
 // node inside <chat> tags, with the newlines of the text as spaces.
 func (v *View) Render() string {